From 5bd5709fd4e7ba13047847a9dcf4684b5c7c5730 Mon Sep 17 00:00:00 2001 From: stefanbaxter Date: Mon, 31 Aug 2026 12:54:19 +0000 Subject: [PATCH] refactor(spec102): remove superseded enrichment runtime --- .../__tests__/enrichmentEntitlement.test.js | 128 ----- .../actions/src/rpc/reconcileDefaultModels.js | 40 +- .../src/rpc/reconcileTeamDefaultModels.js | 18 +- .../defaultModels/__tests__/config.test.js | 11 - .../actions/src/utils/defaultModels/config.js | 12 - .../defaultModels/enrichmentEntitlement.js | 181 ------ .../actions/src/utils/defaultModels/shared.js | 4 +- services/cubejs/index.js | 14 - .../src/__tests__/cube17Regression.test.js | 33 -- .../routes/__tests__/enrichmentGuards.test.js | 120 ---- .../routes/__tests__/reconcileTeam.test.js | 35 -- .../routes/__tests__/runSqlMetering.test.js | 102 ---- services/cubejs/src/routes/columnValues.js | 4 +- services/cubejs/src/routes/discoverNested.js | 4 +- services/cubejs/src/routes/dynamicMeta.js | 4 +- .../cubejs/src/routes/generateDataSchema.js | 4 +- services/cubejs/src/routes/getSchema.js | 4 +- services/cubejs/src/routes/loadExport.js | 64 +-- services/cubejs/src/routes/profileTable.js | 4 +- services/cubejs/src/routes/reconcileTeam.js | 64 +-- services/cubejs/src/routes/runSql.js | 44 +- services/cubejs/src/routes/smartGenerate.js | 4 +- .../utils/__tests__/billingMetrics.test.js | 72 --- .../src/utils/__tests__/billingOutbox.test.js | 213 ------- .../__tests__/enrichmentBilling.live.test.js | 537 ------------------ .../__tests__/enrichmentEntitlement.test.js | 278 --------- .../__tests__/enrichmentMetering.test.js | 317 ----------- .../__tests__/legacyEnrichmentGuard.test.js | 157 +++++ services/cubejs/src/utils/billingMetrics.js | 164 ------ services/cubejs/src/utils/billingOutbox.js | 194 ------- .../cubejs/src/utils/billingOutboxReplay.js | 19 - .../cubejs/src/utils/billingOutboxWorker.js | 174 ------ .../cubejs/src/utils/enrichmentMetering.js | 273 --------- .../cubejs/src/utils/enrichmentPricing.js | 99 ---- ...ntitlement.js => legacyEnrichmentGuard.js} | 127 +---- services/cubejs/src/utils/queryRewrite.js | 46 +- .../cubejs/src/utils/repositoryFactory.js | 25 +- .../enrichment/__tests__/templates.test.js | 135 ----- .../enrichment/compatibility-matrix.yaml | 10 - templates/enrichment/ctx_day_context.yml | 104 ---- templates/enrichment/ctx_weather_context.yml | 110 ---- templates/enrichment/publish.js | 122 ---- 42 files changed, 220 insertions(+), 3854 deletions(-) delete mode 100644 services/actions/src/rpc/__tests__/enrichmentEntitlement.test.js delete mode 100644 services/actions/src/utils/defaultModels/enrichmentEntitlement.js delete mode 100644 services/cubejs/src/routes/__tests__/enrichmentGuards.test.js delete mode 100644 services/cubejs/src/routes/__tests__/runSqlMetering.test.js delete mode 100644 services/cubejs/src/utils/__tests__/billingMetrics.test.js delete mode 100644 services/cubejs/src/utils/__tests__/billingOutbox.test.js delete mode 100644 services/cubejs/src/utils/__tests__/enrichmentBilling.live.test.js delete mode 100644 services/cubejs/src/utils/__tests__/enrichmentEntitlement.test.js delete mode 100644 services/cubejs/src/utils/__tests__/enrichmentMetering.test.js create mode 100644 services/cubejs/src/utils/__tests__/legacyEnrichmentGuard.test.js delete mode 100644 services/cubejs/src/utils/billingMetrics.js delete mode 100644 services/cubejs/src/utils/billingOutbox.js delete mode 100644 services/cubejs/src/utils/billingOutboxReplay.js delete mode 100644 services/cubejs/src/utils/billingOutboxWorker.js delete mode 100644 services/cubejs/src/utils/enrichmentMetering.js delete mode 100644 services/cubejs/src/utils/enrichmentPricing.js rename services/cubejs/src/utils/{enrichmentEntitlement.js => legacyEnrichmentGuard.js} (59%) delete mode 100644 templates/enrichment/__tests__/templates.test.js delete mode 100644 templates/enrichment/compatibility-matrix.yaml delete mode 100644 templates/enrichment/ctx_day_context.yml delete mode 100644 templates/enrichment/ctx_weather_context.yml delete mode 100644 templates/enrichment/publish.js diff --git a/services/actions/src/rpc/__tests__/enrichmentEntitlement.test.js b/services/actions/src/rpc/__tests__/enrichmentEntitlement.test.js deleted file mode 100644 index 9a03d1af..00000000 --- a/services/actions/src/rpc/__tests__/enrichmentEntitlement.test.js +++ /dev/null @@ -1,128 +0,0 @@ -import { createHmac } from "node:crypto"; -import { describe, it } from "node:test"; -import assert from "node:assert/strict"; - -import { - mergeEnrichmentEntitlement, - reconcileEnrichmentEntitlement, - resolveEnrichmentEntitlement, -} from "../../utils/defaultModels/enrichmentEntitlement.js"; - -const KEY = "test-only-entitlement-key-with-at-least-32-bytes"; -const BILLING_CONNECTION_ID = "11111111-1111-4111-8111-111111111111"; -const NOW = new Date("2026-08-30T12:00:00.000Z"); - -const signedLease = (overrides = {}) => { - const payload = { - schema_version: 1, - account_partition: "customer.is", - enabled: true, - entitlement_revision: "7", - issued_at: "2026-08-30T11:55:00.000Z", - valid_until: "2026-08-30T12:25:00.000Z", - products: ["ctx:day-archetype", "ctx:weather-archetype"], - billing_connection_id: BILLING_CONNECTION_ID, - ...overrides, - }; - return { - payload, - signature_version: "hmac-sha256-v1", - signature: createHmac("sha256", KEY) - .update(JSON.stringify(payload)) - .digest("base64url"), - }; -}; - -const config = { - enrichmentEntitlementUrl: - "http://cxs2.cxs2.svc.cluster.local/api/internal/semantic-layer/enrichment-entitlements", - enrichmentServiceKey: "service-key", - enrichmentSigningKey: KEY, - enrichmentTimeoutMs: 1_000, -}; - -describe("enrichment entitlement reconciliation", () => { - it("accepts a matching signed, unexpired lease", async () => { - const result = await resolveEnrichmentEntitlement("customer.is", config, { - now: () => NOW, - fetchImpl: async () => ({ ok: true, json: async () => signedLease() }), - }); - assert.equal(result.valid, true); - assert.equal(result.enabled, true); - assert.equal(result.lease.entitlement_revision, "7"); - assert.equal(result.lease.billing_connection_id, BILLING_CONNECTION_ID); - }); - - it("fails closed on signature tampering, expiry, and account mismatch", async () => { - for (const lease of [ - { ...signedLease(), signature: "tampered" }, - signedLease({ valid_until: "2026-08-30T11:59:59.000Z" }), - signedLease({ account_partition: "other.is" }), - signedLease({ billing_connection_id: null }), - ]) { - const result = await resolveEnrichmentEntitlement("customer.is", config, { - now: () => NOW, - fetchImpl: async () => ({ ok: true, json: async () => lease }), - }); - assert.equal(result.valid, false); - assert.equal(result.enabled, false); - } - }); - - it("fails closed when cxs2 is unavailable", async () => { - const result = await resolveEnrichmentEntitlement("customer.is", config, { - now: () => NOW, - fetchImpl: async () => { - throw new Error("connection refused"); - }, - }); - assert.equal(result.valid, false); - assert.equal(result.enabled, false); - assert.equal(result.reason, "authority_unavailable"); - }); - - it("merges the lease without clobbering sibling team settings", () => { - const settings = { - partition: "customer.is", - default_models: { opt_out: ["team-model"] }, - premium: { another_product: { enabled: true } }, - }; - const merged = mergeEnrichmentEntitlement(settings, { - valid: true, - enabled: true, - lease: { - enabled: true, - entitlement_revision: "7", - issued_at: "2026-08-30T11:55:00.000Z", - valid_until: "2026-08-30T12:25:00.000Z", - signature_version: "hmac-sha256-v1", - signature: "opaque", - products: ["ctx:day-archetype", "ctx:weather-archetype"], - billing_connection_id: BILLING_CONNECTION_ID, - }, - }); - assert.deepEqual(merged.default_models, settings.default_models); - assert.deepEqual(merged.premium.another_product, { enabled: true }); - assert.equal(merged.premium.enrichment.enabled, true); - }); - - it("persists a disabled lease on outage so a prior grant is revoked", async () => { - const persisted = []; - const team = { - id: "team-1", - settings: { - partition: "customer.is", - premium: { enrichment: { enabled: true, entitlement_revision: "6" } }, - }, - }; - const result = await reconcileEnrichmentEntitlement(team, config, { - resolve: async () => ({ valid: false, enabled: false, reason: "authority_unavailable" }), - persist: async (teamId, settings) => persisted.push({ teamId, settings }), - }); - assert.equal(result.team.settings.premium.enrichment.enabled, false); - assert.equal(result.enrichmentEnabled, false); - assert.equal(result.enrichmentRevokeRequired, true); - assert.equal(persisted.length, 1); - assert.equal(persisted[0].settings.partition, "customer.is"); - }); -}); diff --git a/services/actions/src/rpc/reconcileDefaultModels.js b/services/actions/src/rpc/reconcileDefaultModels.js index d0fa9a11..d81223ef 100644 --- a/services/actions/src/rpc/reconcileDefaultModels.js +++ b/services/actions/src/rpc/reconcileDefaultModels.js @@ -21,11 +21,6 @@ import { captureDriftSnapshot, diffDriftSnapshots, } from "../utils/defaultModels/drift.js"; -import { - ENRICHMENT_TEMPLATE_NAMES, - reconcileEnrichmentEntitlement, - templatesForEntitlement, -} from "../utils/defaultModels/enrichmentEntitlement.js"; const TEAM_CONCURRENCY = 4; @@ -49,7 +44,6 @@ export default async (session, input, headers, deps = {}) => { listTeams = listAllTeams, reconcileOneTeam = reconcileOneTeamImpl, captureDrift = captureDriftSnapshot, - reconcileEntitlement = reconcileEnrichmentEntitlement, isAdmin, } = deps; @@ -159,25 +153,11 @@ export default async (session, input, headers, deps = {}) => { for (;;) { const team = queue.shift(); if (!team) return; - let entitlement; - try { - entitlement = await reconcileEntitlement(team, config, { dryRun }); - } catch (err) { - await record({ - team_id: team.id, - result: "failed", - reason: `enrichment_entitlement: ${err?.message || String(err)}`, - }); - continue; - } - const effectiveTeam = entitlement.team; const unchanged = changedPartitions !== null && - !changedPartitions.has(effectiveTeam.settings?.partition); + !changedPartitions.has(team.settings?.partition); // unchanged team on a schedule tick: skip before any per-team probe - // unless an entitlement revocation still needs to remove managed - // artifacts. A revoke-only pass sends no ordinary templates. - if (unchanged && !entitlement.enrichmentRevokeRequired) { + if (unchanged) { await record({ team_id: team.id, result: "skipped_no_change", @@ -187,20 +167,10 @@ export default async (session, input, headers, deps = {}) => { } try { const teamOutcomes = await reconcileOneTeam( - effectiveTeam, - unchanged - ? [] - : templatesForEntitlement( - templates, - entitlement.enrichmentEnabled - ), + team, + templates, config, - { - dryRun, - revokeTemplates: entitlement.enrichmentRevokeRequired - ? ENRICHMENT_TEMPLATE_NAMES - : [], - } + { dryRun } ); for (const outcome of teamOutcomes) { const row = { team_id: team.id, ...outcome }; diff --git a/services/actions/src/rpc/reconcileTeamDefaultModels.js b/services/actions/src/rpc/reconcileTeamDefaultModels.js index 48387f49..45c51ab6 100644 --- a/services/actions/src/rpc/reconcileTeamDefaultModels.js +++ b/services/actions/src/rpc/reconcileTeamDefaultModels.js @@ -12,11 +12,6 @@ import { getTeam as getTeamImpl, reconcileOneTeam as reconcileOneTeamImpl, } from "../utils/defaultModels/shared.js"; -import { - ENRICHMENT_TEMPLATE_NAMES, - reconcileEnrichmentEntitlement, - templatesForEntitlement, -} from "../utils/defaultModels/enrichmentEntitlement.js"; export default async (session, input, headers, deps = {}) => { const { @@ -26,7 +21,6 @@ export default async (session, input, headers, deps = {}) => { fetchTemplates = fetchPublishedTemplates, getTeam = getTeamImpl, reconcileOneTeam = reconcileOneTeamImpl, - reconcileEntitlement = reconcileEnrichmentEntitlement, isAdmin, } = deps; @@ -74,17 +68,7 @@ export default async (session, input, headers, deps = {}) => { let outcomes; try { - const entitlement = await reconcileEntitlement(team, config); - outcomes = await reconcileOneTeam( - entitlement.team, - templatesForEntitlement(templates, entitlement.enrichmentEnabled), - config, - { - revokeTemplates: entitlement.enrichmentRevokeRequired - ? ENRICHMENT_TEMPLATE_NAMES - : [], - } - ); + outcomes = await reconcileOneTeam(team, templates, config); } catch (err) { outcomes = [ { diff --git a/services/actions/src/utils/defaultModels/__tests__/config.test.js b/services/actions/src/utils/defaultModels/__tests__/config.test.js index 6f1d6dc1..fe905072 100644 --- a/services/actions/src/utils/defaultModels/__tests__/config.test.js +++ b/services/actions/src/utils/defaultModels/__tests__/config.test.js @@ -50,10 +50,6 @@ test("applies defaults for the optional keys", () => { assert.deepEqual(config.canaryTeamIds, []); // empty drift probes = treat all teams as changed assert.deepEqual(config.driftProbes, []); - assert.equal(config.enrichmentEntitlementUrl, null); - assert.equal(config.enrichmentServiceKey, null); - assert.equal(config.enrichmentSigningKey, null); - assert.equal(config.enrichmentTimeoutMs, 5000); }); test("parses optional keys into typed values", () => { @@ -64,12 +60,6 @@ test("parses optional keys into typed values", () => { DEFAULT_MODELS_COHORTS: "6", DEFAULT_MODELS_DRIFT_PROBES: '[{"table":"cst.semantic_events","timeColumn":"timestamp"}]', - CXS2_ENRICHMENT_ENTITLEMENT_URL: - "http://cxs2.cxs2.svc/api/internal/semantic-layer/enrichment-entitlements", - INTERNAL_SERVICE_API_KEY: "service-key", - ENRICHMENT_ENTITLEMENT_SIGNING_KEY: - "test-only-entitlement-key-with-at-least-32-bytes", - ENRICHMENT_ENTITLEMENT_TIMEOUT_MS: "1234", }); assert.deepEqual(config.canaryTeamIds, [UUID_A, UUID_B]); @@ -78,7 +68,6 @@ test("parses optional keys into typed values", () => { assert.deepEqual(config.driftProbes, [ { table: "cst.semantic_events", timeColumn: "timestamp" }, ]); - assert.equal(config.enrichmentTimeoutMs, 1234); }); test("rejects malformed DEFAULT_MODELS_DRIFT_PROBES", () => { diff --git a/services/actions/src/utils/defaultModels/config.js b/services/actions/src/utils/defaultModels/config.js index 0493a00d..88e4a48b 100644 --- a/services/actions/src/utils/defaultModels/config.js +++ b/services/actions/src/utils/defaultModels/config.js @@ -83,13 +83,6 @@ export const loadDefaultModelsConfig = (env = process.env) => { .map((id) => id.trim()) .filter(Boolean); - const enrichmentTimeoutMs = env.ENRICHMENT_ENTITLEMENT_TIMEOUT_MS - ? Number(env.ENRICHMENT_ENTITLEMENT_TIMEOUT_MS) - : 5_000; - if (!Number.isInteger(enrichmentTimeoutMs) || enrichmentTimeoutMs <= 0) { - fail("ENRICHMENT_ENTITLEMENT_TIMEOUT_MS must be a positive integer"); - } - return { templateDatasourceId, systemUserId, @@ -99,11 +92,6 @@ export const loadDefaultModelsConfig = (env = process.env) => { cohorts, driftProbes: parseDriftProbes(env.DEFAULT_MODELS_DRIFT_PROBES), cronSecret: env.ACTIONS_CRON_SECRET || null, - enrichmentEntitlementUrl: - env.CXS2_ENRICHMENT_ENTITLEMENT_URL?.trim() || null, - enrichmentServiceKey: env.INTERNAL_SERVICE_API_KEY || null, - enrichmentSigningKey: env.ENRICHMENT_ENTITLEMENT_SIGNING_KEY || null, - enrichmentTimeoutMs, }; }; diff --git a/services/actions/src/utils/defaultModels/enrichmentEntitlement.js b/services/actions/src/utils/defaultModels/enrichmentEntitlement.js deleted file mode 100644 index 6d3763dc..00000000 --- a/services/actions/src/utils/defaultModels/enrichmentEntitlement.js +++ /dev/null @@ -1,181 +0,0 @@ -import { createHmac, timingSafeEqual } from "node:crypto"; - -import { fetchGraphQL } from "../graphql.js"; - -const PRODUCTS = ["ctx:day-archetype", "ctx:weather-archetype"]; - -const UPDATE_TEAM_SETTINGS = ` - mutation ($teamId: uuid!, $settings: jsonb!) { - update_teams_by_pk(pk_columns: { id: $teamId }, _set: { settings: $settings }) { - id - } - } -`; - -const canonicalPayload = (payload) => - JSON.stringify({ - schema_version: payload?.schema_version, - account_partition: payload?.account_partition, - enabled: payload?.enabled, - entitlement_revision: payload?.entitlement_revision, - issued_at: payload?.issued_at, - valid_until: payload?.valid_until, - products: payload?.products, - billing_connection_id: payload?.billing_connection_id, - }); - -const signaturesMatch = (left, right) => { - const a = Buffer.from(String(left || "")); - const b = Buffer.from(String(right || "")); - return a.length === b.length && timingSafeEqual(a, b); -}; - -const disabled = (reason) => ({ valid: false, enabled: false, reason }); - -export const resolveEnrichmentEntitlement = async ( - partition, - config, - deps = {} -) => { - const fetchImpl = deps.fetchImpl || globalThis.fetch; - const now = (deps.now || (() => new Date()))(); - if ( - !partition || - !config?.enrichmentEntitlementUrl || - !config?.enrichmentServiceKey || - !config?.enrichmentSigningKey - ) { - return disabled("authority_unconfigured"); - } - - try { - const url = new URL(config.enrichmentEntitlementUrl); - url.searchParams.set("partition", partition); - const response = await fetchImpl(url, { - headers: { Authorization: `Bearer ${config.enrichmentServiceKey}` }, - signal: AbortSignal.timeout(config.enrichmentTimeoutMs || 5_000), - }); - if (!response.ok) return disabled("authority_unavailable"); - const envelope = await response.json(); - const payload = envelope?.payload; - if ( - envelope?.signature_version !== "hmac-sha256-v1" || - payload?.schema_version !== 1 || - payload?.account_partition !== partition || - typeof payload?.enabled !== "boolean" || - typeof payload?.entitlement_revision !== "string" || - !Array.isArray(payload?.products) || - payload.products.join(",") !== PRODUCTS.join(",") || - (payload.enabled - ? !/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test( - payload.billing_connection_id || "", - ) - : payload.billing_connection_id !== null) - ) { - return disabled("lease_malformed"); - } - const issuedAt = Date.parse(payload.issued_at); - const validUntil = Date.parse(payload.valid_until); - if ( - !Number.isFinite(issuedAt) || - !Number.isFinite(validUntil) || - issuedAt > now.getTime() || - validUntil <= now.getTime() - ) { - return disabled("lease_expired"); - } - const expected = createHmac("sha256", config.enrichmentSigningKey) - .update(canonicalPayload(payload)) - .digest("base64url"); - if (!signaturesMatch(expected, envelope.signature)) { - return disabled("lease_signature_invalid"); - } - - return { - valid: true, - enabled: payload.enabled, - lease: { - enabled: payload.enabled, - entitlement_revision: payload.entitlement_revision, - issued_at: payload.issued_at, - valid_until: payload.valid_until, - signature_version: envelope.signature_version, - signature: envelope.signature, - products: [...payload.products], - billing_connection_id: payload.billing_connection_id, - }, - }; - } catch { - return disabled("authority_unavailable"); - } -}; - -export const mergeEnrichmentEntitlement = (settings = {}, resolution) => { - const current = settings?.premium?.enrichment || {}; - const enrichment = resolution.valid - ? resolution.lease - : { - enabled: false, - entitlement_revision: current.entitlement_revision || null, - issued_at: null, - valid_until: null, - signature_version: null, - signature: null, - products: PRODUCTS, - billing_connection_id: null, - reason: resolution.reason || "invalid_lease", - }; - return { - ...settings, - premium: { - ...(settings.premium || {}), - enrichment, - }, - }; -}; - -export const persistEnrichmentEntitlement = async (teamId, settings) => { - await fetchGraphQL(UPDATE_TEAM_SETTINGS, { teamId, settings }); -}; - -export const reconcileEnrichmentEntitlement = async ( - team, - config, - deps = {} -) => { - const resolve = deps.resolve || resolveEnrichmentEntitlement; - const persist = deps.persist || persistEnrichmentEntitlement; - const resolution = await resolve(team.settings?.partition, config, deps); - const hadLease = Boolean(team.settings?.premium?.enrichment); - const settings = - !hadLease && resolution.reason === "authority_unconfigured" - ? team.settings || {} - : mergeEnrichmentEntitlement(team.settings || {}, resolution); - if (JSON.stringify(settings) !== JSON.stringify(team.settings || {})) { - if (!deps.dryRun) await persist(team.id, settings); - } - return { - team: { ...team, settings }, - enrichmentEnabled: resolution.valid && resolution.enabled === true, - enrichmentRevokeRequired: - !(resolution.valid && resolution.enabled === true) && - (hadLease || - Boolean( - config?.enrichmentEntitlementUrl && - config?.enrichmentServiceKey && - config?.enrichmentSigningKey - )), - resolution, - }; -}; - -export const isEnrichmentTemplate = (template) => - template?.name === "ctx_day_context" || template?.name === "ctx_weather_context"; - -export const templatesForEntitlement = (templates, enabled) => - enabled ? templates : templates.filter((template) => !isEnrichmentTemplate(template)); - -export const ENRICHMENT_TEMPLATE_NAMES = [ - "ctx_day_context", - "ctx_weather_context", -]; diff --git a/services/actions/src/utils/defaultModels/shared.js b/services/actions/src/utils/defaultModels/shared.js index 318fd861..9506e9c1 100644 --- a/services/actions/src/utils/defaultModels/shared.js +++ b/services/actions/src/utils/defaultModels/shared.js @@ -179,7 +179,7 @@ export const resolveTeamTarget = async (team, config) => { * (contracts/cubejs-internal.md). */ export const callWorker = async ( - { team, datasourceId, branchId, templates, optOut, dryRun, revokeTemplates }, + { team, datasourceId, branchId, templates, optOut, dryRun }, config ) => { const { default: generateUserAccessToken } = await import("../jwt.js"); @@ -202,7 +202,6 @@ export const callWorker = async ( templates, optOut, dryRun, - revokeTemplates, }); return res?.outcomes || []; }; @@ -253,7 +252,6 @@ export const reconcileOneTeam = async (team, templates, config, options = {}) => templates, optOut, dryRun: options.dryRun || false, - revokeTemplates: options.revokeTemplates || [], }, config ); diff --git a/services/cubejs/index.js b/services/cubejs/index.js index 43194500..5a09747d 100644 --- a/services/cubejs/index.js +++ b/services/cubejs/index.js @@ -18,10 +18,6 @@ import createQueryPreprocessor from "./src/utils/queryPreprocessor.js"; import queryRewrite from "./src/utils/queryRewrite.js"; import repositoryFactory from "./src/utils/repositoryFactory.js"; import scheduledRefreshContexts from "./src/utils/scheduledRefreshContexts.js"; -import redisClient from "./src/utils/redis.js"; -import { BillingOutboxWorker } from "./src/utils/billingOutboxWorker.js"; -import { createBillingMetricsHandler } from "./src/utils/billingMetrics.js"; -import { installEnrichmentGatewayMetering } from "./src/utils/enrichmentMetering.js"; // Installed before anything else so failures during startup are covered too. // A pre-aggregation whose build query fails rejects past Cube's orchestrator; @@ -42,9 +38,6 @@ const { const port = parseInt(process.env.PORT, 10) || 4000; const app = express(); -const billingOutboxWorker = redisClient - ? new BillingOutboxWorker(redisClient) - : null; // Hasura auth proxy — mounted BEFORE body parsers for raw body passthrough (R8) const hasuraProxy = createHasuraProxy(); @@ -52,10 +45,6 @@ app.use(hasuraProxy); app.use(express.json({ limit: "50mb", extended: true })); app.use(express.urlencoded({ limit: "50mb", extended: true })); -app.get( - "/internal/metrics/billing", - createBillingMetricsHandler({ redis: redisClient, worker: billingOutboxWorker }), -); const contextToOrchestratorId = ({ securityContext }) => `CUBEJS_APP_${securityContext?.userScope?.dataSource?.dataSourceVersion}_${securityContext?.userScope?.dataSource?.schemaVersion}}`; @@ -111,7 +100,6 @@ const cubejs = new ServerCore(options); // Custom raw-SQL routes require driver instances; Cube's server-level factory // must remain config-only so 1.7 can derive the dialect per tenant context. cubejs.tenantDriverFactory = driverFactory; -installEnrichmentGatewayMetering(cubejs, redisClient); const file = fs.readFileSync("./src/swagger.yaml", "utf8"); const swaggerDocument = YAML.parse(file); @@ -146,13 +134,11 @@ app.use((err, req, res, next) => { }); const server = app.listen(port); -billingOutboxWorker?.start(); let shuttingDown = false; const shutdown = async () => { if (shuttingDown) return; shuttingDown = true; - await billingOutboxWorker?.stop(); server.close(); }; process.once("SIGTERM", shutdown); diff --git a/services/cubejs/src/__tests__/cube17Regression.test.js b/services/cubejs/src/__tests__/cube17Regression.test.js index b28a94f4..56990da6 100644 --- a/services/cubejs/src/__tests__/cube17Regression.test.js +++ b/services/cubejs/src/__tests__/cube17Regression.test.js @@ -5,11 +5,6 @@ import { describe, it } from "node:test"; import { prepareCompiler } from "@cubejs-backend/schema-compiler"; import { escapeCSVField } from "../utils/csvSerializer.js"; -import { - queryUsesEnrichment, - sqlEnrichmentBillingItems, -} from "../utils/enrichmentEntitlement.js"; -import { deterministicBillingMessageId } from "../utils/enrichmentMetering.js"; import { validateFormat } from "../utils/formatValidator.js"; import { patchCompilerSource } from "../../scripts/patchCubeYamlCompiler.mjs"; @@ -24,20 +19,13 @@ const CORPUS = { "../utils/smart-generation/__tests__/cubeBuilder.test.js", "../routes/__tests__/reconcileTeam.test.js", ], - guards: ["../routes/__tests__/enrichmentGuards.test.js"], formats: [ "../utils/formatValidator.js", "../utils/csvSerializer.js", "../utils/arrowSerializer.js", ], - sql_api: ["../routes/__tests__/runSqlMetering.test.js"], auth: ["../utils/__tests__/workosAuth.test.js"], pre_aggregations: ["../routes/__tests__/validateInBranch.corpus.test.js"], - outbox: ["../utils/__tests__/billingOutbox.test.js"], - metering: [ - "../utils/__tests__/enrichmentMetering.test.js", - "../utils/__tests__/connectionCalledBilling.test.js", - ], }; describe("Cube 1.6.68 to 1.7.30 comparative corpus", () => { @@ -65,27 +53,6 @@ describe("Cube 1.6.68 to 1.7.30 comparative corpus", () => { assert.throws(() => validateFormat("parquet"), /Unsupported format/); }); - it("keeps enrichment detection and charge identity stable", () => { - assert.equal( - queryUsesEnrichment({ filters: [{ member: "CtxWeatherContext.marker" }] }), - true, - ); - assert.deepEqual( - sqlEnrichmentBillingItems( - "SELECT * FROM enrich.day_context_v JOIN enrich.weather_context_v USING (event_date)", - ), - ["ctx:day-archetype", "ctx:weather-archetype"], - ); - assert.equal( - deterministicBillingMessageId("logical-1", "ctx:day-archetype"), - deterministicBillingMessageId("logical-1", "ctx:day-archetype"), - ); - assert.notEqual( - deterministicBillingMessageId("logical-1", "ctx:day-archetype"), - deterministicBillingMessageId("logical-1", "ctx:weather-archetype"), - ); - }); - it("preserves JSON-valued metadata as a literal string", async () => { const value = JSON.stringify({ time_zone: "Atlantic/Reykjavik", diff --git a/services/cubejs/src/routes/__tests__/enrichmentGuards.test.js b/services/cubejs/src/routes/__tests__/enrichmentGuards.test.js deleted file mode 100644 index 4e527b88..00000000 --- a/services/cubejs/src/routes/__tests__/enrichmentGuards.test.js +++ /dev/null @@ -1,120 +0,0 @@ -import assert from "node:assert/strict"; -import { createHmac } from "node:crypto"; -import { describe, it } from "node:test"; - -import queryRewrite from "../../utils/queryRewrite.js"; -import { authorizeNativeEnrichmentSql } from "../loadExport.js"; -import { authorizeRunSqlQuery } from "../runSql.js"; - -const KEY = "test-enrichment-signing-key-at-least-32-bytes"; -const NOW = new Date("2026-08-30T10:00:00.000Z"); -const PRODUCTS = ["ctx:day-archetype", "ctx:weather-archetype"]; -const BILLING_CONNECTION_ID = "11111111-1111-4111-8111-111111111111"; - -function securityContext() { - const payload = { - schema_version: 1, - account_partition: "tenant-is", - enabled: true, - entitlement_revision: "12", - issued_at: "2026-08-30T09:00:00.000Z", - valid_until: "2026-08-30T11:00:00.000Z", - products: PRODUCTS, - billing_connection_id: BILLING_CONNECTION_ID, - }; - const signature = createHmac("sha256", KEY) - .update(JSON.stringify(payload)) - .digest("base64url"); - return { - userScope: { - teamProperties: { - partition: payload.account_partition, - premium: { - enrichment: { - enabled: payload.enabled, - entitlement_revision: payload.entitlement_revision, - issued_at: payload.issued_at, - valid_until: payload.valid_until, - products: payload.products, - billing_connection_id: payload.billing_connection_id, - signature_version: "hmac-sha256-v1", - signature, - }, - }, - }, - }, - }; -} - -const OPTIONS = { signingKey: KEY, now: NOW }; - -describe("enrichment side-door authorization", () => { - it("denies a nested resolved member at the compiler boundary before rule loading", async () => { - await assert.rejects( - queryRewrite( - { - measures: ["Orders.count"], - filters: [{ or: [{ member: "CtxWeatherContext.temperatureAvg" }] }], - }, - { securityContext: {} }, - ), - (error) => - error.status === 403 && - error.code === "enrichment_not_available" && - !/weather|temperature/i.test(error.message), - ); - }); - - it("applies the parser guard to run-sql independently of SQL provenance", () => { - assert.doesNotThrow(() => - authorizeRunSqlQuery( - 'SELECT * FROM "enrich"."weather_context_v"', - securityContext(), - OPTIONS, - ), - ); - assert.throws( - () => - authorizeRunSqlQuery( - "SELECT * FROM /* misleading enrich.day_context_v */ enrich.release_pointer", - securityContext(), - OPTIONS, - ), - (error) => error.status === 403 && !/release_pointer/.test(error.message), - ); - }); - - it("applies the same guard to native CSV and Arrow export SQL", () => { - const nativeSql = ` - -- FROM enrich.day_context - SELECT * FROM \`enrich\`.\`day_context_v\` - JOIN \"enrich\".\"weather_context_v\" USING (date) - `; - for (const format of ["csv", "arrow"]) { - assert.doesNotThrow(() => - authorizeNativeEnrichmentSql(nativeSql, securityContext(), OPTIONS), - ); - assert.throws( - () => authorizeNativeEnrichmentSql(nativeSql, {}, OPTIONS), - (error) => - error.status === 403 && - error.code === "enrichment_not_available" && - !error.message.includes(format), - ); - } - }); - - it("cannot be bypassed with quoted, qualified, or commented physical names", () => { - const blocked = [ - 'SELECT * FROM "enrich"."day_context"', - "SELECT * FROM `enrich`.`weather_context`", - "SELECT * FROM /* enrich.day_context_v */ enrich.release_manifest", - ]; - for (const sql of blocked) { - assert.throws( - () => authorizeRunSqlQuery(sql, securityContext(), OPTIONS), - (error) => error.status === 403, - ); - } - }); -}); diff --git a/services/cubejs/src/routes/__tests__/reconcileTeam.test.js b/services/cubejs/src/routes/__tests__/reconcileTeam.test.js index cd369c66..c00c18ee 100644 --- a/services/cubejs/src/routes/__tests__/reconcileTeam.test.js +++ b/services/cubejs/src/routes/__tests__/reconcileTeam.test.js @@ -108,41 +108,6 @@ describe('computeVersionChecksum', () => { }); describe('reconcileTeamCore — worker pipeline', () => { - it('revokes only managed enrichment cubes and preserves team-authored siblings', async () => { - const mixed = `cubes: - - name: CtxDayContext - sql_table: enrich.day_context_v - meta: - default_model: true - managed_by: ctx-enrichment - template: ctx_day_context - - name: TeamNotes - sql_table: team.notes - dimensions: - - name: note - sql: note - type: string -`; - const { deps, calls } = makeDeps({ - loadCurrentSchemas: async () => [ - { id: 'ds-1', name: 'ctx_day_context.yml', code: mixed }, - ], - }); - const result = await reconcileTeamCore( - baseParams({ - templates: [], - revokeTemplates: ['ctx_day_context', 'ctx_weather_context'], - }), - deps - ); - - assert.equal(result.outcomes[0].template, 'ctx_day_context'); - assert.equal(result.outcomes[0].result, 'removed'); - assert.equal(calls.publish.length, 1); - assert.match(calls.publish[0].files[0].code, /TeamNotes/); - assert.doesNotMatch(calls.publish[0].files[0].code, /CtxDayContext/); - }); - it('collision: team-authored file without provenance meta is skipped, nothing written', async () => { const { deps, calls } = makeDeps({ loadCurrentSchemas: async () => [ diff --git a/services/cubejs/src/routes/__tests__/runSqlMetering.test.js b/services/cubejs/src/routes/__tests__/runSqlMetering.test.js deleted file mode 100644 index 9d667e78..00000000 --- a/services/cubejs/src/routes/__tests__/runSqlMetering.test.js +++ /dev/null @@ -1,102 +0,0 @@ -import assert from "node:assert/strict"; -import { EventEmitter } from "node:events"; -import { beforeEach, describe, it, mock } from "node:test"; - -const commitMock = mock.fn(); -mock.module("../../utils/enrichmentMetering.js", { - namedExports: { commitSqlEnrichmentBilling: commitMock }, -}); -mock.module("../../utils/queryRewrite.js", { - namedExports: { loadRules: async () => [] }, -}); -mock.module("../../utils/enrichmentEntitlement.js", { - namedExports: { assertSqlEnrichmentAuthorized: () => {} }, -}); -mock.module("../../utils/eventEmitter.js", { - namedExports: { emitQueryEvent: () => {} }, -}); - -const { default: runSql } = await import("../runSql.js"); - -class Response extends EventEmitter { - constructor() { - super(); - this.statusCode = 200; - this.headersSent = false; - this.writableEnded = false; - this.payloads = []; - } - - status(value) { - this.statusCode = value; - return this; - } - - set() { - return this; - } - - type() { - return this; - } - - send(value) { - this.headersSent = true; - this.writableEnded = true; - this.payloads.push(value); - this.emit("finish"); - return this; - } - - json(value) { - return this.send(value); - } -} - -const request = () => ({ - body: { query: "SELECT * FROM enrich.day_context_v", format: "json" }, - securityContext: { - userId: "person-1", - tokenPayload: { accountId: "account-1", partition: "tenant.is" }, - userScope: { - dataSource: { dataSourceId: "datasource-1", dbType: "clickhouse" }, - }, - }, - get: (name) => (name === "x-request-id" ? "request-1" : null), -}); -const cubejs = { - options: { - driverFactory: async () => ({ query: async () => [{ value: 1 }] }), - }, -}; - -describe("run-sql billing commit", () => { - beforeEach(() => { - commitMock.mock.resetCalls(); - commitMock.mock.mockImplementation(async () => []); - }); - - it("durably meters after execution and before returning a successful result", async () => { - const res = new Response(); - await runSql(request(), res, cubejs); - assert.equal(commitMock.mock.callCount(), 1); - assert.equal(commitMock.mock.calls[0].arguments[1].returnedRows, 1); - assert.equal(commitMock.mock.calls[0].arguments[1].surface, "run-sql"); - assert.equal(res.statusCode, 200); - assert.equal(res.payloads[0], '[{"value":1}]'); - }); - - it("returns an error instead of a successful result when enqueue fails", async () => { - commitMock.mock.mockImplementation(async () => { - const error = new Error("outbox unavailable"); - error.status = 503; - throw error; - }); - const res = new Response(); - await runSql(request(), res, cubejs); - assert.equal(res.statusCode, 503); - assert.deepEqual(res.payloads, [ - { code: "run_sql_failed", message: "outbox unavailable" }, - ]); - }); -}); diff --git a/services/cubejs/src/routes/columnValues.js b/services/cubejs/src/routes/columnValues.js index b8722b2f..1edb5a05 100644 --- a/services/cubejs/src/routes/columnValues.js +++ b/services/cubejs/src/routes/columnValues.js @@ -10,7 +10,7 @@ */ import { buildWhereClause } from "../utils/smart-generation/profiler.js"; -import { assertNoDirectEnrichmentObject } from "../utils/enrichmentEntitlement.js"; +import { assertNoDirectLegacyEnrichmentObject } from "../utils/legacyEnrichmentGuard.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; export default async (req, res, cubejs) => { @@ -36,7 +36,7 @@ export default async (req, res, cubejs) => { let driver; try { - assertNoDirectEnrichmentObject(schema, table); + assertNoDirectLegacyEnrichmentObject(schema, table); // Same extraction as profile-table — all values from team settings in the database const partition = securityContext.userScope?.dataSource?.partition || null; const internalTables = diff --git a/services/cubejs/src/routes/discoverNested.js b/services/cubejs/src/routes/discoverNested.js index 0e95471b..1775b89b 100644 --- a/services/cubejs/src/routes/discoverNested.js +++ b/services/cubejs/src/routes/discoverNested.js @@ -1,4 +1,4 @@ -import { assertNoDirectEnrichmentObject } from "../utils/enrichmentEntitlement.js"; +import { assertNoDirectLegacyEnrichmentObject } from "../utils/legacyEnrichmentGuard.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; /** Naming patterns that indicate a lookup/discriminator column. */ @@ -37,7 +37,7 @@ export default async function discoverNested(req, res, cubejs) { let driver; try { - assertNoDirectEnrichmentObject(schema, table); + assertNoDirectLegacyEnrichmentObject(schema, table); driver = await tenantDriverFactory(cubejs)({ securityContext }); // 1. Fetch all columns for the table from system.columns diff --git a/services/cubejs/src/routes/dynamicMeta.js b/services/cubejs/src/routes/dynamicMeta.js index 63c409de..27b94761 100644 --- a/services/cubejs/src/routes/dynamicMeta.js +++ b/services/cubejs/src/routes/dynamicMeta.js @@ -12,7 +12,7 @@ import { shapeJsonEntries, createProbeCache, } from "../utils/dynamicPropertyProbe.js"; -import { assertNoDirectEnrichmentObject } from "../utils/enrichmentEntitlement.js"; +import { assertNoDirectLegacyEnrichmentObject } from "../utils/legacyEnrichmentGuard.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; /** @@ -122,7 +122,7 @@ export default async function dynamicMeta(req, res, cubejs, deps = {}) { const tableParts = String(table) .replace(/[`"\[\]]/g, "") .split("."); - assertNoDirectEnrichmentObject( + assertNoDirectLegacyEnrichmentObject( tableParts.length > 1 ? tableParts.at(-2) : null, table, ); diff --git a/services/cubejs/src/routes/generateDataSchema.js b/services/cubejs/src/routes/generateDataSchema.js index 66b30a96..60680d8d 100644 --- a/services/cubejs/src/routes/generateDataSchema.js +++ b/services/cubejs/src/routes/generateDataSchema.js @@ -7,7 +7,7 @@ import { import { emitModelEvent } from "../utils/eventEmitter.js"; import createMd5Hex from "../utils/md5Hex.js"; import { NO_SCHEMA_KEY } from "./getSchema.js"; -import { removeEnrichmentSchema } from "../utils/enrichmentEntitlement.js"; +import { removeLegacyEnrichmentSchema } from "../utils/legacyEnrichmentGuard.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; const camelize = (value) => value.replace(/_([a-z0-9])/g, (_, c) => c.toUpperCase()); @@ -110,7 +110,7 @@ export default async (req, res, cubejs) => { try { driver = await tenantDriverFactory(cubejs)({ securityContext }); - let schema = removeEnrichmentSchema(await driver.tablesSchema()); + let schema = removeLegacyEnrichmentSchema(await driver.tablesSchema()); const { tables = [], overwrite = false, diff --git a/services/cubejs/src/routes/getSchema.js b/services/cubejs/src/routes/getSchema.js index d2f9dc7d..9bb4f662 100644 --- a/services/cubejs/src/routes/getSchema.js +++ b/services/cubejs/src/routes/getSchema.js @@ -1,4 +1,4 @@ -import { removeEnrichmentSchema } from "../utils/enrichmentEntitlement.js"; +import { removeLegacyEnrichmentSchema } from "../utils/legacyEnrichmentGuard.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; export const NO_SCHEMA_KEY = "no_schema"; @@ -20,7 +20,7 @@ export default async (req, res, cubejs) => { try { driver = await tenantDriverFactory(cubejs)({ securityContext }); - const schema = removeEnrichmentSchema(await driver.tablesSchema()); + const schema = removeLegacyEnrichmentSchema(await driver.tablesSchema()); if (schema?.[""]) { schema[NO_SCHEMA_KEY] = schema[""]; diff --git a/services/cubejs/src/routes/loadExport.js b/services/cubejs/src/routes/loadExport.js index 0bd4702c..ce6daa59 100644 --- a/services/cubejs/src/routes/loadExport.js +++ b/services/cubejs/src/routes/loadExport.js @@ -26,11 +26,9 @@ import { import { emitQueryEvent } from "../utils/eventEmitter.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; import { - assertEnrichmentQueryAuthorized, - assertSqlEnrichmentAuthorized, -} from "../utils/enrichmentEntitlement.js"; -import { commitEnrichmentBilling } from "../utils/enrichmentMetering.js"; -import redisClient from "../utils/redis.js"; + assertNoLegacyEnrichmentQuery, + assertNoLegacyEnrichmentSql, +} from "../utils/legacyEnrichmentGuard.js"; const prepareAnnotation = typeof prepareAnnotationModule.prepareAnnotation === "function" @@ -303,7 +301,7 @@ async function buildLoadExportPlan(req, res, cubejs, query) { try { await apiGateway.assertApiScope("data", context.securityContext); - assertEnrichmentQueryAuthorized(query, context.securityContext); + assertNoLegacyEnrichmentQuery(query); const [queryType, normalizedQueries] = await apiGateway.getNormalizedQueries(query, context, true); @@ -373,8 +371,8 @@ async function buildLoadExportPlan(req, res, cubejs, query) { } } -export function authorizeNativeEnrichmentSql(sql, securityContext, options) { - assertSqlEnrichmentAuthorized(sql, securityContext, options); +export function authorizeNativeLegacySql(sql) { + assertNoLegacyEnrichmentSql(sql); } async function prepareNativeClickHouseExport(plan) { @@ -478,17 +476,6 @@ async function executeNativeClickHouseArrow( return Number(result.summary?.result_rows || 0); } -function countedRows(stream) { - const state = { count: 0 }; - state.rows = (async function* count() { - for await (const row of stream) { - state.count += 1; - yield row; - } - })(); - return state; -} - async function streamSemanticRows(plan) { const adapterApi = await plan.apiGateway.getAdapterApi(plan.context); return adapterApi.streamQuery(plan.streamingQuery); @@ -518,22 +505,6 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { const exportStart = Date.now(); let exported = false; let exportPath = null; - const commitExportBilling = (returnedRows) => - commitEnrichmentBilling( - redisClient, - { - query, - logicalExecutionId: plan.context.requestId, - signal: abortController.signal, - context: plan.context, - }, - { data: [] }, - { - surface: "export", - returnedRows, - cacheStatus: "cache_miss", - }, - ); const emitDatasetExported = (status, extra) => emitQueryEvent({ event: "Dataset Exported", @@ -568,15 +539,12 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { } if (format === "csv" && nativeQuery?.query) { - authorizeNativeEnrichmentSql( - nativeQuery.query, - plan.context.securityContext, - ); + authorizeNativeLegacySql(nativeQuery.query); const driver = await tenantDriverFactory(cubejs)({ securityContext: plan.context.securityContext, }); res.set(CSV_HEADERS); - const returnedRows = await executeNativeClickHouseCsv( + await executeNativeClickHouseCsv( res, nativeQuery.query, nativeQuery.values, @@ -584,7 +552,6 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { abortController.signal, getAliasNameToMember(plan), ); - await commitExportBilling(returnedRows); exported = true; exportPath = "native-clickhouse"; res.end(); @@ -592,23 +559,19 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { } if (format === "arrow" && nativeQuery?.query) { - authorizeNativeEnrichmentSql( - nativeQuery.query, - plan.context.securityContext, - ); + authorizeNativeLegacySql(nativeQuery.query); const driver = await tenantDriverFactory(cubejs)({ securityContext: plan.context.securityContext, }); res.set(ARROW_HEADERS); setNativeArrowFieldMappingHeaders(res, plan); - const returnedRows = await executeNativeClickHouseArrow( + await executeNativeClickHouseArrow( res, nativeQuery.query, nativeQuery.values, driver, abortController.signal, ); - await commitExportBilling(returnedRows); exported = true; exportPath = "native-clickhouse"; res.end(); @@ -620,15 +583,13 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { } const stream = await streamSemanticRows(plan); - const counted = countedRows(stream); if (format === "csv") { res.set(CSV_HEADERS); - await writeRowStreamAsCSV(res, counted.rows, { + await writeRowStreamAsCSV(res, stream, { columns: plan.columns, signal: abortController.signal, }); - await commitExportBilling(counted.count); exported = true; exportPath = "semantic-stream"; res.end(); @@ -636,12 +597,11 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { } res.set(ARROW_HEADERS); - await writeRowStreamAsArrow(res, counted.rows, { + await writeRowStreamAsArrow(res, stream, { columns: plan.columns, annotation: plan.annotation, signal: abortController.signal, }); - await commitExportBilling(counted.count); exported = true; exportPath = "semantic-stream"; res.end(); diff --git a/services/cubejs/src/routes/profileTable.js b/services/cubejs/src/routes/profileTable.js index c48d6a87..3c706afc 100644 --- a/services/cubejs/src/routes/profileTable.js +++ b/services/cubejs/src/routes/profileTable.js @@ -9,7 +9,7 @@ import { ColumnType } from "../utils/smart-generation/typeParser.js"; import { parseCubesFromJs } from "../utils/smart-generation/diffModels.js"; import { loadRules } from "../utils/queryRewrite.js"; import { emitQueryEvent } from "../utils/eventEmitter.js"; -import { assertNoDirectEnrichmentObject } from "../utils/enrichmentEntitlement.js"; +import { assertNoDirectLegacyEnrichmentObject } from "../utils/legacyEnrichmentGuard.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; /** @@ -136,7 +136,7 @@ export default async (req, res, cubejs) => { let driver; try { - assertNoDirectEnrichmentObject(schema, table); + assertNoDirectLegacyEnrichmentObject(schema, table); const partition = securityContext.userScope?.dataSource?.partition || null; const internalTables = securityContext.userScope?.dataSource?.internalTables || []; diff --git a/services/cubejs/src/routes/reconcileTeam.js b/services/cubejs/src/routes/reconcileTeam.js index 11a79a4a..d0964ae3 100644 --- a/services/cubejs/src/routes/reconcileTeam.js +++ b/services/cubejs/src/routes/reconcileTeam.js @@ -57,7 +57,6 @@ export async function reconcileTeamCore(params, deps) { systemUserId, partition, internalTables = [], - revokeTemplates = [], } = params; const current = (await deps.loadCurrentSchemas()) || []; @@ -73,62 +72,6 @@ export async function reconcileTeamCore(params, deps) { const templateNames = new Set(templates.map((t) => t.name)); const probeCache = new Map(); - // Premium entitlement revocation is destructive only toward cubes carrying - // both the feature's managed_by stamp and an explicitly revoked template - // name. Team-authored cubes sharing the same file survive unchanged. - const revokeNames = new Set(revokeTemplates); - if (revokeNames.size > 0) { - for (const file of current) { - let doc; - try { - doc = YAML.parse(file.code); - } catch { - continue; - } - if (!Array.isArray(doc?.cubes)) continue; - const removed = doc.cubes.filter( - (cube) => - cube?.meta?.managed_by === "ctx-enrichment" && - revokeNames.has(cube?.meta?.template) - ); - if (removed.length === 0) continue; - - const kept = doc.cubes.filter((cube) => !removed.includes(cube)); - const trial = new Map(working); - if (kept.length === 0) { - trial.delete(file.name); - } else { - trial.set(file.name, { - name: file.name, - code: YAML.stringify({ ...doc, cubes: kept }, { lineWidth: 0 }), - }); - } - const validation = await deps.validate([...trial.values()]); - const retiredTemplates = [...new Set(removed.map((cube) => cube.meta.template))]; - if (!validation.valid) { - for (const template of retiredTemplates) { - outcomes.push({ - template, - result: "failed", - reason: `entitlement revocation failed validation: ${formatErrors(validation.errors)}`, - }); - } - continue; - } - working.clear(); - for (const [name, value] of trial) working.set(name, value); - for (const template of retiredTemplates) { - const outcome = { - template, - result: "removed", - reason: "entitlement_revoked", - }; - outcomes.push(outcome); - pendingUpdated.push(outcome); - } - } - } - // Lazy baseline compile of the ORIGINAL current set: discriminates // "this template broke the branch" (counts toward rollout halt) from // "the branch was already broken before this run" (reported, but a @@ -185,7 +128,6 @@ export async function reconcileTeamCore(params, deps) { (c) => c?.meta?.default_model === true && c?.meta?.template && - !revokeNames.has(c.meta.template) && !templateNames.has(c.meta.template) && c?.meta?.default_model_unmanaged !== true ); @@ -202,7 +144,6 @@ export async function reconcileTeamCore(params, deps) { if ( cube?.meta?.default_model === true && cube?.meta?.template && - !revokeNames.has(cube.meta.template) && !templateNames.has(cube.meta.template) && cube?.meta?.default_model_unmanaged !== true ) { @@ -557,7 +498,6 @@ export default async function reconcileTeam(req, res, cubejs) { templates, optOut = [], dryRun = false, - revokeTemplates = [], } = req.body || {}; if ( @@ -565,8 +505,7 @@ export default async function reconcileTeam(req, res, cubejs) { !datasourceId || !branchId || !partition || - !Array.isArray(templates) || - !Array.isArray(revokeTemplates) + !Array.isArray(templates) ) { return res.status(400).json({ code: "invalid_input", @@ -740,7 +679,6 @@ export default async function reconcileTeam(req, res, cubejs) { dryRun, systemUserId, internalTables, - revokeTemplates, }, deps ); diff --git a/services/cubejs/src/routes/runSql.js b/services/cubejs/src/routes/runSql.js index ac08bb60..ab345b15 100644 --- a/services/cubejs/src/routes/runSql.js +++ b/services/cubejs/src/routes/runSql.js @@ -6,9 +6,7 @@ import { validateFormat } from "../utils/formatValidator.js"; import { writeRowsAsCSV, writeTextChunk } from "../utils/csvSerializer.js"; import { buildJSONStat } from "../utils/jsonstatBuilder.js"; import { emitQueryEvent } from "../utils/eventEmitter.js"; -import { assertSqlEnrichmentAuthorized } from "../utils/enrichmentEntitlement.js"; -import { commitSqlEnrichmentBilling } from "../utils/enrichmentMetering.js"; -import redisClient from "../utils/redis.js"; +import { assertNoLegacyEnrichmentSql } from "../utils/legacyEnrichmentGuard.js"; import tenantDriverFactory from "../utils/tenantDriverFactory.js"; const { JWT_KEY } = process.env; @@ -30,8 +28,8 @@ function isSignedSql(sql, signature) { ); } -export function authorizeRunSqlQuery(sql, securityContext, options) { - assertSqlEnrichmentAuthorized(sql, securityContext, options); +export function authorizeRunSqlQuery(sql) { + assertNoLegacyEnrichmentSql(sql); } const CSV_HEADERS = { @@ -158,22 +156,6 @@ export default async (req, res, cubejs) => { const dataSourceId = securityContext?.userScope?.dataSource?.dataSourceId ?? null; const dbType = securityContext?.userScope?.dataSource?.dbType ?? null; - const logicalExecutionId = - req.get?.("x-idempotency-key") || - req.get?.("x-request-id") || - req.get?.("traceparent") || - crypto.randomUUID(); - const meterResult = (sql, returnedRows, signal = null) => - commitSqlEnrichmentBilling(redisClient, { - sql, - securityContext, - logicalExecutionId, - surface: "run-sql", - returnedRows, - cacheStatus: "cache_miss", - signal, - }); - if (!req.body.query) { res.status(400).json({ code: "query_missing", @@ -222,10 +204,10 @@ export default async (req, res, cubejs) => { auditFormat = format; const sql = req.body.query; - // A gen_sql signature proves provenance, not premium entitlement. Apply - // the same parser-based table policy before JSON, JSON-Stat, CSV, Arrow, - // generic driver, or native ClickHouse execution can diverge. - authorizeRunSqlQuery(sql, securityContext); + // The retired physical enrichment namespace remains unavailable before + // JSON, JSON-Stat, CSV, Arrow, generic driver, or native ClickHouse + // execution can diverge. + authorizeRunSqlQuery(sql); // Block freeform SQL when access control rules are active. // SQL that was generated by gen_sql is HMAC-signed — if the signature @@ -277,7 +259,6 @@ export default async (req, res, cubejs) => { } auditRowCount = rows.length; const body = JSON.stringify(rows); - await meterResult(sql, rows.length, abortController.signal); res.type("json").send(body); return; } @@ -299,8 +280,6 @@ export default async (req, res, cubejs) => { } const body = JSON.stringify(dataset); - await meterResult(sql, rows.length, abortController.signal); - res.set("Content-Type", "application/json"); res.set( "Content-Disposition", @@ -332,11 +311,6 @@ export default async (req, res, cubejs) => { throw streamErr; } - await meterResult( - sql, - Number(result.summary?.result_rows || 0), - abortController.signal, - ); res.end(); return; } @@ -380,7 +354,6 @@ export default async (req, res, cubejs) => { throw streamErr; } - await meterResult(sql, rowCount, abortController.signal); res.end(); return; } @@ -401,14 +374,12 @@ export default async (req, res, cubejs) => { } const columns = deriveExportColumnsFromRunSql(req.body, rows); const body = serializeRowsToArrow(rows, { columns }); - await meterResult(sql, rows.length, abortController.signal); res.set(ARROW_HEADERS); res.send(body); return; } if (!rows || rows.length === 0) { - await meterResult(sql, 0, abortController.signal); res.set(CSV_HEADERS); res.set("Content-Length", "0"); res.send(""); @@ -417,7 +388,6 @@ export default async (req, res, cubejs) => { res.set(CSV_HEADERS); await writeRowsAsCSV(res, rows, { signal: abortController.signal }); - await meterResult(sql, rows.length, abortController.signal); res.end(); } catch (err) { console.error(err); diff --git a/services/cubejs/src/routes/smartGenerate.js b/services/cubejs/src/routes/smartGenerate.js index 2c0df05c..ed084fea 100644 --- a/services/cubejs/src/routes/smartGenerate.js +++ b/services/cubejs/src/routes/smartGenerate.js @@ -42,7 +42,7 @@ import { smokeTestQuery, } from "../utils/smart-generation/modelValidator.js"; import { fetchClickHouseAliasColumnNames } from "../utils/smart-generation/clickHouseAliasColumns.js"; -import { assertNoDirectEnrichmentObject } from "../utils/enrichmentEntitlement.js"; +import { assertNoDirectLegacyEnrichmentObject } from "../utils/legacyEnrichmentGuard.js"; function reorderProfileColumns(profiledTable) { if (!profiledTable?.columns || !(profiledTable.columns instanceof Map)) @@ -218,7 +218,7 @@ export default async (req, res, cubejs) => { let driver; try { - assertNoDirectEnrichmentObject(schema, table); + assertNoDirectLegacyEnrichmentObject(schema, table); const { userId } = securityContext; const partition = securityContext.userScope?.dataSource?.partition || null; // 099 T087/T088 (FR-091): one tenant-attribution source of truth, shared by diff --git a/services/cubejs/src/utils/__tests__/billingMetrics.test.js b/services/cubejs/src/utils/__tests__/billingMetrics.test.js deleted file mode 100644 index 2585e332..00000000 --- a/services/cubejs/src/utils/__tests__/billingMetrics.test.js +++ /dev/null @@ -1,72 +0,0 @@ -import assert from "node:assert/strict"; -import { beforeEach, describe, it } from "node:test"; - -import { - billingMetricsSnapshot, - incrementBillingMetric, - readBillingOutboxGauges, - renderBillingMetrics, - resetBillingMetricsForTest, -} from "../billingMetrics.js"; - -describe("bounded billing Prometheus metrics", () => { - beforeEach(() => resetBillingMetricsForTest()); - - it("accepts only the closed counter vocabulary", () => { - incrementBillingMetric("emitted", 2); - incrementBillingMetric("deduplicated"); - assert.equal(billingMetricsSnapshot().emitted, 2); - assert.equal(billingMetricsSnapshot().deduplicated, 1); - assert.throws( - () => incrementBillingMetric("tenant-account-1"), - /unknown billing metric/, - ); - }); - - it("reads aggregate stream, pending, DLQ, and oldest-age gauges", async () => { - const nowMs = 2_000_000; - const redis = { - xlen: async () => 3, - xpending: async () => [4, "1000000-0", "1900000-0", []], - xinfo: async () => [ - [ - "name", - "synmetrix-billing-delivery", - "pending", - 4, - "last-delivered-id", - "1100000-0", - "lag", - 8, - ], - ], - xrange: async () => [["1200000-0", []]], - }; - assert.deepEqual(await readBillingOutboxGauges(redis, { nowMs }), { - backlog: 12, - pending: 4, - dlq: 3, - oldestUndeliveredAgeSeconds: 1000, - }); - }); - - it("renders no sensitive labels or values", () => { - incrementBillingMetric("unmeterable"); - const output = renderBillingMetrics( - billingMetricsSnapshot(), - { - backlog: 2, - pending: 1, - dlq: 0, - oldestUndeliveredAgeSeconds: 4, - }, - { workerUp: true }, - ); - assert.match(output, /synmetrix_billing_unmeterable_total 1/); - assert.match(output, /synmetrix_billing_outbox_worker_up 1/); - assert.doesNotMatch( - output, - /account|partition|tenant|coordinate|geohash|execution|query|sql/i, - ); - }); -}); diff --git a/services/cubejs/src/utils/__tests__/billingOutbox.test.js b/services/cubejs/src/utils/__tests__/billingOutbox.test.js deleted file mode 100644 index f7c4e534..00000000 --- a/services/cubejs/src/utils/__tests__/billingOutbox.test.js +++ /dev/null @@ -1,213 +0,0 @@ -import assert from "node:assert/strict"; -import { describe, it } from "node:test"; - -import { - enqueueBillingBatch, - enqueueBillingEvent, - loadAndReplayBillingDlqEntry, - replayBillingDlqEntry, -} from "../billingOutbox.js"; -import { - BillingOutboxWorker, - processBillingEntry, -} from "../billingOutboxWorker.js"; - -const ENVELOPE = { - event: "Connection Called", - message_id: "billing-message-1", - involves: [], -}; -const CONTEXT = { accountId: "account-1", partition: "tenant.is" }; - -describe("durable billing outbox", () => { - it("atomically enqueues a complete multi-item result commit", async () => { - let evalArgs; - const redis = { - eval: async (...args) => { - evalArgs = args; - return ["1700000000000-0", "duplicate"]; - }, - }; - const results = await enqueueBillingBatch(redis, [ - { envelope: ENVELOPE, context: CONTEXT }, - { - envelope: { ...ENVELOPE, message_id: "billing-message-2" }, - context: CONTEXT, - }, - ]); - assert.match(evalArgs[0], /for index = 1, count/); - assert.match(evalArgs[0], /XADD/); - assert.equal(evalArgs[1], 3); - assert.equal(results[0].enqueued, true); - assert.equal(results[1].duplicate, true); - }); - - it("atomically deduplicates enqueue by deterministic message id", async () => { - const calls = []; - const redis = { - eval: async (...args) => { - calls.push(args); - return calls.length === 1 ? "1700000000000-0" : "duplicate"; - }, - }; - const first = await enqueueBillingEvent(redis, ENVELOPE, CONTEXT); - const second = await enqueueBillingEvent(redis, ENVELOPE, CONTEXT); - assert.equal(first.enqueued, true); - assert.equal(second.enqueued, false); - assert.match(calls[0][0], /XADD/); - assert.match(calls[0][0], /SET/); - assert.equal(calls[0].at(-3), ENVELOPE.message_id); - assert.match(calls[0][0], /MAXLEN/); - }); - - it("acknowledges only after ingress success", async () => { - const calls = []; - const redis = { - xack: async (...args) => calls.push(["ack", ...args]), - hdel: async (...args) => calls.push(["hdel", ...args]), - }; - const result = await processBillingEntry( - redis, - [ - "1-0", - [ - "envelope", - JSON.stringify(ENVELOPE), - "context", - JSON.stringify(CONTEXT), - ], - ], - { send: async () => ({ ok: true }) }, - ); - assert.equal(result, "acknowledged"); - assert.equal(calls[0][0], "ack"); - }); - - it("leaves a failed entry pending for restart/reclaim before the retry limit", async () => { - const calls = []; - const redis = { - hincrby: async () => 2, - xack: async (...args) => calls.push(["ack", ...args]), - xadd: async (...args) => calls.push(["dlq", ...args]), - }; - const result = await processBillingEntry( - redis, - [ - "2-0", - [ - "envelope", - JSON.stringify(ENVELOPE), - "context", - JSON.stringify(CONTEXT), - ], - ], - { send: async () => ({ ok: false }), maxAttempts: 3 }, - ); - assert.equal(result, "pending"); - assert.deepEqual(calls, []); - }); - - it("moves poison entries to a DLQ and acknowledges the original", async () => { - const calls = []; - const redis = { - hincrby: async () => 3, - xadd: async (...args) => calls.push(["dlq", ...args]), - xack: async (...args) => calls.push(["ack", ...args]), - hdel: async (...args) => calls.push(["hdel", ...args]), - }; - const result = await processBillingEntry( - redis, - [ - "3-0", - [ - "envelope", - JSON.stringify(ENVELOPE), - "context", - JSON.stringify(CONTEXT), - ], - ], - { send: async () => ({ ok: false }), maxAttempts: 3 }, - ); - assert.equal(result, "dead_lettered"); - assert.equal(calls[0][0], "dlq"); - assert.equal(calls[1][0], "ack"); - }); - - it("replays a DLQ entry with a distinct replay idempotency key", async () => { - let evalArgs; - const redis = { - eval: async (...args) => { - evalArgs = args; - return "4-0"; - }, - xdel: async () => 1, - }; - const result = await replayBillingDlqEntry(redis, "9-0", { - envelope: ENVELOPE, - context: CONTEXT, - }); - assert.equal(result.enqueued, true); - assert.match(evalArgs.at(-3), /billing-message-1:replay:9-0/); - }); - - it("loads an exact DLQ stream id for operator replay", async () => { - const redis = { - xrange: async () => [ - [ - "9-1", - [ - "envelope", - JSON.stringify(ENVELOPE), - "context", - JSON.stringify(CONTEXT), - ], - ], - ], - eval: async () => "10-0", - xdel: async () => 1, - }; - const result = await loadAndReplayBillingDlqEntry(redis, "9-1"); - assert.equal(result.enqueued, true); - }); - - it("reclaims pending entries after a worker restart", async () => { - const calls = []; - const redis = { - xautoclaim: async (...args) => { - calls.push(["claim", ...args]); - return [ - "0-0", - [ - [ - "7-0", - [ - "envelope", - JSON.stringify(ENVELOPE), - "context", - JSON.stringify(CONTEXT), - ], - ], - ], - ]; - }, - xack: async (...args) => calls.push(["ack", ...args]), - hdel: async () => 1, - }; - const worker = new BillingOutboxWorker(redis, { - consumer: "replacement-worker", - send: async () => ({ ok: true }), - }); - await worker.reclaim(); - assert.equal(calls[0][0], "claim"); - assert.equal(calls[1][0], "ack"); - }); - - it("treats an existing consumer group as an idempotent restart", async () => { - const worker = new BillingOutboxWorker({ - xgroup: async () => { - throw new Error("BUSYGROUP Consumer Group name already exists"); - }, - }); - await assert.doesNotReject(worker.ensureGroup()); - }); -}); diff --git a/services/cubejs/src/utils/__tests__/enrichmentBilling.live.test.js b/services/cubejs/src/utils/__tests__/enrichmentBilling.live.test.js deleted file mode 100644 index e0098f15..00000000 --- a/services/cubejs/src/utils/__tests__/enrichmentBilling.live.test.js +++ /dev/null @@ -1,537 +0,0 @@ -import assert from "node:assert/strict"; -import { randomUUID } from "node:crypto"; -import { readFile } from "node:fs/promises"; -import http from "node:http"; -import https from "node:https"; -import { after, before, describe, it } from "node:test"; - -import Redis from "ioredis"; -import pg from "pg"; - -import { - BILLING_ATTEMPTS_HASH, - BILLING_DLQ_STREAM, - BILLING_GROUP, - BILLING_STREAM, - enqueueBillingEvent, - loadAndReplayBillingDlqEntry, -} from "../billingOutbox.js"; -import { - BillingOutboxWorker, - processBillingEntry, -} from "../billingOutboxWorker.js"; - -/* - * Opt-in real-service gate. The request manifest contains only query shapes, - * expected items, surfaces, and frozen public prices; credentials remain in - * environment variables. The Redis lane refuses a non-empty keyspace and - * removes only the exact keys it creates. - */ - -const SERVICE_ENABLED = process.env.SPEC102_LIVE_BILLING_SERVICE === "1"; -const REDIS_ENABLED = process.env.SPEC102_LIVE_BILLING_REDIS === "1"; -const REQUIRED_HTTP_SURFACES = new Set(["rest", "run-sql", "export"]); -const REQUIRED_CASES = new Set([ - "day", - "weather", - "both_items", - "cache_hit", - "zero_rows", - "stable_retry", - "disconnect", - "non_entitled", - "engine_failure", - "system_query", - "pre_aggregation", -]); -const EXACT_CASE_ITEMS = new Map([ - ["day", ["ctx:day-archetype"]], - ["weather", ["ctx:weather-archetype"]], - ["both_items", ["ctx:day-archetype", "ctx:weather-archetype"]], -]); -const REQUIRED_BILLABLE_CASES = new Set([ - "cache_hit", - "zero_rows", - "stable_retry", - "disconnect", -]); -const REQUIRED_UNBILLED_CASES = new Set([ - "non_entitled", - "engine_failure", - "system_query", - "pre_aggregation", -]); -const BILLING_KEYS = [ - BILLING_STREAM, - BILLING_DLQ_STREAM, - BILLING_ATTEMPTS_HASH, -]; - -function requiredEnv(name) { - const value = String(process.env[name] || "").trim(); - if (!value) throw new Error(`${name} is required for the live billing gate`); - return value; -} - -function sleep(ms) { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -async function readManifest() { - const path = requiredEnv("SPEC102_LIVE_BILLING_MANIFEST"); - const manifest = JSON.parse(await readFile(path, "utf8")); - assert.equal(manifest.schema_version, 1); - assert.ok(Array.isArray(manifest.requests)); - assert.ok(manifest.requests.length > 0); - - const cases = new Set( - manifest.requests.flatMap((request) => request.cases || []), - ); - const surfaces = new Set(manifest.requests.map((request) => request.surface)); - for (const name of REQUIRED_CASES) { - assert.ok(cases.has(name), `live manifest is missing case ${name}`); - } - for (const surface of REQUIRED_HTTP_SURFACES) { - assert.ok( - surfaces.has(surface), - `live manifest is missing surface ${surface}`, - ); - } - assert.ok(manifest.sql_api?.sql, "live manifest is missing SQL API coverage"); - assert.ok(Array.isArray(manifest.sql_api.expected_items)); - assert.ok(manifest.sql_api.expected_items.length > 0); - - for (const request of manifest.requests) { - assert.ok(request.name); - assert.ok(request.path?.startsWith("/")); - assert.ok(["GET", "POST"].includes(request.method || "POST")); - assert.ok(Array.isArray(request.expected_items)); - assert.equal( - new Set(request.expected_items).size, - request.expected_items.length, - `${request.name} contains duplicate expected items`, - ); - if (request.cases?.includes("non_entitled")) { - assert.equal(request.auth, "non_entitled"); - } - for (const [name, items] of EXACT_CASE_ITEMS) { - if (request.cases?.includes(name)) { - assert.deepEqual( - [...request.expected_items].sort(), - [...items].sort(), - `${request.name} does not exercise the exact ${name} item set`, - ); - } - } - for (const name of REQUIRED_BILLABLE_CASES) { - if (request.cases?.includes(name)) { - assert.ok( - request.expected_items.length > 0, - `${request.name} cannot prove billable case ${name} without an item`, - ); - } - } - for (const name of REQUIRED_UNBILLED_CASES) { - if (request.cases?.includes(name)) { - assert.deepEqual( - request.expected_items, - [], - `${request.name} must not bill case ${name}`, - ); - } - } - } - for (const item of ["ctx:day-archetype", "ctx:weather-archetype"]) { - const price = manifest.expected_pricing?.[item]; - assert.equal(price?.source, "legacy_runtime_rate"); - assert.ok(String(price?.code_version || "").trim()); - assert.ok(/^\d+(?:\.\d+)?$/.test(String(price?.unit_amount || ""))); - assert.ok(/^[A-Z]{3}$/.test(String(price?.currency || ""))); - } - return manifest; -} - -function requestHeaders(token, logicalExecutionId, extra = {}) { - return { - authorization: `Bearer ${token}`, - "content-type": "application/json", - "x-idempotency-key": logicalExecutionId, - "x-request-id": logicalExecutionId, - ...extra, - }; -} - -async function ordinaryRequest(baseUrl, token, scenario, logicalExecutionId) { - const response = await fetch(new URL(scenario.path, baseUrl), { - method: scenario.method || "POST", - headers: requestHeaders(token, logicalExecutionId, scenario.headers), - body: - (scenario.method || "POST") === "GET" - ? undefined - : JSON.stringify(scenario.body || {}), - }); - assert.equal(response.status, scenario.expected_status ?? 200, scenario.name); - await response.arrayBuffer(); -} - -async function disconnectRequest(baseUrl, token, scenario, logicalExecutionId) { - const url = new URL(scenario.path, baseUrl); - const transport = url.protocol === "https:" ? https : http; - const body = JSON.stringify(scenario.body || {}); - await new Promise((resolve, reject) => { - const request = transport.request( - url, - { - method: scenario.method || "POST", - headers: { - ...requestHeaders(token, logicalExecutionId, scenario.headers), - "content-length": Buffer.byteLength(body), - }, - }, - (response) => { - response.resume(); - response.once("end", resolve); - }, - ); - request.once("error", (error) => { - if (error.code === "ECONNRESET") resolve(); - else reject(error); - }); - request.end(body); - setTimeout( - () => { - request.destroy(); - resolve(); - }, - Number(scenario.disconnect_after_ms || 25), - ); - }); -} - -async function queryLedger({ - fromUtc, - accountGid, - logicalPrefix = "", - surface = "", -}) { - const endpoint = new URL(requiredEnv("SPEC102_CLICKHOUSE_HTTP_URL")); - endpoint.searchParams.set("param_from_utc", fromUtc); - endpoint.searchParams.set("param_account_gid", accountGid); - endpoint.searchParams.set( - "param_logical_prefix", - logicalPrefix ? `${logicalPrefix}%` : "", - ); - endpoint.searchParams.set("param_surface", surface); - const auth = Buffer.from( - `${requiredEnv("SPEC102_CLICKHOUSE_USER")}:${requiredEnv("SPEC102_CLICKHOUSE_PASSWORD")}`, - ).toString("base64"); - const sql = ` - SELECT - JSONExtractString(toString(properties), 'logical_execution_id') AS logical_execution_id, - dimensions['item'] AS item, - dimensions['surface'] AS surface, - count() AS ledger_rows, - uniqExact(message_id) AS distinct_messages, - any(JSONExtractString(toString(properties), 'pricing_source')) AS pricing_source, - any(JSONExtractString(toString(properties), 'pricing_code_version')) AS pricing_code_version, - any(JSONExtractString(toString(properties), 'unit_amount')) AS unit_amount, - any(analysis.currency[1]) AS currency - FROM cst.semantic_events - WHERE event = 'Connection Called' - AND dimensions['provider'] = 'ctx' - AND dimensions['accounting_scope'] = 'customer_usage' - AND timestamp >= parseDateTimeBestEffort({from_utc:String}) - AND toString(entity_gid) = {account_gid:String} - AND ({logical_prefix:String} = '' OR logical_execution_id LIKE {logical_prefix:String}) - AND ({surface:String} = '' OR dimensions['surface'] = {surface:String}) - GROUP BY logical_execution_id, item, surface - ORDER BY logical_execution_id, item - FORMAT JSON - `; - const response = await fetch(endpoint, { - method: "POST", - headers: { authorization: `Basic ${auth}` }, - body: sql, - }); - assert.equal(response.status, 200, "event-store reconciliation query failed"); - const result = await response.json(); - return result.data || []; -} - -async function waitForLedger(expectedCount, query) { - const timeoutMs = Number( - process.env.SPEC102_LIVE_BILLING_TIMEOUT_MS || 300_000, - ); - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - const rows = await query(); - if (rows.length >= expectedCount) { - await sleep(2_000); - return query(); - } - await sleep(2_000); - } - throw new Error( - "billing ledger did not reconcile within the configured timeout", - ); -} - -describe( - "Spec 102 real-service billing matrix", - { skip: !SERVICE_ENABLED }, - () => { - let manifest; - let runPrefix; - let startedAt; - let token; - let baseUrl; - let accountGid; - let sqlAccountGid; - - before(async () => { - manifest = await readManifest(); - runPrefix = `spec102-${randomUUID()}`; - startedAt = new Date(Date.now() - 1_000).toISOString(); - token = requiredEnv("SPEC102_LIVE_BILLING_TOKEN"); - baseUrl = requiredEnv("SPEC102_LIVE_BILLING_BASE_URL"); - accountGid = requiredEnv("SPEC102_LIVE_ACCOUNT_GID"); - sqlAccountGid = requiredEnv("SPEC102_LIVE_SQL_ACCOUNT_GID"); - assert.notEqual( - sqlAccountGid, - accountGid, - "SQL API coverage requires a separate isolated test Account", - ); - }); - - it("drives every tenant HTTP surface and reconciles exact historic prices", async () => { - const expected = new Map(); - for (const scenario of manifest.requests) { - const logicalExecutionId = `${runPrefix}:${scenario.name}`; - const attempts = scenario.cases?.includes("stable_retry") ? 2 : 1; - const scenarioToken = - scenario.auth === "non_entitled" - ? requiredEnv("SPEC102_LIVE_NON_ENTITLED_TOKEN") - : token; - for (let attempt = 0; attempt < attempts; attempt += 1) { - if (scenario.cases?.includes("disconnect")) { - await disconnectRequest( - baseUrl, - scenarioToken, - scenario, - logicalExecutionId, - ); - } else { - await ordinaryRequest( - baseUrl, - scenarioToken, - scenario, - logicalExecutionId, - ); - } - } - for (const item of scenario.expected_items) { - expected.set(`${logicalExecutionId}\u0000${item}`, scenario.surface); - } - } - - const sqlClient = new pg.Client({ - host: requiredEnv("SPEC102_SQL_API_HOST"), - port: Number(process.env.SPEC102_SQL_API_PORT || 15432), - user: requiredEnv("SPEC102_SQL_API_USER"), - password: requiredEnv("SPEC102_SQL_API_PASSWORD"), - database: requiredEnv("SPEC102_SQL_API_DATABASE"), - ssl: - process.env.SPEC102_SQL_API_TLS === "0" - ? false - : { rejectUnauthorized: true }, - }); - await sqlClient.connect(); - try { - await sqlClient.query(manifest.sql_api.sql); - } finally { - await sqlClient.end(); - } - - const rows = await waitForLedger(expected.size, () => - queryLedger({ - fromUtc: startedAt, - accountGid, - logicalPrefix: runPrefix, - }), - ); - assert.equal( - rows.length, - expected.size, - "missing or unexpected billing rows", - ); - for (const row of rows) { - const key = `${row.logical_execution_id}\u0000${row.item}`; - assert.ok(expected.has(key), `unexpected logical charge ${key}`); - assert.equal(Number(row.ledger_rows), 1, key); - assert.equal(Number(row.distinct_messages), 1, key); - assert.equal(row.surface, expected.get(key)); - const price = manifest.expected_pricing[row.item]; - assert.equal(row.pricing_source, price.source); - assert.equal(row.pricing_code_version, price.code_version); - assert.equal(row.unit_amount, price.unit_amount); - assert.equal(row.currency, price.currency); - } - - const sqlRows = await waitForLedger( - manifest.sql_api.expected_items.length, - () => - queryLedger({ - fromUtc: startedAt, - accountGid: sqlAccountGid, - surface: "sql-api", - }), - ); - assert.equal(sqlRows.length, manifest.sql_api.expected_items.length); - assert.deepEqual( - sqlRows.map((row) => row.item).sort(), - [...manifest.sql_api.expected_items].sort(), - ); - for (const row of sqlRows) { - assert.equal(Number(row.ledger_rows), 1); - assert.equal(Number(row.distinct_messages), 1); - const price = manifest.expected_pricing[row.item]; - assert.equal(row.pricing_source, price.source); - assert.equal(row.pricing_code_version, price.code_version); - assert.equal(row.unit_amount, price.unit_amount); - assert.equal(row.currency, price.currency); - } - }); - }, -); - -describe( - "Spec 102 real Redis outage, restart, DLQ, and replay", - { skip: !REDIS_ENABLED }, - () => { - let redis; - let dedupeKeys = []; - - before(async () => { - redis = new Redis(requiredEnv("SPEC102_LIVE_BILLING_REDIS_URL"), { - lazyConnect: true, - maxRetriesPerRequest: 1, - }); - await redis.connect(); - const occupied = await redis.exists(...BILLING_KEYS); - assert.equal( - occupied, - 0, - "live billing Redis gate requires a dedicated empty keyspace", - ); - }); - - after(async () => { - if (!redis) return; - await redis.del(...BILLING_KEYS, ...dedupeKeys); - await redis.quit(); - }); - - it("survives outage and restart, then replays an exact DLQ entry", async () => { - const firstId = `spec102-redis-${randomUUID()}`; - const firstEnvelope = { - event: "Connection Called", - message_id: firstId, - involves: [], - }; - dedupeKeys.push(`synmetrix-billing-dedupe:${firstId}`); - await enqueueBillingEvent(redis, firstEnvelope, { synthetic: true }); - - const failedWorker = new BillingOutboxWorker(redis, { - consumer: "spec102-failed-worker", - reclaimIdleMs: 0, - send: async () => ({ ok: false }), - }); - await failedWorker.ensureGroup(); - const claimed = await redis.xreadgroup( - "GROUP", - BILLING_GROUP, - failedWorker.consumer, - "COUNT", - 1, - "STREAMS", - BILLING_STREAM, - ">", - ); - await failedWorker.process(claimed?.[0]?.[1] || []); - assert.equal( - Number((await redis.xpending(BILLING_STREAM, BILLING_GROUP))[0]), - 1, - ); - - const replacement = new BillingOutboxWorker(redis, { - consumer: "spec102-replacement-worker", - reclaimIdleMs: 0, - send: async () => ({ ok: true }), - }); - await replacement.reclaim(); - assert.equal( - Number((await redis.xpending(BILLING_STREAM, BILLING_GROUP))[0]), - 0, - ); - - const poisonId = `spec102-poison-${randomUUID()}`; - const poisonEnvelope = { - event: "Connection Called", - message_id: poisonId, - involves: [], - }; - dedupeKeys.push(`synmetrix-billing-dedupe:${poisonId}`); - await enqueueBillingEvent(redis, poisonEnvelope, { synthetic: true }); - const poisonRows = await redis.xreadgroup( - "GROUP", - BILLING_GROUP, - "spec102-poison-worker", - "COUNT", - 1, - "STREAMS", - BILLING_STREAM, - ">", - ); - const poisonEntry = poisonRows?.[0]?.[1]?.[0]; - assert.ok(poisonEntry); - assert.equal( - await processBillingEntry(redis, poisonEntry, { - send: async () => ({ ok: false }), - maxAttempts: 1, - }), - "dead_lettered", - ); - assert.equal(await redis.xlen(BILLING_DLQ_STREAM), 1); - - const [[dlqId]] = await redis.xrange( - BILLING_DLQ_STREAM, - "-", - "+", - "COUNT", - 1, - ); - dedupeKeys.push(`synmetrix-billing-dedupe:${poisonId}:replay:${dlqId}`); - const replay = await loadAndReplayBillingDlqEntry(redis, dlqId); - assert.equal(replay.enqueued, true); - assert.equal(await redis.xlen(BILLING_DLQ_STREAM), 0); - - const replayRows = await redis.xreadgroup( - "GROUP", - BILLING_GROUP, - "spec102-replay-worker", - "COUNT", - 1, - "STREAMS", - BILLING_STREAM, - ">", - ); - await processBillingEntry(redis, replayRows?.[0]?.[1]?.[0], { - send: async () => ({ ok: true }), - }); - assert.equal( - Number((await redis.xpending(BILLING_STREAM, BILLING_GROUP))[0]), - 0, - ); - }); - }, -); diff --git a/services/cubejs/src/utils/__tests__/enrichmentEntitlement.test.js b/services/cubejs/src/utils/__tests__/enrichmentEntitlement.test.js deleted file mode 100644 index 3051e835..00000000 --- a/services/cubejs/src/utils/__tests__/enrichmentEntitlement.test.js +++ /dev/null @@ -1,278 +0,0 @@ -import assert from "node:assert/strict"; -import { createHmac } from "node:crypto"; -import { describe, it } from "node:test"; - -import { - assertEnrichmentQueryAuthorized, - assertSqlEnrichmentAuthorized, - collectResolvedMembers, - parseSqlTableReferences, - sqlEnrichmentBillingItems, - validateEnrichmentLease, - assertNoDirectEnrichmentObject, - removeEnrichmentSchema, -} from "../enrichmentEntitlement.js"; -import { filterUnentitledEnrichmentSchemas } from "../repositoryFactory.js"; - -const KEY = "test-enrichment-signing-key-at-least-32-bytes"; -const PRODUCTS = ["ctx:day-archetype", "ctx:weather-archetype"]; -const BILLING_CONNECTION_ID = "11111111-1111-4111-8111-111111111111"; - -const canonicalPayload = (payload) => - JSON.stringify({ - schema_version: payload.schema_version, - account_partition: payload.account_partition, - enabled: payload.enabled, - entitlement_revision: payload.entitlement_revision, - issued_at: payload.issued_at, - valid_until: payload.valid_until, - products: payload.products, - billing_connection_id: payload.billing_connection_id, - }); - -function makeSecurityContext(overrides = {}) { - const payload = { - schema_version: 1, - account_partition: "tenant-is", - enabled: true, - entitlement_revision: "7", - issued_at: "2026-08-30T09:00:00.000Z", - valid_until: "2026-08-30T11:00:00.000Z", - products: PRODUCTS, - billing_connection_id: BILLING_CONNECTION_ID, - ...overrides, - }; - const signature = createHmac("sha256", KEY) - .update(canonicalPayload(payload)) - .digest("base64url"); - return { - userScope: { - teamProperties: { - partition: payload.account_partition, - premium: { - enrichment: { - enabled: payload.enabled, - entitlement_revision: payload.entitlement_revision, - issued_at: payload.issued_at, - valid_until: payload.valid_until, - signature_version: "hmac-sha256-v1", - signature, - products: payload.products, - billing_connection_id: payload.billing_connection_id, - }, - }, - }, - }, - }; -} - -const NOW = new Date("2026-08-30T10:00:00.000Z"); - -describe("resolved enrichment member guard", () => { - it("walks dimensions, measures, segments, time dimensions, nested filters, and every order form", () => { - const members = collectResolvedMembers({ - dimensions: ["Orders.id"], - measures: ["Orders.count"], - segments: ["Orders.active"], - timeDimensions: [{ dimension: "Orders.createdAt" }], - filters: [ - { - and: [ - { - member: "CtxDayContext.dayType", - operator: "equals", - values: ["workday"], - }, - { or: [{ dimension: "CtxWeatherContext.weatherType" }] }, - ], - }, - ], - order: [ - ["CtxDayContext.date", "asc"], - { id: "CtxWeatherContext.temperature", desc: true }, - { member: "Orders.total", direction: "desc" }, - ], - }); - assert.deepEqual( - new Set(members), - new Set([ - "Orders.id", - "Orders.count", - "Orders.active", - "Orders.createdAt", - "CtxDayContext.dayType", - "CtxWeatherContext.weatherType", - "CtxDayContext.date", - "CtxWeatherContext.temperature", - "Orders.total", - ]), - ); - - assert.deepEqual( - collectResolvedMembers({ order: { "CtxDayContext.date": "asc" } }), - ["CtxDayContext.date"], - ); - }); - - it("accepts a current, correctly signed lease", () => { - assert.deepEqual( - validateEnrichmentLease(makeSecurityContext(), { - signingKey: KEY, - now: NOW, - }), - { valid: true, connectionId: BILLING_CONNECTION_ID }, - ); - assert.doesNotThrow(() => - assertEnrichmentQueryAuthorized( - { measures: ["CtxWeatherContext.temperatureAvg"] }, - makeSecurityContext(), - { signingKey: KEY, now: NOW }, - ), - ); - }); - - it("denies missing, expired, disabled, malformed, and incorrectly signed leases without member disclosure", () => { - const cases = [ - {}, - makeSecurityContext({ valid_until: "2026-08-30T09:59:59.000Z" }), - makeSecurityContext({ enabled: false }), - makeSecurityContext({ products: [PRODUCTS[0]] }), - makeSecurityContext({ billing_connection_id: null }), - makeSecurityContext(), - ]; - cases[5].userScope.teamProperties.premium.enrichment.signature = "invalid"; - - for (const securityContext of cases) { - assert.throws( - () => - assertEnrichmentQueryAuthorized( - { filters: [{ member: "CtxDayContext.secretMarker" }] }, - securityContext, - { signingKey: KEY, now: NOW }, - ), - (error) => { - assert.equal(error.status, 403); - assert.equal(error.code, "enrichment_not_available"); - assert.doesNotMatch(error.message, /Ctx|weather|day|marker/i); - return true; - }, - ); - } - }); - - it("does not require an enrichment lease for unrelated members", () => { - assert.doesNotThrow(() => - assertEnrichmentQueryAuthorized({ measures: ["Orders.count"] }, {}), - ); - }); -}); - -describe("physical SQL enrichment guard", () => { - it("parses qualified and quoted table references and ignores comments and string literals", () => { - const refs = parseSqlTableReferences(` - /* FROM enrich.release_pointer */ - WITH source AS ( - SELECT * FROM \"enrich\".\"day_context_v\" - ) - SELECT '-- JOIN enrich.weather_context' AS note - FROM source - JOIN \`enrich\`.\`weather_context_v\` AS weather ON 1 = 1 - `); - assert.deepEqual(refs, [ - { schema: "enrich", table: "day_context_v" }, - { schema: null, table: "source" }, - { schema: "enrich", table: "weather_context_v" }, - ]); - }); - - it("allows only public serving views with a valid lease", () => { - assert.doesNotThrow(() => - assertSqlEnrichmentAuthorized( - 'SELECT * FROM "enrich"."day_context_v" d JOIN enrich.weather_context_v w ON d.date = w.date', - makeSecurityContext(), - { signingKey: KEY, now: NOW }, - ), - ); - }); - - it("maps only explicit serving-view references to direct-SQL billing items", () => { - assert.deepEqual( - sqlEnrichmentBillingItems(` - SELECT '-- enrich.weather_context_v' AS note - FROM enrich.day_context_v - JOIN enrich.weather_context_v USING (local_date) - `), - ["ctx:day-archetype", "ctx:weather-archetype"], - ); - assert.deepEqual( - sqlEnrichmentBillingItems("SELECT * FROM cst.semantic_events"), - [], - ); - }); - - it("denies physical/control tables even when the lease is valid", () => { - for (const table of [ - "day_context", - "weather_context", - "release_manifest", - "release_pointer", - ]) { - assert.throws( - () => - assertSqlEnrichmentAuthorized( - `SELECT * FROM enrich.${table}`, - makeSecurityContext(), - { signingKey: KEY, now: NOW }, - ), - (error) => error.status === 403 && !error.message.includes(table), - ); - } - }); - - it("denies a serving view without a valid lease", () => { - assert.throws( - () => - assertSqlEnrichmentAuthorized("SELECT * FROM enrich.day_context_v", {}), - (error) => - error.status === 403 && error.code === "enrichment_not_available", - ); - }); -}); - -describe("metadata and helper isolation", () => { - it("removes only managed enrichment cubes and preserves authored siblings", () => { - const schemas = [ - { - name: "ctx_day_context.yml", - code: `cubes:\n - name: CtxDayContext\n meta:\n managed_by: ctx-enrichment\n - name: TeamAuthored\n sql_table: cst.events\n`, - }, - { name: "orders.yml", code: "cubes:\n - name: Orders\n" }, - ]; - const filtered = filterUnentitledEnrichmentSchemas(schemas, {}); - assert.equal(filtered.length, 2); - assert.doesNotMatch(filtered[0].code, /CtxDayContext|ctx-enrichment/); - assert.match(filtered[0].code, /TeamAuthored/); - assert.equal(filtered[1], schemas[1]); - }); - - it("keeps managed models when the lease is valid", () => { - const schemas = [{ name: "ctx_weather_context.yml", code: "cubes: []" }]; - assert.equal( - filterUnentitledEnrichmentSchemas(schemas, makeSecurityContext(), { - signingKey: KEY, - now: NOW, - }), - schemas, - ); - }); - - it("hides the enrich database and rejects direct helper object access", () => { - const schema = { cst: { events: [] }, enrich: { day_context_v: [] } }; - assert.deepEqual(removeEnrichmentSchema(schema), { cst: { events: [] } }); - assert.throws( - () => assertNoDirectEnrichmentObject("enrich", "day_context_v"), - (error) => error.status === 403 && !/day_context/.test(error.message), - ); - assert.doesNotThrow(() => assertNoDirectEnrichmentObject("cst", "events")); - }); -}); diff --git a/services/cubejs/src/utils/__tests__/enrichmentMetering.test.js b/services/cubejs/src/utils/__tests__/enrichmentMetering.test.js deleted file mode 100644 index 3a68f415..00000000 --- a/services/cubejs/src/utils/__tests__/enrichmentMetering.test.js +++ /dev/null @@ -1,317 +0,0 @@ -import assert from "node:assert/strict"; -import { createHmac } from "node:crypto"; -import { describe, it } from "node:test"; - -import { - collectResolvedMembers, - validateEnrichmentLease, -} from "../enrichmentEntitlement.js"; -import { - buildEnrichmentBillingBatch, - deterministicBillingMessageId, - installEnrichmentGatewayMetering, -} from "../enrichmentMetering.js"; -import { - parseEnrichmentCodePricing, - resolveEnrichmentCodePrice, -} from "../enrichmentPricing.js"; - -const KEY = "test-enrichment-signing-key-at-least-32-bytes"; -const NOW = new Date("2026-08-30T10:00:00.000Z"); -const BILLING_CONNECTION_ID = "11111111-1111-4111-8111-111111111111"; -const LEASE_PAYLOAD = { - schema_version: 1, - account_partition: "tenant.is", - enabled: true, - entitlement_revision: "7", - issued_at: "2026-08-30T09:00:00.000Z", - valid_until: "2026-08-30T11:00:00.000Z", - products: ["ctx:day-archetype", "ctx:weather-archetype"], - billing_connection_id: BILLING_CONNECTION_ID, -}; -const SECURITY_CONTEXT = { - userId: "person-1", - tokenPayload: { accountId: "account-real-1", partition: "tenant.is" }, - userScope: { - teamProperties: { - partition: "tenant.is", - premium: { - enrichment: { - enabled: LEASE_PAYLOAD.enabled, - entitlement_revision: LEASE_PAYLOAD.entitlement_revision, - issued_at: LEASE_PAYLOAD.issued_at, - valid_until: LEASE_PAYLOAD.valid_until, - products: LEASE_PAYLOAD.products, - billing_connection_id: LEASE_PAYLOAD.billing_connection_id, - signature_version: "hmac-sha256-v1", - signature: createHmac("sha256", KEY) - .update(JSON.stringify(LEASE_PAYLOAD)) - .digest("base64url"), - }, - }, - }, - }, -}; -const PRICING = Object.freeze({ - pricingCodeVersion: "ctx-pricing-v1", - items: Object.freeze({ - "ctx:day-archetype": Object.freeze({ - amount: "0.10", - currency: "ISK", - }), - "ctx:weather-archetype": Object.freeze({ - amount: "0.25", - currency: "ISK", - }), - }), -}); -const ITEM_BY_CUBE = new Map([ - ["CtxDayContext", "ctx:day-archetype"], - ["CtxWeatherContext", "ctx:weather-archetype"], -]); - -const resolveItems = async (query) => [ - ...new Set( - collectResolvedMembers(query) - .map((member) => ITEM_BY_CUBE.get(member.split(".", 1)[0])) - .filter(Boolean), - ), -]; -const resolvePrice = (item) => - resolveEnrichmentCodePrice(item, { config: PRICING }); -const request = (query, overrides = {}) => ({ - query, - apiType: "rest", - context: { - requestId: "logical-1", - securityContext: SECURITY_CONTEXT, - }, - ...overrides, -}); -const response = (data = [], extra = {}) => ({ data, ...extra }); -const build = (req, result, options = {}) => - buildEnrichmentBillingBatch(req, result, { - resolveItems, - resolvePrice, - validateLease: (securityContext) => - validateEnrichmentLease(securityContext, { signingKey: KEY, now: NOW }), - ...options, - }); - -describe("enrichment result-commit metering", () => { - it("detects logical members in filters and every order shape, not SQL text", async () => { - const result = await build( - request({ - dimensions: ["Events.id"], - filters: [ - { - and: [ - { - member: "CtxDayContext.isHoliday", - operator: "equals", - values: ["1"], - }, - ], - }, - ], - order: { "CtxDayContext.localDate": "asc" }, - generatedSql: "SELECT * FROM enrich.weather_context_v", - }), - response([{ "Events.id": "1" }]), - ); - assert.equal(result.length, 1); - assert.equal(result[0].envelope.dimensions.item, "ctx:day-archetype"); - }); - - it("deduplicates members from one item and charges both distinct items", async () => { - const result = await build( - request({ - dimensions: [ - "CtxDayContext.isHoliday", - "CtxDayContext.seasonName", - "CtxWeatherContext.temperature", - ], - }), - response([]), - ); - assert.deepEqual( - result.map((entry) => entry.envelope.dimensions.item), - ["ctx:day-archetype", "ctx:weather-archetype"], - ); - }); - - it("charges successful cache hits and zero-row results", async () => { - const [entry] = await build( - request({ dimensions: ["CtxWeatherContext.temperature"] }), - response([], { cacheStatus: "hit" }), - ); - assert.equal(entry.envelope.properties.returned_rows, 0); - assert.equal(entry.envelope.properties.cache_status, "cache_hit"); - assert.equal(entry.envelope.metrics.record_count, 1); - }); - - it("uses stable retry identities and child ids for multi-query requests", async () => { - const query = { dimensions: ["CtxDayContext.isHoliday"] }; - const first = await build(request(query), response([])); - const retry = await build(request(query), response([])); - assert.equal(first[0].idempotencyKey, retry[0].idempotencyKey); - assert.equal( - first[0].envelope.message_id, - deterministicBillingMessageId("logical-1", "ctx:day-archetype"), - ); - - const children = await build(request([query, query]), { - results: [response([]), response([{ value: 1 }])], - }); - assert.deepEqual( - children.map((entry) => entry.envelope.properties.logical_execution_id), - ["logical-1:query:0", "logical-1:query:1"], - ); - assert.deepEqual( - children.map((entry) => entry.envelope.properties.returned_rows), - [0, 1], - ); - }); - - it("does not charge errors, disconnects, non-enrichment, scheduled refreshes, or pre-aggregations", async () => { - const enrichment = { dimensions: ["CtxDayContext.isHoliday"] }; - assert.deepEqual( - await build(request(enrichment), { error: "failure" }), - [], - ); - assert.deepEqual( - await build( - request(enrichment, { signal: { aborted: true } }), - response([]), - ), - [], - ); - assert.deepEqual( - await build(request({ dimensions: ["Events.id"] }), response([])), - [], - ); - assert.deepEqual( - await build( - request(enrichment, { scheduledRefresh: true }), - response([]), - ), - [], - ); - assert.deepEqual( - await build(request(enrichment, { preAggregation: true }), response([])), - [], - ); - }); - - it("requires real account/connection and never falls back to partition", async () => { - const req = request({ dimensions: ["CtxDayContext.isHoliday"] }); - req.context.securityContext = { - ...SECURITY_CONTEXT, - tokenPayload: { accountId: null, partition: "tenant.is" }, - }; - await assert.rejects(build(req, response([])), /real Account/); - - const missingConnection = structuredClone(SECURITY_CONTEXT); - missingConnection.userScope.teamProperties.premium.enrichment.billing_connection_id = null; - const missingConnectionRequest = request({ - dimensions: ["CtxDayContext.isHoliday"], - }); - missingConnectionRequest.context.securityContext = missingConnection; - await assert.rejects( - build(missingConnectionRequest, response([])), - /per-account enrichment billing Connection/, - ); - }); - - it("does not commit a result when durable enqueue fails", async () => { - let responseCommitted = false; - const gateway = { - load: async (req) => req.res(response([])), - sqlApiLoad: async (req) => req.res(response([])), - }; - installEnrichmentGatewayMetering( - { apiGateway: () => gateway }, - {}, - { - commit: async () => { - throw new Error("redis unavailable"); - }, - }, - ); - await assert.rejects( - gateway.load({ - ...request({ dimensions: ["CtxDayContext.isHoliday"] }), - res: async () => { - responseCommitted = true; - }, - }), - /redis unavailable/, - ); - assert.equal(responseCommitted, false); - }); - - it("commits billing before REST and SQL API responses", async () => { - const order = []; - const gateway = { - load: async (req) => req.res(response([])), - sqlApiLoad: async (req) => req.res(response([])), - }; - installEnrichmentGatewayMetering( - { apiGateway: () => gateway }, - {}, - { - commit: async (_redis, _request, _message, options) => - order.push(options.surface), - }, - ); - const base = { - ...request({ dimensions: ["CtxDayContext.isHoliday"] }), - res: async () => order.push("response"), - }; - await gateway.load(base); - await gateway.sqlApiLoad(base); - assert.deepEqual(order, ["rest", "response", "sql-api", "response"]); - }); - - it("uses legacy_runtime_rate without requiring or querying Convex pricing", () => { - const parsed = parseEnrichmentCodePricing( - JSON.stringify({ - pricing_code_version: "ctx-pricing-v1", - items: { - "ctx:day-archetype": { - amount: "0.10", - currency: "isk", - }, - }, - }), - ); - const options = { config: parsed }; - Object.defineProperty(options, "convexPricing", { - get() { - throw new Error("Convex pricing must not be queried"); - }, - }); - const price = resolveEnrichmentCodePrice("ctx:day-archetype", options); - assert.equal(price.pricingResolution.pricingSource, "legacy_runtime_rate"); - assert.equal(price.pricingCodeVersion, "ctx-pricing-v1"); - assert.equal(price.unitAmount, "0.10"); - assert.equal("connectionId" in price, false); - }); - - it("attributes every enrichment item to the Connection in the signed account lease", async () => { - const result = await build( - request({ - dimensions: [ - "CtxDayContext.isHoliday", - "CtxWeatherContext.temperature", - ], - }), - response([]), - ); - assert.equal(result.length, 2); - assert.deepEqual( - result.map((entry) => entry.envelope.properties.connection_id), - [BILLING_CONNECTION_ID, BILLING_CONNECTION_ID], - ); - }); -}); diff --git a/services/cubejs/src/utils/__tests__/legacyEnrichmentGuard.test.js b/services/cubejs/src/utils/__tests__/legacyEnrichmentGuard.test.js new file mode 100644 index 00000000..8d4b39dd --- /dev/null +++ b/services/cubejs/src/utils/__tests__/legacyEnrichmentGuard.test.js @@ -0,0 +1,157 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { + assertNoDirectLegacyEnrichmentObject, + assertNoLegacyEnrichmentQuery, + assertNoLegacyEnrichmentSql, + collectResolvedMembers, + parseSqlTableReferences, + removeLegacyEnrichmentSchema, +} from "../legacyEnrichmentGuard.js"; +import { filterLegacyEnrichmentSchemas } from "../repositoryFactory.js"; + +describe("resolved enrichment member guard", () => { + it("walks dimensions, measures, segments, time dimensions, nested filters, and every order form", () => { + const members = collectResolvedMembers({ + dimensions: ["Orders.id"], + measures: ["Orders.count"], + segments: ["Orders.active"], + timeDimensions: [{ dimension: "Orders.createdAt" }], + filters: [ + { + and: [ + { + member: "CtxDayContext.dayType", + operator: "equals", + values: ["workday"], + }, + { or: [{ dimension: "CtxWeatherContext.weatherType" }] }, + ], + }, + ], + order: [ + ["CtxDayContext.date", "asc"], + { id: "CtxWeatherContext.temperature", desc: true }, + { member: "Orders.total", direction: "desc" }, + ], + }); + assert.deepEqual( + new Set(members), + new Set([ + "Orders.id", + "Orders.count", + "Orders.active", + "Orders.createdAt", + "CtxDayContext.dayType", + "CtxWeatherContext.weatherType", + "CtxDayContext.date", + "CtxWeatherContext.temperature", + "Orders.total", + ]), + ); + + assert.deepEqual( + collectResolvedMembers({ order: { "CtxDayContext.date": "asc" } }), + ["CtxDayContext.date"], + ); + }); + + it("always denies retired cubes without member disclosure", () => { + assert.throws( + () => + assertNoLegacyEnrichmentQuery({ + filters: [{ member: "CtxDayContext.secretMarker" }], + }), + (error) => { + assert.equal(error.status, 403); + assert.equal(error.code, "enrichment_not_available"); + assert.doesNotMatch(error.message, /Ctx|weather|day|marker/i); + return true; + }, + ); + }); + + it("allows the ordinary semantic query surface", () => { + assert.doesNotThrow(() => + assertNoLegacyEnrichmentQuery({ measures: ["FftWeather.temperatureAvg"] }), + ); + }); +}); + +describe("physical SQL enrichment guard", () => { + it("parses qualified and quoted table references and ignores comments and string literals", () => { + const refs = parseSqlTableReferences(` + /* FROM enrich.release_pointer */ + WITH source AS ( + SELECT * FROM \"enrich\".\"day_context_v\" + ) + SELECT '-- JOIN enrich.weather_context' AS note + FROM source + JOIN \`enrich\`.\`weather_context_v\` AS weather ON 1 = 1 + `); + assert.deepEqual(refs, [ + { schema: "enrich", table: "day_context_v" }, + { schema: null, table: "source" }, + { schema: "enrich", table: "weather_context_v" }, + ]); + }); + + it("denies every retired physical object", () => { + for (const table of [ + "day_context", + "weather_context", + "release_manifest", + "release_pointer", + ]) { + assert.throws( + () => + assertNoLegacyEnrichmentSql(`SELECT * FROM enrich.${table}`), + (error) => error.status === 403 && !error.message.includes(table), + ); + } + }); + + it("allows the current FFT physical query surface", () => { + assert.doesNotThrow(() => + assertNoLegacyEnrichmentSql("SELECT * FROM public.fft_weather"), + ); + assert.throws( + () => assertNoLegacyEnrichmentSql("SELECT * FROM enrich.day_context_v"), + (error) => + error.status === 403 && error.code === "enrichment_not_available", + ); + }); +}); + +describe("metadata and helper isolation", () => { + it("removes only managed enrichment cubes and preserves authored siblings", () => { + const schemas = [ + { + name: "ctx_day_context.yml", + code: `cubes:\n - name: CtxDayContext\n meta:\n managed_by: ctx-enrichment\n - name: TeamAuthored\n sql_table: cst.events\n`, + }, + { name: "orders.yml", code: "cubes:\n - name: Orders\n" }, + ]; + const filtered = filterLegacyEnrichmentSchemas(schemas); + assert.equal(filtered.length, 2); + assert.doesNotMatch(filtered[0].code, /CtxDayContext|ctx-enrichment/); + assert.match(filtered[0].code, /TeamAuthored/); + assert.equal(filtered[1], schemas[1]); + }); + + it("hides the enrich database and rejects direct helper object access", () => { + const schema = { cst: { events: [] }, enrich: { day_context_v: [] } }; + assert.deepEqual(removeLegacyEnrichmentSchema(schema), { + cst: { events: [] }, + }); + assert.throws( + () => + assertNoDirectLegacyEnrichmentObject("enrich", "day_context_v"), + (error) => error.status === 403 && !/day_context/.test(error.message), + ); + assert.doesNotThrow(() => + assertNoDirectLegacyEnrichmentObject("cst", "events"), + ); + }); +}); diff --git a/services/cubejs/src/utils/billingMetrics.js b/services/cubejs/src/utils/billingMetrics.js deleted file mode 100644 index d495f45f..00000000 --- a/services/cubejs/src/utils/billingMetrics.js +++ /dev/null @@ -1,164 +0,0 @@ -const COUNTER_NAMES = Object.freeze([ - "emitted", - "deduplicated", - "delivered", - "retried", - "replayed", - "enqueue_failures", - "dead_lettered", - "unknown_pricing", - "unmeterable", -]); - -const counters = Object.fromEntries(COUNTER_NAMES.map((name) => [name, 0])); - -export function incrementBillingMetric(name, count = 1) { - if (!Object.hasOwn(counters, name)) { - throw new Error(`unknown billing metric: ${name}`); - } - const value = Number(count); - if (!Number.isInteger(value) || value < 0) { - throw new Error("billing metric increments must be non-negative integers"); - } - counters[name] += value; -} - -export function billingMetricsSnapshot() { - return { ...counters }; -} - -export function resetBillingMetricsForTest() { - for (const name of COUNTER_NAMES) counters[name] = 0; -} - -export function recordBillingFailure(error) { - const message = String(error?.message || error || ""); - if (/price|pricing|amount|currency|legacy_runtime_rate/i.test(message)) { - incrementBillingMetric("unknown_pricing"); - } else { - incrementBillingMetric("unmeterable"); - } -} - -export async function readBillingOutboxGauges( - redis, - { nowMs = Date.now() } = {}, -) { - if (!redis) throw new Error("billing Redis is unavailable"); - const [dlq, pendingSummary, groups] = await Promise.all([ - redis.xlen("streams:synmetrix-billing-dlq"), - redis.xpending( - "streams:synmetrix-billing-outbox", - "synmetrix-billing-delivery", - ), - redis.xinfo("GROUPS", "streams:synmetrix-billing-outbox"), - ]); - const group = (groups || []).find((fields) => { - const values = Object.fromEntries( - Array.from({ length: Math.floor(fields.length / 2) }, (_, index) => [ - fields[index * 2], - fields[index * 2 + 1], - ]), - ); - return values.name === "synmetrix-billing-delivery"; - }); - if (!group) throw new Error("billing consumer group is unavailable"); - const groupValues = Object.fromEntries( - Array.from({ length: Math.floor(group.length / 2) }, (_, index) => [ - group[index * 2], - group[index * 2 + 1], - ]), - ); - const pending = Number(pendingSummary?.[0] || 0); - const lag = Number(groupValues.lag || 0); - const pendingOldestId = pending ? String(pendingSummary?.[1] || "") : ""; - let lagOldestId = ""; - if (lag > 0) { - const rows = await redis.xrange( - "streams:synmetrix-billing-outbox", - `(${groupValues["last-delivered-id"] || "0-0"}`, - "+", - "COUNT", - 1, - ); - lagOldestId = String(rows?.[0]?.[0] || ""); - } - const ages = [pendingOldestId, lagOldestId] - .filter(Boolean) - .map((id) => Number(id.split("-", 1)[0])) - .filter(Number.isFinite) - .map((timestamp) => Math.max(0, (nowMs - timestamp) / 1000)); - return { - backlog: pending + lag, - pending, - dlq: Number(dlq || 0), - oldestUndeliveredAgeSeconds: ages.length ? Math.max(...ages) : 0, - }; -} - -const metricLine = (name, help, type, value) => - `# HELP ${name} ${help}\n# TYPE ${name} ${type}\n${name} ${value}`; - -export function renderBillingMetrics( - snapshot, - gauges, - { workerUp = false } = {}, -) { - const lines = COUNTER_NAMES.map((name) => - metricLine( - `synmetrix_billing_${name}_total`, - `Bounded billing ${name.replaceAll("_", " ")} count.`, - "counter", - Number(snapshot[name] || 0), - ), - ); - lines.push( - metricLine( - "synmetrix_billing_outbox_backlog_entries", - "Undelivered billing entries (consumer-group lag plus pending).", - "gauge", - gauges.backlog, - ), - metricLine( - "synmetrix_billing_outbox_pending_entries", - "Entries currently pending in the billing consumer group.", - "gauge", - gauges.pending, - ), - metricLine( - "synmetrix_billing_outbox_dlq_entries", - "Entries currently retained in the billing dead-letter stream.", - "gauge", - gauges.dlq, - ), - metricLine( - "synmetrix_billing_outbox_oldest_undelivered_age_seconds", - "Age of the oldest lagging or pending billing entry.", - "gauge", - gauges.oldestUndeliveredAgeSeconds, - ), - metricLine( - "synmetrix_billing_outbox_worker_up", - "Whether this Cube process has a running billing delivery worker.", - "gauge", - workerUp ? 1 : 0, - ), - ); - return `${lines.join("\n")}\n`; -} - -export function createBillingMetricsHandler({ redis, worker }) { - return async (_req, res) => { - try { - const gauges = await readBillingOutboxGauges(redis); - res.type("text/plain; version=0.0.4"); - return res.send( - renderBillingMetrics(billingMetricsSnapshot(), gauges, { - workerUp: Boolean(worker?.running), - }), - ); - } catch { - return res.status(503).type("text/plain").send("billing metrics unavailable\n"); - } - }; -} diff --git a/services/cubejs/src/utils/billingOutbox.js b/services/cubejs/src/utils/billingOutbox.js deleted file mode 100644 index d462ce4a..00000000 --- a/services/cubejs/src/utils/billingOutbox.js +++ /dev/null @@ -1,194 +0,0 @@ -import { incrementBillingMetric } from "./billingMetrics.js"; - -export const BILLING_STREAM = "streams:synmetrix-billing-outbox"; -export const BILLING_DLQ_STREAM = "streams:synmetrix-billing-dlq"; -export const BILLING_GROUP = "synmetrix-billing-delivery"; -export const BILLING_ATTEMPTS_HASH = "synmetrix-billing-attempts"; -const DEDUPE_PREFIX = "synmetrix-billing-dedupe:"; -const DEFAULT_MAX_LENGTH = 100_000; -const DEFAULT_DEDUPE_TTL_SECONDS = 400 * 24 * 60 * 60; - -const ENQUEUE_SCRIPT = ` -local existing = redis.call('GET', KEYS[2]) -if existing then return 'duplicate' end -local stream_id = redis.call( - 'XADD', KEYS[1], 'MAXLEN', '~', ARGV[4], '*', - 'envelope', ARGV[1], 'context', ARGV[2], 'idempotency_key', ARGV[3] -) -redis.call('SET', KEYS[2], stream_id, 'EX', ARGV[5]) -return stream_id -`; - -const ENQUEUE_BATCH_SCRIPT = ` -local count = tonumber(ARGV[1]) -local max_length = ARGV[2] -local ttl = ARGV[3] -local result = {} - -for index = 1, count do - local key_index = index + 1 - if redis.call('GET', KEYS[key_index]) then - result[index] = 'duplicate' - end -end - -for index = 1, count do - if not result[index] then - local offset = 3 + ((index - 1) * 3) - local envelope = ARGV[offset + 1] - local context = ARGV[offset + 2] - local idempotency_key = ARGV[offset + 3] - local stream_id = redis.call( - 'XADD', KEYS[1], 'MAXLEN', '~', max_length, '*', - 'envelope', envelope, 'context', context, - 'idempotency_key', idempotency_key - ) - redis.call('SET', KEYS[index + 1], stream_id, 'EX', ttl) - result[index] = stream_id - end -end -return result -`; - -export async function enqueueBillingEvent( - redis, - envelope, - context, - { - idempotencyKey = envelope?.message_id, - maxLength = DEFAULT_MAX_LENGTH, - dedupeTtlSeconds = DEFAULT_DEDUPE_TTL_SECONDS, - } = {}, -) { - if (!redis || !envelope || !String(idempotencyKey || "").trim()) { - throw new Error( - "billing outbox requires Redis, an envelope, and an idempotency key", - ); - } - const key = String(idempotencyKey); - let result; - try { - result = await redis.eval( - ENQUEUE_SCRIPT, - 2, - BILLING_STREAM, - `${DEDUPE_PREFIX}${key}`, - JSON.stringify(envelope), - JSON.stringify(context || {}), - key, - String(maxLength), - String(dedupeTtlSeconds), - ); - } catch (error) { - incrementBillingMetric("enqueue_failures"); - throw error; - } - incrementBillingMetric(result === "duplicate" ? "deduplicated" : "emitted"); - return result === "duplicate" - ? { enqueued: false, duplicate: true } - : { enqueued: true, streamId: result }; -} - -/** - * Atomically enqueue every charge for one logical result commit. This avoids - * returning a result after only one item from a multi-item query was recorded. - */ -export async function enqueueBillingBatch( - redis, - entries, - { - maxLength = DEFAULT_MAX_LENGTH, - dedupeTtlSeconds = DEFAULT_DEDUPE_TTL_SECONDS, - } = {}, -) { - if (!redis || !Array.isArray(entries) || entries.length === 0) { - throw new Error("billing batch requires Redis and at least one entry"); - } - - const normalized = entries.map(({ envelope, context, idempotencyKey }) => { - const key = String(idempotencyKey || envelope?.message_id || "").trim(); - if (!envelope || !key) { - throw new Error( - "billing batch entry requires an envelope and idempotency key", - ); - } - return { envelope, context: context || {}, key }; - }); - const keys = [ - BILLING_STREAM, - ...normalized.map(({ key }) => `${DEDUPE_PREFIX}${key}`), - ]; - const args = [ - String(normalized.length), - String(maxLength), - String(dedupeTtlSeconds), - ...normalized.flatMap(({ envelope, context, key }) => [ - JSON.stringify(envelope), - JSON.stringify(context), - key, - ]), - ]; - let results; - try { - results = await redis.eval( - ENQUEUE_BATCH_SCRIPT, - keys.length, - ...keys, - ...args, - ); - } catch (error) { - incrementBillingMetric("enqueue_failures"); - throw error; - } - - incrementBillingMetric( - "deduplicated", - results.filter((value) => value === "duplicate").length, - ); - incrementBillingMetric( - "emitted", - results.filter((value) => value !== "duplicate").length, - ); - - return normalized.map(({ key }, index) => { - const value = results?.[index]; - return value === "duplicate" - ? { idempotencyKey: key, enqueued: false, duplicate: true } - : { idempotencyKey: key, enqueued: true, streamId: value }; - }); -} - -export async function replayBillingDlqEntry(redis, dlqId, entry) { - const replayKey = `${entry?.envelope?.message_id}:replay:${dlqId}`; - const result = await enqueueBillingEvent( - redis, - entry?.envelope, - entry?.context, - { - idempotencyKey: replayKey, - }, - ); - if (result.enqueued) { - await redis.xdel(BILLING_DLQ_STREAM, dlqId); - incrementBillingMetric("replayed"); - } - return result; -} - -function fieldsToObject(fields) { - const result = {}; - for (let index = 0; index < (fields || []).length; index += 2) { - result[fields[index]] = fields[index + 1]; - } - return result; -} - -export async function loadAndReplayBillingDlqEntry(redis, dlqId) { - const rows = await redis.xrange(BILLING_DLQ_STREAM, dlqId, dlqId); - if (!rows?.length) throw new Error("billing DLQ entry not found"); - const fields = fieldsToObject(rows[0][1]); - return replayBillingDlqEntry(redis, dlqId, { - envelope: JSON.parse(fields.envelope), - context: JSON.parse(fields.context || "{}"), - }); -} diff --git a/services/cubejs/src/utils/billingOutboxReplay.js b/services/cubejs/src/utils/billingOutboxReplay.js deleted file mode 100644 index ed3d48e5..00000000 --- a/services/cubejs/src/utils/billingOutboxReplay.js +++ /dev/null @@ -1,19 +0,0 @@ -import Redis from "ioredis"; - -import { loadAndReplayBillingDlqEntry } from "./billingOutbox.js"; - -const dlqId = process.argv[2]; -if (!dlqId || !process.env.REDIS_ADDR) { - process.stderr.write( - "Usage: REDIS_ADDR=redis://... node src/utils/billingOutboxReplay.js \n", - ); - process.exitCode = 2; -} else { - const redis = new Redis(process.env.REDIS_ADDR); - try { - const result = await loadAndReplayBillingDlqEntry(redis, dlqId); - process.stdout.write(`${JSON.stringify(result)}\n`); - } finally { - await redis.quit(); - } -} diff --git a/services/cubejs/src/utils/billingOutboxWorker.js b/services/cubejs/src/utils/billingOutboxWorker.js deleted file mode 100644 index 7f4fca99..00000000 --- a/services/cubejs/src/utils/billingOutboxWorker.js +++ /dev/null @@ -1,174 +0,0 @@ -import { hostname } from "node:os"; - -import { emitSemanticEvent } from "./eventEmitter.js"; -import { incrementBillingMetric } from "./billingMetrics.js"; -import { - BILLING_ATTEMPTS_HASH, - BILLING_DLQ_STREAM, - BILLING_GROUP, - BILLING_STREAM, -} from "./billingOutbox.js"; - -function fieldsToObject(fields) { - const result = {}; - for (let index = 0; index < (fields || []).length; index += 2) { - result[fields[index]] = fields[index + 1]; - } - return result; -} - -export async function processBillingEntry( - redis, - [streamId, fields], - { - send = emitSemanticEvent, - maxAttempts = Number(process.env.BILLING_OUTBOX_MAX_ATTEMPTS || 8), - } = {}, -) { - const data = fieldsToObject(fields); - let envelope; - let context; - try { - envelope = JSON.parse(data.envelope); - context = JSON.parse(data.context || "{}"); - } catch { - envelope = null; - context = {}; - } - - let delivered = false; - let failure = "invalid_outbox_payload"; - if (envelope) { - try { - const result = await send(envelope, context); - delivered = result?.ok === true; - if (!delivered) failure = "ingress_rejected"; - } catch (error) { - failure = error?.message || "ingress_error"; - } - } - - if (delivered) { - await redis.xack(BILLING_STREAM, BILLING_GROUP, streamId); - await redis.hdel(BILLING_ATTEMPTS_HASH, streamId); - incrementBillingMetric("delivered"); - return "acknowledged"; - } - - const attempts = Number( - await redis.hincrby(BILLING_ATTEMPTS_HASH, streamId, 1), - ); - if (attempts < maxAttempts) { - incrementBillingMetric("retried"); - return "pending"; - } - - await redis.xadd( - BILLING_DLQ_STREAM, - "*", - "envelope", - data.envelope || "null", - "context", - data.context || "{}", - "source_id", - streamId, - "attempts", - String(attempts), - "failure", - String(failure).slice(0, 512), - ); - await redis.xack(BILLING_STREAM, BILLING_GROUP, streamId); - await redis.hdel(BILLING_ATTEMPTS_HASH, streamId); - incrementBillingMetric("dead_lettered"); - return "dead_lettered"; -} - -export class BillingOutboxWorker { - constructor( - redis, - { - consumer = `${hostname()}-${process.pid}`, - blockMs = 5_000, - reclaimIdleMs = 30_000, - batchSize = 25, - send = emitSemanticEvent, - } = {}, - ) { - this.redis = redis; - this.consumer = consumer; - this.blockMs = blockMs; - this.reclaimIdleMs = reclaimIdleMs; - this.batchSize = batchSize; - this.send = send; - this.running = false; - this.loopPromise = null; - } - - async ensureGroup() { - try { - await this.redis.xgroup( - "CREATE", - BILLING_STREAM, - BILLING_GROUP, - "0", - "MKSTREAM", - ); - } catch (error) { - if (!String(error?.message || error).includes("BUSYGROUP")) throw error; - } - } - - async process(entries) { - for (const entry of entries || []) { - await processBillingEntry(this.redis, entry, { send: this.send }); - } - } - - async reclaim() { - const result = await this.redis.xautoclaim( - BILLING_STREAM, - BILLING_GROUP, - this.consumer, - this.reclaimIdleMs, - "0-0", - "COUNT", - this.batchSize, - ); - await this.process(result?.[1] || []); - } - - async loop() { - await this.ensureGroup(); - await this.reclaim(); - while (this.running) { - const response = await this.redis.xreadgroup( - "GROUP", - BILLING_GROUP, - this.consumer, - "COUNT", - this.batchSize, - "BLOCK", - this.blockMs, - "STREAMS", - BILLING_STREAM, - ">", - ); - await this.process(response?.[0]?.[1] || []); - } - } - - start() { - if (this.running) return this.loopPromise; - this.running = true; - this.loopPromise = this.loop().catch((error) => { - this.running = false; - console.error("billing outbox worker stopped", error); - }); - return this.loopPromise; - } - - async stop() { - this.running = false; - await this.loopPromise; - } -} diff --git a/services/cubejs/src/utils/enrichmentMetering.js b/services/cubejs/src/utils/enrichmentMetering.js deleted file mode 100644 index f8efc6d0..00000000 --- a/services/cubejs/src/utils/enrichmentMetering.js +++ /dev/null @@ -1,273 +0,0 @@ -import { createHash, randomUUID } from "node:crypto"; - -import { enqueueBillingBatch } from "./billingOutbox.js"; -import { recordBillingFailure } from "./billingMetrics.js"; -import { buildConnectionCalled } from "./eventEmitter.js"; -import { resolveEnrichmentCodePrice } from "./enrichmentPricing.js"; -import { resolveEnrichmentBillingItems } from "./queryRewrite.js"; -import { - sqlEnrichmentBillingItems, - validateEnrichmentLease, -} from "./enrichmentEntitlement.js"; - -function billingError(message) { - const error = new Error(`enrichment billing unavailable: ${message}`); - error.status = 503; - error.code = "enrichment_billing_unavailable"; - return error; -} - -export function deterministicBillingMessageId(logicalExecutionId, item) { - const bytes = Buffer.from( - createHash("sha256") - .update(`${logicalExecutionId}\u0000${item}`) - .digest() - .subarray(0, 16), - ); - bytes[6] = (bytes[6] & 0x0f) | 0x50; - bytes[8] = (bytes[8] & 0x3f) | 0x80; - const hex = bytes.toString("hex"); - return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`; -} - -export function logicalExecutionIdForRequest(request = {}, childIndex = null) { - const context = request.context || {}; - const headers = request.headers || {}; - const base = String( - request.logicalExecutionId || - headers["x-idempotency-key"] || - headers["x-request-id"] || - context.requestId || - randomUUID(), - ); - return childIndex == null ? base : `${base}:query:${childIndex}`; -} - -function rootResults(message) { - if (message?.isWrapper && typeof message.getRootResultObject === "function") { - const root = message.getRootResultObject(); - return Array.isArray(root) ? root : [root]; - } - if (Array.isArray(message?.results)) return message.results; - return [message || {}]; -} - -function returnedRowsFor(message, index) { - const result = rootResults(message)[index] || {}; - return Array.isArray(result?.data) ? result.data.length : 0; -} - -function cacheStatusFor(message, index, fallback = "unknown") { - const result = rootResults(message)[index] || {}; - const value = result.cacheStatus || result.cache_status; - return value === "hit" || value === "cache_hit" - ? "cache_hit" - : value === "miss" || value === "cache_miss" - ? "cache_miss" - : fallback; -} - -function queriesFor(request) { - const query = request?.query; - if (!query) return []; - return Array.isArray(query) ? query : [query]; -} - -export function isSystemEnrichmentExecution(request = {}) { - const context = request.context || {}; - return Boolean( - request.systemOperation || - request.preAggregation || - request.scheduledRefresh || - context.systemOperation || - context.preAggregation || - context.scheduledRefresh || - context.requestId?.includes("scheduler"), - ); -} - -export function isCancelledEnrichmentExecution(request = {}) { - const context = request.context || {}; - return Boolean( - request.aborted || - request.cancelled || - request.signal?.aborted || - context.aborted || - context.cancelled || - context.signal?.aborted, - ); -} - -export async function buildEnrichmentBillingBatch( - request, - message, - { - resolveItems = resolveEnrichmentBillingItems, - resolvePrice = resolveEnrichmentCodePrice, - validateLease = validateEnrichmentLease, - surface = request?.apiType === "sql" ? "sql-api" : "rest", - cacheStatus = "unknown", - returnedRows = null, - } = {}, -) { - if ( - isSystemEnrichmentExecution(request) || - isCancelledEnrichmentExecution(request) - ) { - return []; - } - const queries = queriesFor(request); - if (queries.length === 0 || message?.error) return []; - - const securityContext = request.context?.securityContext || {}; - const token = securityContext.tokenPayload || {}; - const accountId = String(token.accountId || "").trim(); - const partition = String( - token.partition || - securityContext.userScope?.teamProperties?.partition || - "", - ).trim(); - let billingConnectionId = null; - - const entries = []; - for (let index = 0; index < queries.length; index += 1) { - const items = await resolveItems(queries[index], securityContext); - if (!items.length) continue; - if (!accountId || !partition) { - throw billingError("real Account and tenant partition are required"); - } - if (!billingConnectionId) { - const entitlement = validateLease(securityContext); - if (!entitlement?.valid || !entitlement.connectionId) { - throw billingError( - "a valid per-account enrichment billing Connection is required", - ); - } - billingConnectionId = entitlement.connectionId; - } - const logicalExecutionId = logicalExecutionIdForRequest( - request, - queries.length > 1 ? index : null, - ); - for (const item of [...new Set(items)].sort()) { - const price = resolvePrice(item); - const messageId = deterministicBillingMessageId(logicalExecutionId, item); - const envelope = buildConnectionCalled({ - partition, - accountId, - userId: securityContext.userId || null, - provider: "ctx", - item, - connectionId: billingConnectionId, - billingMode: true, - messageId, - logicalExecutionId, - surface, - accountingScope: "customer_usage", - cacheStatus: cacheStatusFor(message, index, cacheStatus), - returnedRows: - returnedRows == null - ? returnedRowsFor(message, index) - : Number(returnedRows), - recordCount: 1, - unitAmount: price.unitAmount, - pricingCodeVersion: price.pricingCodeVersion, - pricingResolution: price.pricingResolution, - }); - entries.push({ - envelope, - context: { - accountId, - partition, - userId: securityContext.userId || null, - }, - idempotencyKey: `${logicalExecutionId}:${item}`, - }); - } - } - return entries; -} - -export async function commitSqlEnrichmentBilling( - redis, - { - sql, - securityContext, - logicalExecutionId, - surface, - returnedRows, - cacheStatus = "unknown", - signal = null, - }, - options = {}, -) { - const items = sqlEnrichmentBillingItems(sql); - if (!items.length) return []; - return commitEnrichmentBilling( - redis, - { - query: {}, - logicalExecutionId, - signal, - context: { securityContext }, - }, - { data: [] }, - { - ...options, - resolveItems: async () => items, - surface, - returnedRows, - cacheStatus, - }, - ); -} - -export async function commitEnrichmentBilling( - redis, - request, - message, - options, -) { - try { - const entries = await buildEnrichmentBillingBatch(request, message, options); - if (!entries.length) return []; - if (!redis) throw billingError("durable Redis outbox is required"); - return await enqueueBillingBatch(redis, entries); - } catch (error) { - recordBillingFailure(error); - throw error; - } -} - -/** - * Cube awaits this result callback after query execution and before writing the - * HTTP/SQL response. Wrapping it gives billing a real result-commit boundary: - * an enqueue failure becomes a gateway error and no successful result is sent. - */ -export function installEnrichmentGatewayMetering( - cubejs, - redis, - { commit = commitEnrichmentBilling } = {}, -) { - const gateway = cubejs.apiGateway(); - if (gateway.__enrichmentMeteringInstalled) return gateway; - - for (const methodName of ["load", "sqlApiLoad"]) { - if (typeof gateway[methodName] !== "function") continue; - const original = gateway[methodName].bind(gateway); - gateway[methodName] = async (request) => { - const originalResponse = request.res; - return original({ - ...request, - res: async (message, responseOptions) => { - await commit(redis, request, message, { - surface: methodName === "sqlApiLoad" ? "sql-api" : "rest", - }); - return originalResponse(message, responseOptions); - }, - }); - }; - } - gateway.__enrichmentMeteringInstalled = true; - return gateway; -} diff --git a/services/cubejs/src/utils/enrichmentPricing.js b/services/cubejs/src/utils/enrichmentPricing.js deleted file mode 100644 index 9be9fb76..00000000 --- a/services/cubejs/src/utils/enrichmentPricing.js +++ /dev/null @@ -1,99 +0,0 @@ -import { resolveCodePricedCost } from "./pricingResolver.js"; - -export const ENRICHMENT_PRICING_ENV = "ENRICHMENT_CODE_PRICING_JSON"; - -let cachedRaw; -let cachedConfig; - -function configurationError(message) { - const error = new Error( - `enrichment billing configuration unavailable: ${message}`, - ); - error.status = 503; - error.code = "enrichment_billing_unavailable"; - return error; -} - -function parseAmount(value) { - const amount = String(value ?? ""); - if (!/^(0|[1-9]\d*)(?:\.\d+)?$/.test(amount)) { - throw configurationError("invalid code-based amount"); - } - return amount; -} - -export function parseEnrichmentCodePricing( - raw = process.env[ENRICHMENT_PRICING_ENV], -) { - const source = String(raw || "").trim(); - if (!source) throw configurationError(`${ENRICHMENT_PRICING_ENV} is missing`); - if (source === cachedRaw && cachedConfig) return cachedConfig; - - let parsed; - try { - parsed = JSON.parse(source); - } catch { - throw configurationError(`${ENRICHMENT_PRICING_ENV} is not valid JSON`); - } - if (!String(parsed?.pricing_code_version || "").trim()) { - throw configurationError("pricing_code_version is missing"); - } - if (!parsed.items || typeof parsed.items !== "object") { - throw configurationError("items are missing"); - } - - const items = {}; - for (const [item, entry] of Object.entries(parsed.items)) { - const currency = String(entry?.currency || "") - .trim() - .toUpperCase(); - if (!/^[A-Z]{3}$/.test(currency)) { - throw configurationError(`${item} requires currency`); - } - items[item] = { - amount: parseAmount(entry.amount), - currency, - }; - } - - cachedRaw = source; - cachedConfig = Object.freeze({ - pricingCodeVersion: String(parsed.pricing_code_version), - items: Object.freeze(items), - }); - return cachedConfig; -} - -/** - * Resolve only through the existing in-code/legacy-rate branch. The function - * has no Convex projection argument, making that dependency impossible here. - */ -export function resolveEnrichmentCodePrice(item, options = {}) { - const config = options.config || parseEnrichmentCodePricing(options.raw); - const entry = config.items[item]; - if (!entry) throw configurationError(`${item} has no code-based price`); - - const pricingResolution = resolveCodePricedCost({ - usage: { request: "1" }, - legacyRates: [ - { - meter: "request", - amount: entry.amount, - unitSize: 1, - qualifiers: {}, - }, - ], - legacyCurrency: entry.currency, - pricingCodeVersion: config.pricingCodeVersion, - }); - if (pricingResolution.pricingSource !== "legacy_runtime_rate") { - throw configurationError( - `${item} did not resolve through legacy_runtime_rate`, - ); - } - return { - pricingCodeVersion: config.pricingCodeVersion, - unitAmount: entry.amount, - pricingResolution, - }; -} diff --git a/services/cubejs/src/utils/enrichmentEntitlement.js b/services/cubejs/src/utils/legacyEnrichmentGuard.js similarity index 59% rename from services/cubejs/src/utils/enrichmentEntitlement.js rename to services/cubejs/src/utils/legacyEnrichmentGuard.js index 3d8db6d6..01db16ed 100644 --- a/services/cubejs/src/utils/enrichmentEntitlement.js +++ b/services/cubejs/src/utils/legacyEnrichmentGuard.js @@ -1,13 +1,6 @@ -import { createHmac, timingSafeEqual } from "node:crypto"; - -export const ENRICHMENT_PRODUCTS = [ - "ctx:day-archetype", - "ctx:weather-archetype", -]; -export const ENRICHMENT_CUBES = new Set(["CtxDayContext", "CtxWeatherContext"]); -export const ENRICHMENT_SERVING_TABLES = new Set([ - "day_context_v", - "weather_context_v", +export const LEGACY_ENRICHMENT_CUBES = new Set([ + "CtxDayContext", + "CtxWeatherContext", ]); const MEMBER_KEYS = new Set(["member", "dimension", "id"]); @@ -19,24 +12,6 @@ const QUERY_MEMBER_ARRAYS = [ "filters", ]; -const canonicalPayload = (payload) => - JSON.stringify({ - schema_version: payload.schema_version, - account_partition: payload.account_partition, - enabled: payload.enabled, - entitlement_revision: payload.entitlement_revision, - issued_at: payload.issued_at, - valid_until: payload.valid_until, - products: payload.products, - billing_connection_id: payload.billing_connection_id, - }); - -function signaturesMatch(left, right) { - const a = Buffer.from(String(left || "")); - const b = Buffer.from(String(right || "")); - return a.length > 0 && a.length === b.length && timingSafeEqual(a, b); -} - function unavailableError() { const error = new Error("403: Requested data is not available"); error.status = 403; @@ -99,74 +74,14 @@ export function collectResolvedMembers(query) { return [...members]; } -export function queryUsesEnrichment(query) { +export function queryUsesLegacyEnrichment(query) { return collectResolvedMembers(query).some((member) => - ENRICHMENT_CUBES.has(member.split(".", 1)[0]), + LEGACY_ENRICHMENT_CUBES.has(member.split(".", 1)[0]), ); } -export function validateEnrichmentLease( - securityContext, - { - signingKey = process.env.ENRICHMENT_ENTITLEMENT_SIGNING_KEY, - now = new Date(), - } = {}, -) { - const teamSettings = securityContext?.userScope?.teamProperties; - const partition = teamSettings?.partition; - const lease = teamSettings?.premium?.enrichment; - if (!partition || !signingKey || !lease) return { valid: false }; - - const issuedAt = Date.parse(lease.issued_at); - const validUntil = Date.parse(lease.valid_until); - const nowMs = now instanceof Date ? now.getTime() : new Date(now).getTime(); - if ( - lease.signature_version !== "hmac-sha256-v1" || - lease.enabled !== true || - typeof lease.entitlement_revision !== "string" || - !Array.isArray(lease.products) || - lease.products.join(",") !== ENRICHMENT_PRODUCTS.join(",") || - !/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test( - lease.billing_connection_id || "", - ) || - !Number.isFinite(issuedAt) || - !Number.isFinite(validUntil) || - !Number.isFinite(nowMs) || - issuedAt > nowMs || - validUntil <= nowMs - ) { - return { valid: false }; - } - - const payload = { - schema_version: 1, - account_partition: partition, - enabled: lease.enabled, - entitlement_revision: lease.entitlement_revision, - issued_at: lease.issued_at, - valid_until: lease.valid_until, - products: lease.products, - billing_connection_id: lease.billing_connection_id, - }; - const expected = createHmac("sha256", signingKey) - .update(canonicalPayload(payload)) - .digest("base64url"); - const valid = signaturesMatch(expected, lease.signature); - return { - valid, - connectionId: valid ? lease.billing_connection_id : null, - }; -} - -export function assertEnrichmentQueryAuthorized( - query, - securityContext, - options, -) { - if (!queryUsesEnrichment(query)) return; - if (!validateEnrichmentLease(securityContext, options).valid) { - throw unavailableError(); - } +export function assertNoLegacyEnrichmentQuery(query) { + if (queryUsesLegacyEnrichment(query)) throw unavailableError(); } function tokenizeSql(sql) { @@ -287,34 +202,14 @@ export function parseSqlTableReferences(sql) { return references; } -export function sqlEnrichmentBillingItems(sql) { - const items = new Set(); - for (const reference of parseSqlTableReferences(sql)) { - if (reference.schema !== "enrich") continue; - if (reference.table === "day_context_v") items.add("ctx:day-archetype"); - if (reference.table === "weather_context_v") { - items.add("ctx:weather-archetype"); - } - } - return ENRICHMENT_PRODUCTS.filter((item) => items.has(item)); -} - -export function assertSqlEnrichmentAuthorized(sql, securityContext, options) { +export function assertNoLegacyEnrichmentSql(sql) { const enrichmentRefs = parseSqlTableReferences(sql).filter( (reference) => reference.schema === "enrich", ); - if (enrichmentRefs.length === 0) return; - if ( - enrichmentRefs.some( - (reference) => !ENRICHMENT_SERVING_TABLES.has(reference.table), - ) || - !validateEnrichmentLease(securityContext, options).valid - ) { - throw unavailableError(); - } + if (enrichmentRefs.length > 0) throw unavailableError(); } -export function assertNoDirectEnrichmentObject(schema, table) { +export function assertNoDirectLegacyEnrichmentObject(schema, table) { const schemaName = String(schema || "") .replace(/^[`"[]|[`"\]]$/g, "") .toLowerCase(); @@ -330,7 +225,7 @@ export function assertNoDirectEnrichmentObject(schema, table) { } } -export function removeEnrichmentSchema(schema) { +export function removeLegacyEnrichmentSchema(schema) { if (!schema || typeof schema !== "object" || Array.isArray(schema)) return schema; return Object.fromEntries( diff --git a/services/cubejs/src/utils/queryRewrite.js b/services/cubejs/src/utils/queryRewrite.js index ffb9a6f6..34c0beb1 100644 --- a/services/cubejs/src/utils/queryRewrite.js +++ b/services/cubejs/src/utils/queryRewrite.js @@ -3,11 +3,7 @@ import YAML from "yaml"; import { fetchGraphQL } from "./graphql.js"; import { findDataSchemasByIds } from "./dataSourceHelpers.js"; import { parseCubesFromJs } from "./smart-generation/diffModels.js"; -import { - assertEnrichmentQueryAuthorized, - collectResolvedMembers, - ENRICHMENT_PRODUCTS, -} from "./enrichmentEntitlement.js"; +import { assertNoLegacyEnrichmentQuery } from "./legacyEnrichmentGuard.js"; const getColumnsArray = (cube) => [ ...(cube?.dimensions || []), @@ -140,14 +136,7 @@ async function buildCubeToTableMap(schemaVersion, fileIds) { const dims = new Set( (cube.dimensions || []).map((d) => d.name).filter(Boolean), ); - const billingItems = - cube.meta?.managed_by === "ctx-enrichment" && - Array.isArray(cube.meta?.billing_items) - ? cube.meta.billing_items.filter((item) => - ENRICHMENT_PRODUCTS.includes(item), - ) - : []; - mapping.set(cube.name, { sourceTable, dimensions: dims, billingItems }); + mapping.set(cube.name, { sourceTable, dimensions: dims }); } } } catch (err) { @@ -167,30 +156,6 @@ async function buildCubeToTableMap(schemaVersion, fileIds) { return mapping; } -/** - * Resolve billable products from the immutable metadata on the compiled - * managed cubes. Generated SQL is deliberately not inspected: a physical - * join can be present for planning reasons without the caller selecting an - * enrichment member. - */ -export async function resolveEnrichmentBillingItems(query, securityContext) { - const dataSource = securityContext?.userScope?.dataSource; - if (!dataSource?.schemaVersion || !Array.isArray(dataSource?.files)) - return []; - - const cubeMap = await buildCubeToTableMap( - dataSource.schemaVersion, - dataSource.files, - ); - const items = new Set(); - for (const member of collectResolvedMembers(query)) { - const cubeName = member.split(".", 1)[0]; - for (const item of cubeMap.get(cubeName)?.billingItems || []) - items.add(item); - } - return ENRICHMENT_PRODUCTS.filter((item) => items.has(item)); -} - /** * Extract cube names from query dimensions and measures. * Cube.js query members are formatted as "CubeName.memberName". @@ -215,10 +180,9 @@ function extractCubeNames(query) { * 2. Apply field-level access list check (non-owner/non-admin only) */ const queryRewrite = async (query, { securityContext }) => { - // Premium model authorization is evaluated against every resolved-member - // location before any rewrite can remove or replace query fields. This is - // the shared compiler boundary for REST and native SQL API sessions. - assertEnrichmentQueryAuthorized(query, securityContext); + // Retired managed enrichment cubes remain unavailable even if an old + // historical tenant version is restored. + assertNoLegacyEnrichmentQuery(query); const { userScope } = securityContext; const { diff --git a/services/cubejs/src/utils/repositoryFactory.js b/services/cubejs/src/utils/repositoryFactory.js index 6bbc652e..2f63a196 100644 --- a/services/cubejs/src/utils/repositoryFactory.js +++ b/services/cubejs/src/utils/repositoryFactory.js @@ -1,9 +1,6 @@ import mapSchemaToFile from "./mapSchemaToFile.js"; import { findDataSchemasByIds } from "./dataSourceHelpers.js"; -import { - validateEnrichmentLease, - ENRICHMENT_CUBES, -} from "./enrichmentEntitlement.js"; +import { LEGACY_ENRICHMENT_CUBES } from "./legacyEnrichmentGuard.js"; import YAML from "yaml"; const MANAGED_TEMPLATE_NAMES = new Set([ @@ -11,19 +8,12 @@ const MANAGED_TEMPLATE_NAMES = new Set([ "ctx_weather_context", ]); -const isManagedEnrichmentCube = (cube) => - ENRICHMENT_CUBES.has(cube?.name) || +const isLegacyEnrichmentCube = (cube) => + LEGACY_ENRICHMENT_CUBES.has(cube?.name) || cube?.meta?.managed_by === "ctx-enrichment" || MANAGED_TEMPLATE_NAMES.has(cube?.meta?.template); -export function filterUnentitledEnrichmentSchemas( - dataSchemas, - securityContext, - options, -) { - if (validateEnrichmentLease(securityContext, options).valid) - return dataSchemas; - +export function filterLegacyEnrichmentSchemas(dataSchemas) { const filtered = []; for (const schema of dataSchemas || []) { const managedFile = MANAGED_TEMPLATE_NAMES.has( @@ -40,7 +30,7 @@ export function filterUnentitledEnrichmentSchemas( continue; } const cubes = document.cubes.filter( - (cube) => !isManagedEnrichmentCube(cube), + (cube) => !isLegacyEnrichmentCube(cube), ); if (cubes.length === 0) continue; if (cubes.length === document.cubes.length) { @@ -76,10 +66,7 @@ const repositoryFactory = ({ securityContext }) => { const ids = securityContext?.userScope?.dataSource?.files; const dataSchemas = await findDataSchemasByIds({ ids }); - return filterUnentitledEnrichmentSchemas( - dataSchemas, - securityContext, - ).map(mapSchemaToFile); + return filterLegacyEnrichmentSchemas(dataSchemas).map(mapSchemaToFile); }, }; }; diff --git a/templates/enrichment/__tests__/templates.test.js b/templates/enrichment/__tests__/templates.test.js deleted file mode 100644 index 4ae6abc0..00000000 --- a/templates/enrichment/__tests__/templates.test.js +++ /dev/null @@ -1,135 +0,0 @@ -import assert from "node:assert/strict"; -import { readFile } from "node:fs/promises"; -import path from "node:path"; -import { describe, it } from "node:test"; -import { fileURLToPath } from "node:url"; - -import YAML from "yaml"; -import { - buildPublication, - generateJoinStubs, - validateCompatibilityMatrix, -} from "../publish.js"; - -const ROOT = path.dirname(path.dirname(fileURLToPath(import.meta.url))); - -async function cube(name) { - const document = YAML.parse(await readFile(path.join(ROOT, name), "utf8")); - assert.equal(document.cubes.length, 1); - return document.cubes[0]; -} - -describe("Spec 102 enrichment templates", () => { - it("carry managed provenance, exact billing items, and public-view-only sources", async () => { - const day = await cube("ctx_day_context.yml"); - const weather = await cube("ctx_weather_context.yml"); - assert.deepEqual(day.meta.billing_items, ["ctx:day-archetype"]); - assert.deepEqual(weather.meta.billing_items, ["ctx:weather-archetype"]); - for (const model of [day, weather]) { - assert.equal(model.meta.default_model, true); - assert.equal(model.meta.managed_by, "ctx-enrichment"); - assert.equal(model.public, true); - assert.doesNotMatch( - JSON.stringify(model), - /release_(manifest|pointer)|enrich\.(day|weather)_context(?!_v)/, - ); - assert.equal(model.measures, undefined); - } - assert.equal(day.sql_table, "enrich.day_context_v"); - assert.match(weather.sql, /enrich\.weather_context_v/); - assert.doesNotMatch(weather.sql, /country_code\s*=\s*'IS'/); - assert.equal(weather.meta.weather_country, undefined); - assert.match( - weather.dimensions.find((d) => d.name === "contextKey").sql, - /country_code/, - ); - }); - - it("keeps categorical markers as strings", async () => { - const day = await cube("ctx_day_context.yml"); - const weather = await cube("ctx_weather_context.yml"); - assert.equal( - day.dimensions.find((d) => d.name === "archetypeMarker").type, - "string", - ); - assert.equal( - weather.dimensions.find((d) => d.name === "weatherMarker").type, - "string", - ); - }); - - it("generates only matrix-approved, country-correlated many-to-one joins", () => { - const matrix = validateCompatibilityMatrix({ - schema_version: 1, - status: "approved", - entries: [ - { - fact_model: "SemanticEvents", - products: ["day", "weather"], - event_time_expr: "{CUBE}.timestamp", - timezone_source: "{CUBE}.context_timezone", - timezone_valid_expr: "{CUBE}.timezone_is_valid = 1", - local_date_expr: "toDate({CUBE}.local_time)", - location_rule: "context_point", - country_source: "{CUBE}.country_code", - country_location_proof: "evidence/gate-1/profile.json", - geohash_expr: "{CUBE}.geohash6", - cardinality: "many_to_one", - verification_owner: "analytics-platform", - }, - ], - }); - const [stub] = generateJoinStubs(matrix); - assert.equal(stub.fact_model, "SemanticEvents"); - assert.equal(stub.joins.length, 2); - assert.ok(stub.joins.every((join) => join.relationship === "many_to_one")); - assert.doesNotMatch(stub.joins[0].sql, /= 'IS'/); - assert.doesNotMatch(stub.joins[1].sql, /= 'IS'/); - assert.ok(stub.joins.every((join) => join.sql.includes("countryCode"))); - assert.ok( - stub.joins.every((join) => - join.sql.startsWith("({CUBE}.timezone_is_valid = 1) AND"), - ), - ); - assert.ok( - stub.joins.every((join) => - join.sql.includes("(toDate({CUBE}.local_time))"), - ), - ); - assert.ok(stub.joins.every((join) => !join.sql.includes("toTimeZone"))); - }); - - it("rejects an entry without an explicit timezone guard and local-date expression", () => { - assert.throws( - () => - validateCompatibilityMatrix({ - schema_version: 1, - status: "approved", - entries: [ - { - fact_model: "SemanticEvents", - products: ["day"], - event_time_expr: "{CUBE}.timestamp", - timezone_source: "{CUBE}.context_timezone", - location_rule: "unique_only", - country_source: "{CUBE}.country_code", - country_location_proof: "evidence/gate-1/profile.json", - geohash_expr: "{CUBE}.geohash6", - cardinality: "many_to_one", - verification_owner: "analytics-platform", - }, - ], - }), - /timezone_valid_expr/, - ); - }); - - it("publishes no join stubs while Gate 1 remains empty and is checksum-idempotent", async () => { - const first = await buildPublication(); - const second = await buildPublication(); - assert.equal(first.matrixStatus, "gate_1_open"); - assert.deepEqual(first.joinStubs, []); - assert.equal(first.checksum, second.checksum); - assert.deepEqual(first.files, second.files); - }); -}); diff --git a/templates/enrichment/compatibility-matrix.yaml b/templates/enrichment/compatibility-matrix.yaml deleted file mode 100644 index de9c8afd..00000000 --- a/templates/enrichment/compatibility-matrix.yaml +++ /dev/null @@ -1,10 +0,0 @@ -schema_version: 1 -status: gate_1_open -# No fact is approved yet. The publisher generates no join stubs from an empty -# matrix; this is deliberate and prevents an unprofiled location/country rule -# from changing fact grain. Approved entries must satisfy the Spec 102 Gate 1 -# schema and carry a real verification owner. ClickHouse 26.7 cannot accept a -# column-valued timezone in toTimeZone(), so every future entry must also carry -# timezone_valid_expr and a separately proven local_date_expr; invalid zones -# must fail the join instead of falling back to UTC. -entries: [] diff --git a/templates/enrichment/ctx_day_context.yml b/templates/enrichment/ctx_day_context.yml deleted file mode 100644 index 800399bc..00000000 --- a/templates/enrichment/ctx_day_context.yml +++ /dev/null @@ -1,104 +0,0 @@ -cubes: - - name: CtxDayContext - sql_table: enrich.day_context_v - public: true - meta: - default_model: true - managed_by: ctx-enrichment - template: ctx_day_context - source_table: day_context_v - billing_items: - - ctx:day-archetype - dimensions: - - name: contextKey - sql: concat({CUBE}.request_geohash6, ':', {CUBE}.country_code, ':', toString({CUBE}.local_date)) - type: string - primary_key: true - public: false - - name: requestGeohash6 - sql: request_geohash6 - type: string - - name: countryCode - sql: country_code - type: string - - name: localDate - sql: local_date - type: time - - name: timezone - sql: timezone - type: string - - name: isWorkday - sql: is_workday - type: boolean - - name: isHoliday - sql: is_holiday - type: boolean - - name: isWeekend - sql: is_weekend - type: boolean - - name: isObservance - sql: is_observance - type: boolean - - name: isSeasonal - sql: is_seasonal - type: boolean - - name: isClosed - sql: is_closed - type: boolean - - name: isClockChange - sql: is_clock_change - type: boolean - - name: isVacationSeason - sql: is_vacation_season - type: boolean - - name: fromNoon - sql: from_noon - type: boolean - - name: dayOfWeek - sql: day_of_week - type: number - - name: dayOfMonth - sql: day_of_month - type: number - - name: dayOfYear - sql: day_of_year - type: number - - name: monthOfYear - sql: month_of_year - type: number - - name: monthName - sql: month_name - type: string - - name: weekOfYear - sql: week_of_year - type: number - - name: weekOfMonth - sql: week_of_month - type: number - - name: firstDayOfWeek - sql: first_day_of_week - type: number - - name: dayName - sql: day_name - type: string - - name: holidayTitle - sql: holiday_title - type: string - - name: seasonName - sql: season_name - type: string - - name: hoursOfDaylight - sql: hours_of_daylight - type: number - - name: archetypeMarker - sql: archetype_marker - type: string - - name: wordCloud - sql: arrayStringConcat(word_cloud, ', ') - type: string - - name: snapshotId - sql: snapshot_id - type: string - - name: refreshedAt - sql: refreshed_at - type: time diff --git a/templates/enrichment/ctx_weather_context.yml b/templates/enrichment/ctx_weather_context.yml deleted file mode 100644 index 25baea80..00000000 --- a/templates/enrichment/ctx_weather_context.yml +++ /dev/null @@ -1,110 +0,0 @@ -cubes: - - name: CtxWeatherContext - sql: SELECT * FROM enrich.weather_context_v - public: true - meta: - default_model: true - managed_by: ctx-enrichment - template: ctx_weather_context - source_table: weather_context_v - billing_items: - - ctx:weather-archetype - dimensions: - - name: contextKey - sql: concat({CUBE}.request_geohash6, ':', {CUBE}.country_code, ':', toString({CUBE}.local_date)) - type: string - primary_key: true - public: false - - name: requestGeohash6 - sql: request_geohash6 - type: string - - name: countryCode - sql: country_code - type: string - - name: localDate - sql: local_date - type: time - - name: timezone - sql: timezone - type: string - - name: valueBasis - sql: value_basis - type: string - - name: forecastLeadHours - sql: forecast_lead_h - type: number - - name: contextAsOf - sql: ctx_as_of - type: time - - name: temperature - sql: temperature - type: number - - name: feelsLike - sql: feels_like - type: number - - name: precipitationRain - sql: precipitation_rain - type: number - - name: precipitationSnow - sql: precipitation_snow - type: number - - name: windSpeed - sql: wind_speed - type: number - - name: windDirection - sql: wind_direction - type: number - - name: windGust - sql: wind_gust - type: number - - name: cloudCover - sql: cloud_cover - type: number - - name: pressure - sql: pressure - type: number - - name: humidity - sql: humidity - type: number - - name: dewPoint - sql: dew_point - type: number - - name: visibility - sql: visibility - type: number - - name: icon - sql: icon - type: string - - name: temperatureRegime - sql: temp_regime - type: string - - name: precipitationType - sql: precip_type - type: string - - name: precipitationIntensity - sql: precip_intensity - type: string - - name: windRegime - sql: wind_regime - type: string - - name: skyCondition - sql: sky_condition - type: string - - name: weatherMarker - sql: weather_marker - type: string - - name: wordCloud - sql: arrayStringConcat(word_cloud, ', ') - type: string - - name: sourceGeohash - sql: source_geohash - type: string - - name: resolution - sql: resolution - type: number - - name: snapshotId - sql: snapshot_id - type: string - - name: refreshedAt - sql: refreshed_at - type: time diff --git a/templates/enrichment/publish.js b/templates/enrichment/publish.js deleted file mode 100644 index 2effd4ec..00000000 --- a/templates/enrichment/publish.js +++ /dev/null @@ -1,122 +0,0 @@ -import { createHash } from "node:crypto"; -import { readFile } from "node:fs/promises"; -import path from "node:path"; -import { fileURLToPath } from "node:url"; - -import YAML from "yaml"; - -const ROOT = path.dirname(fileURLToPath(import.meta.url)); -const PRODUCT_CUBES = { - day: "CtxDayContext", - weather: "CtxWeatherContext", -}; - -function requiredString(entry, key) { - if (typeof entry?.[key] !== "string" || !entry[key].trim()) { - throw new Error(`compatibility entry requires ${key}`); - } - return entry[key].trim(); -} - -export function validateCompatibilityMatrix(matrix) { - if (matrix?.schema_version !== 1 || !Array.isArray(matrix?.entries)) { - throw new Error( - "compatibility matrix must be schema_version 1 with entries[]", - ); - } - for (const entry of matrix.entries) { - for (const key of [ - "fact_model", - "event_time_expr", - "timezone_source", - "timezone_valid_expr", - "local_date_expr", - "location_rule", - "country_source", - "country_location_proof", - "geohash_expr", - "verification_owner", - ]) - requiredString(entry, key); - if (entry.cardinality !== "many_to_one") { - throw new Error("compatibility entry cardinality must be many_to_one"); - } - if ( - !Array.isArray(entry.products) || - entry.products.length === 0 || - entry.products.some((product) => !PRODUCT_CUBES[product]) - ) { - throw new Error( - "compatibility entry products must contain day and/or weather", - ); - } - } - return matrix; -} - -const joinSql = (entry, product) => { - const cube = PRODUCT_CUBES[product]; - // ClickHouse 26.7 requires a constant time-zone argument to toTimeZone(). - // The compatibility gate therefore supplies a separately proven local-date - // expression and an explicit IANA-zone validity predicate. Invalid or - // missing timezone rows fail the join instead of falling back to UTC. - return `(${entry.timezone_valid_expr}) AND (${entry.geohash_expr}) = {${cube}}.requestGeohash6 AND (${entry.country_source}) = {${cube}}.countryCode AND (${entry.local_date_expr}) = {${cube}}.localDate`; -}; - -export function generateJoinStubs(matrix) { - validateCompatibilityMatrix(matrix); - return matrix.entries.map((entry) => ({ - fact_model: entry.fact_model, - verification_owner: entry.verification_owner, - joins: entry.products.map((product) => ({ - name: PRODUCT_CUBES[product], - relationship: "many_to_one", - sql: joinSql(entry, product), - })), - })); -} - -export async function buildPublication() { - const names = ["ctx_day_context.yml", "ctx_weather_context.yml"]; - const files = await Promise.all( - names.map(async (name) => ({ - name, - code: await readFile(path.join(ROOT, name), "utf8"), - })), - ); - const matrixCode = await readFile( - path.join(ROOT, "compatibility-matrix.yaml"), - "utf8", - ); - const matrix = validateCompatibilityMatrix(YAML.parse(matrixCode)); - const joinStubs = generateJoinStubs(matrix); - const joinsCode = YAML.stringify({ - schema_version: 1, - generated_from: "compatibility-matrix.yaml", - join_stubs: joinStubs, - }); - files.push({ name: "enrichment_joins.yaml", code: joinsCode }); - const checksum = createHash("sha256") - .update( - files - .map((file) => `${file.name}\0${file.code}`) - .sort() - .join("\0"), - ) - .digest("hex"); - return { checksum, files, matrixStatus: matrix.status, joinStubs }; -} - -if (process.argv[1] === fileURLToPath(import.meta.url)) { - const publication = await buildPublication(); - // Publication to the configured Global Templates branch is intentionally - // refused while Gate 1 has no approved entries. This check mode still gives - // CI a deterministic artifact/checksum and makes accidental empty rollout - // impossible. - process.stdout.write(`${JSON.stringify(publication, null, 2)}\n`); - if (publication.matrixStatus !== "approved") { - process.stderr.write( - "Gate 1 is not approved; no remote publication was attempted.\n", - ); - } -}