From 4489bc19d40f9cbe49a8f7481431348e5a1419aa Mon Sep 17 00:00:00 2001 From: stefanbaxter Date: Wed, 19 Aug 2026 17:30:03 +0000 Subject: [PATCH] feat(099): Hasura lifecycle triggers, query log, D6 source, naming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit synmetrix side of 099 Semantic Events: - Hasura event triggers → ACTIONS handlers: Model Version Created, Branch Created/Deleted, SQL Credential Created/Deleted (tenant via datasource→team). Access Rule Created/Updated/Deleted from the manageQueryRewriteRule chokepoint (platform partition). New emitter services/actions/src/utils/semanticEvents.js. - Buffered Query Executed log event (cube logger hook; enqueue-only, never affects query perf). - D6: producer identity via envelope `source` (synmetrix); properties.producer removed; producer re-stamp immutable on the caller-merge path. - Naming: Connection Credential Expired rename. Gate-verified: node --check green. Live e2e verified locally (Hasura trigger → handler → canonical envelope → ingress capture). Co-Authored-By: Claude Opus 4.8 --- .../actions/src/rpc/emitBranchLifecycle.js | 79 ++ .../src/rpc/emitModelVersionCreated.js | 85 ++ .../src/rpc/emitSqlCredentialLifecycle.js | 80 ++ .../actions/src/rpc/manageQueryRewriteRule.js | 32 +- services/actions/src/utils/semanticEvents.js | 166 ++++ .../cubejs/src/routes/deleteDataschema.js | 19 + .../cubejs/src/routes/generateDataSchema.js | 28 +- services/cubejs/src/routes/loadExport.js | 52 ++ services/cubejs/src/routes/profileTable.js | 55 ++ services/cubejs/src/routes/reconcileTeam.js | 30 + services/cubejs/src/routes/refreshCompiler.js | 14 + services/cubejs/src/routes/runSql.js | 65 +- services/cubejs/src/routes/smartGenerate.js | 43 +- services/cubejs/src/routes/testConnection.js | 34 + .../cubejs/src/routes/validateInBranch.js | 19 + services/cubejs/src/routes/versionRollback.js | 18 + services/cubejs/src/utils/checkAuth.js | 11 + services/cubejs/src/utils/checkSqlAuth.js | 27 + .../cubejs/src/utils/dataSourceHelpers.js | 24 +- services/cubejs/src/utils/eventEmitter.js | 816 ++++++++++++++++++ services/cubejs/src/utils/logging.js | 33 +- .../src/utils/smart-generation/llmEnricher.js | 48 +- .../utils/smart-generation/modelAdvisor.js | 42 +- services/hasura/metadata/tables.yaml | 73 ++ 24 files changed, 1870 insertions(+), 23 deletions(-) create mode 100644 services/actions/src/rpc/emitBranchLifecycle.js create mode 100644 services/actions/src/rpc/emitModelVersionCreated.js create mode 100644 services/actions/src/rpc/emitSqlCredentialLifecycle.js create mode 100644 services/actions/src/utils/semanticEvents.js create mode 100644 services/cubejs/src/utils/eventEmitter.js diff --git a/services/actions/src/rpc/emitBranchLifecycle.js b/services/actions/src/rpc/emitBranchLifecycle.js new file mode 100644 index 00000000..9687f76b --- /dev/null +++ b/services/actions/src/rpc/emitBranchLifecycle.js @@ -0,0 +1,79 @@ +import { fetchGraphQL } from "../utils/graphql.js"; +import { emitLifecycleEvent } from "../utils/semanticEvents.js"; + +/** + * emit_branch_lifecycle — Hasura event-trigger handler (099 US7, FR-091, T087). + * + * Fires on `branches` INSERT and DELETE (both raw-GraphQL, no JS chokepoint). One + * handler covers both ops: INSERT → `Branch Created`, DELETE → `Branch Deleted`. + * Emits the canonical lifecycle event to the FraiOS ingress via the never-throw + * `emitLifecycleEvent` substrate. + * + * Tenant: partition == the owning synmetrix team (team.settings.partition); + * accountId == the team id — resolved from the branch's datasource → team + * (admin-secret read). On DELETE the tenant is resolved from the OLD row's + * datasource_id (the datasource itself is not cascaded away by a branch delete). + * Emission is fire-and-forget and NEVER blocks or fails the trigger (FR-007). + */ +const DATASOURCE_TENANT = ` + query DatasourceTenant($id: uuid!) { + datasources_by_pk(id: $id) { + id + name + team_id + team { id name settings } + } + } +`; + +export default async (session, input) => { + const op = input?.event?.op; // "INSERT" | "DELETE" | "MANUAL" + const isDelete = op === "DELETE"; + const row = isDelete ? input?.event?.data?.old : input?.event?.data?.new; + if (!row?.id || !row.datasource_id) { + return { ok: true, skipped: true }; + } + + let datasource = null; + try { + const res = await fetchGraphQL(DATASOURCE_TENANT, { id: row.datasource_id }); + datasource = res?.data?.datasources_by_pk || null; + } catch { + // non-fatal + } + + const team = datasource?.team || null; + const partition = team?.settings?.partition ?? null; + const accountId = team?.id ?? null; + if (!accountId && !partition) { + return { ok: true, skipped: true }; + } + + const sessionVars = input?.event?.session_variables || session || {}; + const userId = + row.user_id || + sessionVars["x-hasura-user-id"] || + sessionVars["X-Hasura-User-Id"] || + null; + + const result = await emitLifecycleEvent({ + event: isDelete ? "Branch Deleted" : "Branch Created", + partition, + accountId, + accountLabel: team?.name ?? null, + userId, + about: { + entity_type: "Data Model", + id: row.id, + label: row.name ?? null, + }, + status: isDelete ? "deleted" : "created", + properties: { + branch_name: row.name ?? null, + datasource_id: row.datasource_id, + branch_status: row.status ?? null, + }, + }); + + return { ok: true, emitted: !!result?.ok, skipped: !!result?.skipped }; +}; diff --git a/services/actions/src/rpc/emitModelVersionCreated.js b/services/actions/src/rpc/emitModelVersionCreated.js new file mode 100644 index 00000000..29ca4760 --- /dev/null +++ b/services/actions/src/rpc/emitModelVersionCreated.js @@ -0,0 +1,85 @@ +import { fetchGraphQL } from "../utils/graphql.js"; +import { emitLifecycleEvent } from "../utils/semanticEvents.js"; + +/** + * emit_model_version_created — Hasura event-trigger handler (099 US7, FR-091, T087). + * + * Fires AFTER a `versions.insert` commits. The editor's "pure save" and other + * raw-GraphQL version creations never pass through a cubejs code chokepoint, so + * this trigger is the only place the version-creation fact can be observed. It + * emits the canonical `Model Version Created` lifecycle event to the FraiOS + * ingress via the never-throw `emitLifecycleEvent` substrate. + * + * Tenant: partition == the owning synmetrix team (team.settings.partition — the + * exact key the semantic_events RLS filters on); accountId == the team id. Both + * are resolved from the version's branch → datasource → team (admin-secret read, + * so team.settings is visible). Emission is fire-and-forget and NEVER blocks or + * fails the trigger (FR-007). + */ +const BRANCH_TENANT = ` + query BranchTenant($id: uuid!) { + branches_by_pk(id: $id) { + id + name + datasource_id + datasource { + id + name + team_id + team { id name settings } + } + } + } +`; + +export default async (session, input) => { + const row = input?.event?.data?.new; + if (!row?.id || !row.branch_id) { + return { ok: true, skipped: true }; + } + + let branch = null; + try { + const res = await fetchGraphQL(BRANCH_TENANT, { id: row.branch_id }); + branch = res?.data?.branches_by_pk || null; + } catch { + // non-fatal — never block the trigger on a tenant lookup + } + + const team = branch?.datasource?.team || null; + const partition = team?.settings?.partition ?? null; + const accountId = team?.id ?? null; + if (!accountId && !partition) { + // No resolvable tenant → do not file under a synthetic one (A4). + return { ok: true, skipped: true }; + } + + const sessionVars = input?.event?.session_variables || session || {}; + const userId = + row.user_id || + sessionVars["x-hasura-user-id"] || + sessionVars["X-Hasura-User-Id"] || + null; + + const result = await emitLifecycleEvent({ + event: "Model Version Created", + partition, + accountId, + accountLabel: team?.name ?? null, + userId, + about: { + entity_type: "Data Model", + id: row.branch_id, + label: branch?.name ?? null, + }, + status: "created", + properties: { + version_id: row.id, + datasource_id: branch?.datasource_id ?? null, + origin: row.origin ?? null, + checksum: row.checksum ?? null, + }, + }); + + return { ok: true, emitted: !!result?.ok, skipped: !!result?.skipped }; +}; diff --git a/services/actions/src/rpc/emitSqlCredentialLifecycle.js b/services/actions/src/rpc/emitSqlCredentialLifecycle.js new file mode 100644 index 00000000..8a24ae77 --- /dev/null +++ b/services/actions/src/rpc/emitSqlCredentialLifecycle.js @@ -0,0 +1,80 @@ +import { fetchGraphQL } from "../utils/graphql.js"; +import { emitLifecycleEvent } from "../utils/semanticEvents.js"; + +/** + * emit_sql_credential_lifecycle — Hasura event-trigger handler (099 US7, FR-091, T089). + * + * Fires on `sql_credentials` INSERT and DELETE (raw-GraphQL, no JS chokepoint). + * INSERT → `SQL Credential Created`, DELETE → `SQL Credential Deleted`. Emits the + * canonical lifecycle event to the FraiOS ingress via the never-throw + * `emitLifecycleEvent` substrate. + * + * Tenant: partition == the owning synmetrix team (team.settings.partition); + * accountId == the team id — resolved from the credential's datasource → team + * (admin-secret read). A legacy credential whose datasource has no FraiOS tenant + * (no team / no partition) is skipped rather than filed under a synthetic tenant + * (A4). ABOUT carries the credential id under the Secret family; the secret value + * is NEVER included. Emission is fire-and-forget and never fails the trigger (FR-007). + */ +const DATASOURCE_TENANT = ` + query DatasourceTenant($id: uuid!) { + datasources_by_pk(id: $id) { + id + name + team_id + team { id name settings } + } + } +`; + +export default async (session, input) => { + const op = input?.event?.op; // "INSERT" | "DELETE" | "MANUAL" + const isDelete = op === "DELETE"; + const row = isDelete ? input?.event?.data?.old : input?.event?.data?.new; + if (!row?.id || !row.datasource_id) { + return { ok: true, skipped: true }; + } + + let datasource = null; + try { + const res = await fetchGraphQL(DATASOURCE_TENANT, { id: row.datasource_id }); + datasource = res?.data?.datasources_by_pk || null; + } catch { + // non-fatal + } + + const team = datasource?.team || null; + const partition = team?.settings?.partition ?? null; + const accountId = team?.id ?? null; + if (!accountId && !partition) { + // Legacy credential with no FraiOS tenant → skip (A4). + return { ok: true, skipped: true }; + } + + const sessionVars = input?.event?.session_variables || session || {}; + const userId = + row.user_id || + sessionVars["x-hasura-user-id"] || + sessionVars["X-Hasura-User-Id"] || + null; + + const result = await emitLifecycleEvent({ + event: isDelete ? "SQL Credential Deleted" : "SQL Credential Created", + partition, + accountId, + accountLabel: team?.name ?? null, + userId, + about: { + entity_type: "Secret", + id: row.id, + label: row.username ?? null, // username only — the secret value is never emitted + }, + status: isDelete ? "deleted" : "created", + properties: { + datasource_id: row.datasource_id, + username: row.username ?? null, + }, + }); + + return { ok: true, emitted: !!result?.ok, skipped: !!result?.skipped }; +}; diff --git a/services/actions/src/rpc/manageQueryRewriteRule.js b/services/actions/src/rpc/manageQueryRewriteRule.js index d3c7ecfd..0362dd85 100644 --- a/services/actions/src/rpc/manageQueryRewriteRule.js +++ b/services/actions/src/rpc/manageQueryRewriteRule.js @@ -2,6 +2,25 @@ import apiError from "../utils/apiError.js"; import { invalidateRulesCache } from "../utils/cubeCache.js"; import { fetchGraphQL } from "../utils/graphql.js"; import { isPortalAdmin } from "../utils/portalAdmin.js"; +import { emitLifecycleEvent } from "../utils/semanticEvents.js"; + +// 099 US7 (FR-091, T089): row-level Access Rules (query_rewrite_rules) are +// PLATFORM-GLOBAL config edited only by portal admins — there is no per-tenant +// row. They are attributed to the platform tenant (partition), overridable via +// PLATFORM_PARTITION; ACTED_BY carries the acting admin. Emission is +// fire-and-forget and never affects the mutation (FR-007). +const PLATFORM_PARTITION = process.env.PLATFORM_PARTITION || "fftech.is"; + +async function emitAccessRule(event, status, ruleId, userId, properties) { + await emitLifecycleEvent({ + event, + partition: PLATFORM_PARTITION, + userId, + about: { entity_type: "Policy", id: ruleId, label: properties?.cube_name ?? null }, + status, + properties, + }); +} const insertRuleMutation = ` mutation InsertRule($object: query_rewrite_rules_insert_input!) { @@ -84,7 +103,16 @@ export default async (session, input) => { }); const ruleId = res?.data?.insert_query_rewrite_rules_one?.id; - if (ruleId) invalidateRulesCache(); + if (ruleId) { + invalidateRulesCache(); + void emitAccessRule("Access Rule Created", "created", ruleId, userId, { + cube_name, + dimension, + property_source, + property_key, + operator: op, + }); + } return { success: !!ruleId, rule_id: ruleId || null }; } @@ -119,6 +147,7 @@ export default async (session, input) => { await fetchGraphQL(updateRuleMutation, { id, set: updates }); invalidateRulesCache(); + void emitAccessRule("Access Rule Updated", "updated", id, userId, updates); return { success: true, rule_id: id }; } @@ -133,6 +162,7 @@ export default async (session, input) => { await fetchGraphQL(deleteRuleMutation, { id }); invalidateRulesCache(); + void emitAccessRule("Access Rule Deleted", "deleted", id, userId, null); return { success: true, rule_id: id }; } diff --git a/services/actions/src/utils/semanticEvents.js b/services/actions/src/utils/semanticEvents.js new file mode 100644 index 00000000..9eee246b --- /dev/null +++ b/services/actions/src/utils/semanticEvents.js @@ -0,0 +1,166 @@ +/** + * semanticEvents — focused, never-throw semantic-event emitter for the ACTIONS + * webhook service (099 Semantic Events, FR-091). Hasura event-trigger handlers + * use it to emit lifecycle events (Model Version Created / Branch Created-Deleted + * / access-control) to the FraiOS ingress. Mirrors the canonical contract of + * services/cubejs/src/utils/eventEmitter.js (kept minimal: synmetrix ids are + * UUIDs, so entity_gid is passthrough and no FNV hash is needed here). + * + * Emission is best-effort + fire-and-forget: it NEVER throws (FR-007) and never + * blocks the webhook response. The ingress derives the tenant partition from the + * OWNED_BY account gid (entity_gid), so `accountId` alone attributes correctly. + */ +import crypto from "crypto"; +import { SignJWT } from "jose"; +// Native global fetch (Node 20+) — same choice as utils/graphql.js, avoids the +// node-fetch socket-hang-up issues. + +const { TOKEN_SECRET, INGRESSION_HOST, CXS_EVENT_SOURCE } = process.env; +const DEFAULT_SOURCE = "synmetrix"; +const ID_TYPE = "FraiOS"; +const DNS_NAMESPACE = "6ba7b810-9dad-11d1-80b4-00c04fd430c8"; +const UUID_REGEX = + /^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/; + +/** RFC-4122 v5 (SHA-1) uuid — matches the `uuid` package's v5 used by cubejs. */ +function uuid5(name, namespace = DNS_NAMESPACE) { + const nsBytes = Buffer.from(namespace.replace(/-/g, ""), "hex"); + const hash = crypto + .createHash("sha1") + .update(nsBytes) + .update(String(name), "utf8") + .digest(); + const b = hash.subarray(0, 16); + b[6] = (b[6] & 0x0f) | 0x50; // version 5 + b[8] = (b[8] & 0x3f) | 0x80; // variant + const h = b.toString("hex"); + return `${h.slice(0, 8)}-${h.slice(8, 12)}-${h.slice(12, 16)}-${h.slice( + 16, + 20 + )}-${h.slice(20)}`; +} + +/** Deterministic, replay-stable message_id from stable parts joined by ":". */ +export function messageIdFor(...parts) { + return uuid5(parts.map((p) => String(p ?? "")).join(":")); +} + +/** entity_gid: RFC-4122 passthrough (synmetrix ids are UUIDs), else uuid5. */ +function normalizeEntityGid(raw) { + const s = String(raw ?? ""); + return UUID_REGEX.test(s) ? s : uuid5(s); +} + +/** One involve entry (fraios id model). */ +function involve(role, entity_type, value, label) { + const entry = { role, entity_type, entity_gid: normalizeEntityGid(value), id_type: ID_TYPE }; + const s = String(value ?? ""); + if (s && !UUID_REGEX.test(s)) entry.id = s; + if (label) entry.label = label; + return entry; +} + +/** + * Build a canonical lifecycle envelope: OWNED_BY/Account (+ ACTED_BY/Person when + * a human is attributable) + an optional ABOUT subject. Emitted `type: "track"`. + */ +export function buildLifecycleEvent({ + event, + partition = null, // team.settings.partition — the canonical tenant key + accountId = null, // the owning team (synmetrix's account-equivalent) id + accountLabel = null, + userId = null, + about = null, // { entity_type, id, label } + status = "ok", + dimensions = null, + properties = null, + timestamp = null, +} = {}) { + const ts = timestamp || new Date().toISOString(); + const ownerId = String(accountId ?? partition ?? ""); + const aboutId = about && about.id != null && String(about.id) !== "" ? String(about.id) : null; + const entityGid = aboutId || ownerId; + const message_id = messageIdFor(event, partition ?? "", ownerId, aboutId ?? "", status, ts); + + const involves = [involve("OWNED_BY", "Account", ownerId, accountLabel)]; + if (userId) involves.push(involve("ACTED_BY", "Person", String(userId))); + if (aboutId && about.entity_type) { + involves.push(involve("ABOUT", String(about.entity_type), aboutId, about.label)); + } + + const source = (CXS_EVENT_SOURCE || "").trim() || DEFAULT_SOURCE; + const env = process.env.KUBERNETES_SERVICE_HOST ? "cluster" : "local"; + return { + type: "track", + event, + abstract_event: event, + message_id, + event_gid: uuid5(message_id.toLowerCase()), + timestamp: ts, + // The tenant key. partition == the synmetrix team (team.settings.partition), + // the same key the semantic_events RLS filters on. Fall back to the ingress + // account-derived partition only when a partition was not resolved. + partition: partition ? String(partition) : "", + entity_gid: normalizeEntityGid(entityGid), + customer_facing: 0, + source, + dimensions: { status, ...(dimensions || {}), environment: env }, + involves, + properties: { ...(properties || {}) }, + }; +} + +async function mintServiceToken({ accountId = null, partition = null } = {}) { + if (!TOKEN_SECRET || (!accountId && !partition)) return null; + try { + const claims = { provider: "synmetrix" }; + if (accountId != null) claims.accountId = String(accountId); + if (partition != null) claims.partition = String(partition); + return await new SignJWT(claims) + .setProtectedHeader({ alg: "HS256" }) + .setIssuedAt() + .setIssuer("services:actions") + .setAudience("fraios:ingression") + .setExpirationTime("5m") + .sign(new TextEncoder().encode(TOKEN_SECRET)); + } catch { + return null; + } +} + +/** + * Best-effort, never-throw POST of a lifecycle event to the ingress. Attributes + * the tenant via `partition` (== the owning team; the canonical key) and/or + * `accountId`. Skips silently when no tenant or no ingress/secret is configured + * (A4 — a credential is never invented). + */ +export async function emitLifecycleEvent(args = {}) { + try { + const { accountId = null, partition = null } = args; + if ((!accountId && !partition) || !INGRESSION_HOST) { + return { ok: false, skipped: true }; + } + const envelope = buildLifecycleEvent(args); + const writekey = await mintServiceToken({ accountId, partition }); + if (!writekey) return { ok: false, skipped: true }; + const url = `${INGRESSION_HOST}/api/s/${envelope.type || "track"}`; + // Bounded so a slow/hung ingress can never stall a caller (a Hasura trigger + // handler or an admin RPC). On timeout the POST aborts and we return not-ok — + // emission is best-effort and never blocks the operation (FR-007). + const ctrl = new AbortController(); + const timer = setTimeout(() => ctrl.abort(), 5000); + try { + const res = await fetch(url, { + method: "POST", + headers: { "content-type": "application/json", writekey }, + body: JSON.stringify(envelope), + signal: ctrl.signal, + }); + return { ok: res.ok, status: res.status }; + } finally { + clearTimeout(timer); + } + } catch { + return { ok: false }; // FR-007: never throw into the webhook handler + } +} diff --git a/services/cubejs/src/routes/deleteDataschema.js b/services/cubejs/src/routes/deleteDataschema.js index b56fa09c..0e9249d6 100644 --- a/services/cubejs/src/routes/deleteDataschema.js +++ b/services/cubejs/src/routes/deleteDataschema.js @@ -1,6 +1,7 @@ import YAML from "yaml"; import { verifyAndProvision } from "../utils/directVerifyAuth.js"; +import { emitModelEvent } from "../utils/eventEmitter.js"; import { findUser } from "../utils/dataSourceHelpers.js"; import { fetchGraphQL } from "../utils/graphql.js"; import { mintHasuraToken } from "../utils/mintHasuraToken.js"; @@ -373,5 +374,23 @@ export default async function deleteDataschema(req, res) { // Success path: the Hasura delete event trigger `delete_dataschema_audit` // writes the outcome='success' audit row. Handler does not duplicate. + + // 099 T087 (FR-091): a successful delete is a model lifecycle fact. + // Fire-and-forget; never blocks the response (FR-007). + emitModelEvent({ + event: "Model Deleted", + accountId: payload?.accountId ?? null, + partition: payload?.partition ?? null, + userId, + modelId: dataschemaId, + modelLabel: targetRow.name || null, + status: "ok", + properties: { + datasource_id: datasourceId, + branch_id: branchId, + version_id: version?.id ?? null, + }, + }); + return res.json({ deleted: true, dataschemaId }); } diff --git a/services/cubejs/src/routes/generateDataSchema.js b/services/cubejs/src/routes/generateDataSchema.js index 33dfd504..0f17783e 100644 --- a/services/cubejs/src/routes/generateDataSchema.js +++ b/services/cubejs/src/routes/generateDataSchema.js @@ -4,6 +4,7 @@ import { createDataSchema, findDataSchemas, } from "../utils/dataSourceHelpers.js"; +import { emitModelEvent } from "../utils/eventEmitter.js"; import createMd5Hex from "../utils/md5Hex.js"; import { NO_SCHEMA_KEY } from "./getSchema.js"; const camelize = (value) => @@ -93,6 +94,14 @@ export default async (req, res, cubejs) => { const { userScope, userId, authToken } = securityContext; const { dataSourceId } = userScope.dataSource; + // 099 T087 (FR-091): tenant attribution for the model lifecycle events. + const tokenPayload = securityContext.tokenPayload || {}; + const tenant = { + accountId: tokenPayload.accountId ?? null, + partition: tokenPayload.partition ?? null, + userId, + }; + let driver; try { @@ -175,14 +184,31 @@ export default async (req, res, cubejs) => { dataschemas: { data: [...preparedSchemas], }, + // Persistence chokepoint emits `Model Saved` for the created version. + emit: tenant, }; - await createDataSchema(commitObject); + const genResult = await createDataSchema(commitObject); if (cubejs.compilerCache) { cubejs.compilerCache.purgeStale(); } + // 099 T087 (FR-091): scaffolding persisted a model version. + // Fire-and-forget; never blocks the response (FR-007). + emitModelEvent({ + event: "Model Scaffolded", + ...tenant, + modelId: genResult?.id || null, + status: "ok", + properties: { + branch_id: branchId, + format, + file_count: files.length, + overwrite, + }, + }); + res.json({ code: "ok", message: "Generation finished" }); } catch (err) { console.error(err); diff --git a/services/cubejs/src/routes/loadExport.js b/services/cubejs/src/routes/loadExport.js index 4ea7548d..eac44977 100644 --- a/services/cubejs/src/routes/loadExport.js +++ b/services/cubejs/src/routes/loadExport.js @@ -23,6 +23,7 @@ import { writeBinaryChunk, writeRowStreamAsArrow, } from "../utils/arrowSerializer.js"; +import { emitQueryEvent } from "../utils/eventEmitter.js"; const prepareAnnotation = typeof prepareAnnotationModule.prepareAnnotation === "function" @@ -456,6 +457,37 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { const abortController = createAbortController(res); + // 099 T089 (FR-091): `Dataset Exported` audit. Tenant rides tokenPayload + // (populated by checkAuth, which the gateway checkAuth reuses on /load). Only + // the branches that actually stream a dataset flip `exported`; the finally + // emits `ok` for those, the catch emits `error`. Fire-and-forget, never + // blocks (FR-007), skips when no tenant. + const securityContext = + req.securityContext || plan.context?.securityContext || {}; + const tokenPayload = securityContext.tokenPayload || {}; + const exportTenant = { + accountId: tokenPayload.accountId ?? null, + partition: tokenPayload.partition ?? null, + userId: securityContext.userId ?? null, + }; + const exportDbType = securityContext.userScope?.dataSource?.dbType ?? null; + const exportStart = Date.now(); + let exported = false; + let exportPath = null; + const emitDatasetExported = (status, extra) => + emitQueryEvent({ + event: "Dataset Exported", + ...exportTenant, + status, + dimensions: exportDbType ? { datasource_type: exportDbType } : null, + metrics: { duration_ms: Date.now() - exportStart }, + properties: { + format, + ...(exportPath ? { export_path: exportPath } : {}), + ...(extra || {}), + }, + }); + try { let nativeQuery = null; @@ -488,6 +520,8 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { abortController.signal, getAliasNameToMember(plan) ); + exported = true; + exportPath = "native-clickhouse"; res.end(); return true; } @@ -505,6 +539,8 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { driver, abortController.signal ); + exported = true; + exportPath = "native-clickhouse"; res.end(); return true; } @@ -521,6 +557,8 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { columns: plan.columns, signal: abortController.signal, }); + exported = true; + exportPath = "semantic-stream"; res.end(); return true; } @@ -531,11 +569,22 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { annotation: plan.annotation, signal: abortController.signal, }); + exported = true; + exportPath = "semantic-stream"; res.end(); return true; } catch (err) { if (abortController.signal.aborted) return true; + // 099 T089: the export failed before completing — audit the error outcome. + // Guard on !exported so a post-stream throw can't double-emit (the finally + // already records the ok). + if (!exported) { + emitDatasetExported("error", { + error_message: err?.message || String(err), + }); + } + emitGatewayHandledError( plan.apiGateway, res, @@ -545,6 +594,9 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { plan.requestStarted ); return true; + } finally { + // 099 T089: a dataset was streamed to the client — audit the success once. + if (exported) emitDatasetExported("ok"); } } diff --git a/services/cubejs/src/routes/profileTable.js b/services/cubejs/src/routes/profileTable.js index 0d2af1c9..8c84beaa 100644 --- a/services/cubejs/src/routes/profileTable.js +++ b/services/cubejs/src/routes/profileTable.js @@ -8,6 +8,7 @@ import { serializeProfile } from '../utils/smart-generation/profileSerializer.js 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'; /** * Analyze an existing data schema file for user content and reprofile support. @@ -97,6 +98,19 @@ export default async (req, res, cubejs) => { const filters = Array.isArray(rawFilters) ? rawFilters : []; const nestedFilters = Array.isArray(rawNestedFilters) ? rawNestedFilters : []; + // 099 T089 (FR-091): tenant attribution for `Table Profiled`. NOTE: the + // event tenant partition lives on tokenPayload (FraiOS tenant) — NOT the + // `dataSource.partition` used below for ClickHouse row-scoping, which is a + // different concept. `dbType` rides the datasource_type dimension. + const tokenPayload = securityContext?.tokenPayload || {}; + const tenant = { + accountId: tokenPayload.accountId ?? null, + partition: tokenPayload.partition ?? null, + userId: securityContext?.userId ?? null, + }; + const dbType = securityContext?.userScope?.dataSource?.dbType ?? null; + const profileStart = Date.now(); + if (!table || !schema) { return res.status(400).json({ code: 'profile_table_missing_params', @@ -231,10 +245,51 @@ export default async (req, res, cubejs) => { raw_profile: rawProfile, }; + // 099 T089 (FR-091): the table was profiled. Fire-and-forget; never blocks + // (FR-007). No ABOUT subject id at profile time (the physical table is not a + // model version yet) — the house involves anchor it; schema/table ride + // properties, never dimensions (FR-031). + emitQueryEvent({ + event: 'Table Profiled', + ...tenant, + status: 'ok', + dimensions: dbType ? { datasource_type: dbType } : null, + metrics: { + duration_ms: Date.now() - profileStart, + ...(Number.isFinite(Number(profiledTable.row_count)) + ? { record_count: Number(profiledTable.row_count) } + : {}), + }, + properties: { + schema, + table, + sampled: profiledTable.sampled ?? null, + sample_size: profiledTable.sample_size ?? null, + column_count: columnsOutput.length, + ...(branchId ? { branch_id: branchId } : {}), + ...(partition ? { datasource_partition: partition } : {}), + }, + }); + emitter.complete(payload); } catch (err) { console.error(err); + // 099 T089 (FR-091): profiling failed — audit the error outcome. + emitQueryEvent({ + event: 'Table Profiled', + ...tenant, + status: 'error', + dimensions: dbType ? { datasource_type: dbType } : null, + metrics: { duration_ms: Date.now() - profileStart }, + properties: { + schema, + table, + ...(branchId ? { branch_id: branchId } : {}), + error_message: err?.message || String(err), + }, + }); + if (driver && driver.release) { await driver.release(); } diff --git a/services/cubejs/src/routes/reconcileTeam.js b/services/cubejs/src/routes/reconcileTeam.js index b64f5198..3a8c3665 100644 --- a/services/cubejs/src/routes/reconcileTeam.js +++ b/services/cubejs/src/routes/reconcileTeam.js @@ -2,6 +2,7 @@ import YAML from "yaml"; import { prepareCompiler } from "@cubejs-backend/schema-compiler"; import { verifyAndProvision } from "../utils/directVerifyAuth.js"; +import { emitModelEvent } from "../utils/eventEmitter.js"; import { fetchGraphQL } from "../utils/graphql.js"; import { createDataSchema } from "../utils/dataSourceHelpers.js"; import defineUserScope from "../utils/defineUserScope.js"; @@ -575,6 +576,14 @@ export default async function reconcileTeam(req, res, cubejs) { userScope, }; + // 099 T087 (FR-091): reconcile events are attributed to the TEAM being + // reconciled (its partition), performed by the default-models system user. + const reconcileTenant = { + accountId: verified.payload?.accountId ?? null, + partition, + userId: systemUserId, + }; + const previousDataschemas = branch.versions?.[0]?.dataschemas || []; const previousSchemaVersion = createMd5Hex( previousDataschemas.map((s) => s.id) @@ -645,6 +654,8 @@ export default async function reconcileTeam(req, res, cubejs) { datasource_id: datasourceId, })), }, + // Persistence chokepoint emits `Model Saved` for the created version. + emit: reconcileTenant, }); return { versionId: version?.id || null }; }, @@ -687,6 +698,25 @@ export default async function reconcileTeam(req, res, cubejs) { }); } + // 099 T087 (FR-091): the per-team default-models reconcile completed. + // Fire-and-forget; never blocks the response (FR-007). + emitModelEvent({ + event: "Default Models Reconciled", + ...reconcileTenant, + modelId: versionId || branchId, + status: "ok", + metrics: { record_count: outcomes.length }, + properties: { + team_id: teamId, + datasource_id: datasourceId, + branch_id: branchId, + version_id: versionId, + dry_run: dryRun, + outcomes_count: outcomes.length, + changed: Boolean(versionId), + }, + }); + return res.json({ teamId, outcomes, versionId }); } catch (err) { return res.status(500).json({ diff --git a/services/cubejs/src/routes/refreshCompiler.js b/services/cubejs/src/routes/refreshCompiler.js index 29a4a5d2..47f886f9 100644 --- a/services/cubejs/src/routes/refreshCompiler.js +++ b/services/cubejs/src/routes/refreshCompiler.js @@ -1,4 +1,5 @@ import { verifyAndProvision } from "../utils/directVerifyAuth.js"; +import { emitModelEvent } from "../utils/eventEmitter.js"; import { findUser } from "../utils/dataSourceHelpers.js"; import { resolvePartitionTeamIds } from "./discover.js"; import { requireOwnerOrAdmin } from "../utils/requireOwnerOrAdmin.js"; @@ -129,5 +130,18 @@ export default async function refreshCompiler(req, res, cubejs) { }) ); + // 099 T087 (FR-091): compiler-cache refresh affects every caller on the + // branch — record it as a model lifecycle fact. Fire-and-forget (FR-007). + emitModelEvent({ + event: "Schema Compiler Refreshed", + accountId: payload?.accountId ?? null, + partition: payload?.partition ?? null, + userId, + modelId: branchId, + status: "ok", + metrics: { record_count: evicted }, + properties: { branch_id: branchId, schema_version: schemaVersion, evicted }, + }); + return res.json({ evicted, schemaVersion }); } diff --git a/services/cubejs/src/routes/runSql.js b/services/cubejs/src/routes/runSql.js index d7ee2dbc..a90552b5 100644 --- a/services/cubejs/src/routes/runSql.js +++ b/services/cubejs/src/routes/runSql.js @@ -5,6 +5,7 @@ import { serializeRowsToArrow } from "../utils/arrowSerializer.js"; 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"; const { JWT_KEY } = process.env; @@ -137,6 +138,19 @@ async function writeBinaryChunk(writable, chunk, signal) { export default async (req, res, cubejs) => { const { securityContext } = req; + // 099 T089 (FR-091): tenant attribution for the security-grade `SQL Executed` + // audit event. This route bypasses query rewriting (freeform SQL), so its + // execution must be auditable. Tenant rides tokenPayload; datasource facts + // (id/dbType) come from the resolved user scope. + const tokenPayload = securityContext?.tokenPayload || {}; + const tenant = { + accountId: tokenPayload.accountId ?? null, + partition: tokenPayload.partition ?? null, + userId: securityContext?.userId ?? null, + }; + const dataSourceId = securityContext?.userScope?.dataSource?.dataSourceId ?? null; + const dbType = securityContext?.userScope?.dataSource?.dbType ?? null; + if (!req.body.query) { res.status(400).json({ code: "query_missing", @@ -146,15 +160,52 @@ export default async (req, res, cubejs) => { return; } + // `SQL Executed` audit bookkeeping — fire-and-forget, never blocks (FR-007). + // Emitted exactly once per request: `ok` when the response flushes cleanly + // (res `finish`), `error` from the catch below. Registered only AFTER the + // authorization gate passes, so blocked/short-circuited SQL is never audited + // as executed. + const auditStart = Date.now(); + let auditRowCount = null; + let auditFailed = false; + let audited = false; + let auditFormat = null; + let auditSigned = false; + const auditSqlExecuted = (status, extra) => { + if (audited) return; + audited = true; + const metrics = { duration_ms: Date.now() - auditStart }; + if (auditRowCount != null) metrics.record_count = auditRowCount; + emitQueryEvent({ + event: "SQL Executed", + ...tenant, + status, + dimensions: { + surface: "run-sql", + ...(dbType ? { datasource_type: dbType } : {}), + }, + metrics, + properties: { + ...(dataSourceId ? { datasource_id: dataSourceId } : {}), + ...(auditFormat ? { format: auditFormat } : {}), + signed: auditSigned, + ...(extra || {}), + }, + }); + }; + try { const format = validateFormat(req.body.format); + auditFormat = format; const sql = req.body.query; // Block freeform SQL when access control rules are active. // SQL that was generated by gen_sql is HMAC-signed — if the signature // verifies, the SQL went through the queryRewrite-governed pipeline // and is safe to execute regardless of access control rules. - if (!isSignedSql(sql, req.body.sql_signature)) { + const signed = isSignedSql(sql, req.body.sql_signature); + auditSigned = signed; + if (!signed) { const rules = await loadRules(); if (rules.length > 0) { res.status(403).json({ @@ -166,6 +217,11 @@ export default async (req, res, cubejs) => { } } + // Authorized to execute — audit the outcome once the response completes. + res.on("finish", () => { + if (!auditFailed) auditSqlExecuted("ok"); + }); + const driver = await cubejs.options.driverFactory({ securityContext }); // --- JSON (default): preserve original behavior --- @@ -179,6 +235,7 @@ export default async (req, res, cubejs) => { console.warn(`JSON response truncated: ${rows.length} rows exceeded safety limit of ${JSON_SAFETY_LIMIT}`); rows.length = JSON_SAFETY_LIMIT; } + auditRowCount = rows.length; res.json(rows); return; } @@ -198,6 +255,7 @@ export default async (req, res, cubejs) => { if (format === "jsonstat") { const { measures, timeDimensions } = req.body; const rows = await driver.query(sql); + auditRowCount = rows.length; const columns = deriveExportColumnsFromRunSql(req.body, rows); const dataset = buildJSONStat(rows, columns, { measures, timeDimensions }); @@ -278,6 +336,7 @@ export default async (req, res, cubejs) => { } const rows = await driver.query(sql); + auditRowCount = rows.length; if (format === "arrow") { // Generic (non-ClickHouse) Arrow path buffers all rows + 4x memory @@ -307,6 +366,10 @@ export default async (req, res, cubejs) => { } catch (err) { console.error(err); + // 099 T089 (FR-091): the bypass-path SQL failed — audit the error outcome. + auditFailed = true; + auditSqlExecuted("error", { error_message: err?.message || String(err) }); + if (res.headersSent) { if (!res.writableEnded) res.end(); return; diff --git a/services/cubejs/src/routes/smartGenerate.js b/services/cubejs/src/routes/smartGenerate.js index efed9446..9fce293e 100644 --- a/services/cubejs/src/routes/smartGenerate.js +++ b/services/cubejs/src/routes/smartGenerate.js @@ -2,6 +2,7 @@ import { createDataSchema, findDataSchemas, } from '../utils/dataSourceHelpers.js'; +import { emitModelEvent } from '../utils/eventEmitter.js'; import createMd5Hex from '../utils/md5Hex.js'; import { profileTable } from '../utils/smart-generation/profiler.js'; import { detectPrimaryKeys } from '../utils/smart-generation/primaryKeyDetector.js'; @@ -154,6 +155,16 @@ export default async (req, res, cubejs) => { try { const { userId } = securityContext; const partition = securityContext.userScope?.dataSource?.partition || null; + // 099 T087/T088 (FR-091): one tenant-attribution source of truth, shared by + // the LLM call sites (enrich/advise emit billable `Connection Called`) and + // the model lifecycle events (`Model Generated` / `Model Saved`). Hoisted so + // both run before and after the version-create chokepoint. + const tokenPayload = securityContext.tokenPayload || {}; + const tenant = { + accountId: tokenPayload.accountId ?? null, + partition: tokenPayload.partition ?? null, + userId, + }; let internalTables = securityContext.userScope?.dataSource?.internalTables || []; // 080: template-seeded generation targets the CANONICAL internal tables by // definition — force partition scoping for the target table even when the @@ -441,6 +452,9 @@ export default async (req, res, cubejs) => { existingMeasureNames, profilerFields: profilerFieldNames, profiledTableColumns: tableColumnNames, + // 099 T088 (FR-040/FR-091): attribute the billable `Connection Called` + // record emitted per OpenAI call to the operating tenant. + ...tenant, }, ); @@ -574,7 +588,9 @@ export default async (req, res, cubejs) => { }; try { - advisorResult = await adviseModel(generatedPreAdvise, profileSummaryForAdvisor, cubeResult.cubes); + // 099 T088 (FR-040/FR-091): tenant attribution for the per-pass billable + // `Connection Called` records emitted inside the advisory passes. + advisorResult = await adviseModel(generatedPreAdvise, profileSummaryForAdvisor, cubeResult.cubes, tenant); if (advisorResult.status === 'success' && advisorResult.passes.length > 0) { applyAdvisoryPasses(cubeResult.cubes, advisorResult.passes); @@ -975,6 +991,8 @@ export default async (req, res, cubejs) => { datasource_id: dataSourceId, })); + // 099 T087 (FR-091): `tenant` attribution is hoisted to the top of the + // handler (shared with the T088 LLM call sites). const result = await createDataSchema({ user_id: userId, branch_id: branchId, @@ -982,6 +1000,29 @@ export default async (req, res, cubejs) => { dataschemas: { data: [...preparedSchemas], }, + // Persistence chokepoint emits `Model Saved` for the created version. + emit: tenant, + }); + + // 099 T087 (FR-091): smart generation produced + saved a model version. + // Fire-and-forget; never blocks the response (FR-007). + emitModelEvent({ + event: 'Model Generated', + ...tenant, + modelId: result?.id || null, + modelLabel: fileName, + status: 'ok', + properties: { + file_name: fileName, + branch_id: branchId, + cubes_count: cubeResult.summary.cubes_count, + dimensions_count: cubeResult.summary.dimensions_count, + measures_count: cubeResult.summary.measures_count, + template_name: templateName, + skip_llm: skipLlm, + // 099 T088: the enrichment LLM used (null when skip_llm / no key). + llm_model: aiEnrichment.model, + }, }); // Purge compiler cache diff --git a/services/cubejs/src/routes/testConnection.js b/services/cubejs/src/routes/testConnection.js index ed5f1223..ac0efe2d 100644 --- a/services/cubejs/src/routes/testConnection.js +++ b/services/cubejs/src/routes/testConnection.js @@ -9,13 +9,45 @@ * * @throws {Error} - Throws an error if testing the connection fails. */ +import { emitQueryEvent } from "../utils/eventEmitter.js"; + export default async (req, res, cubejs) => { const { securityContext } = req; + // 099 T089 (FR-091): tenant + subject attribution for `Datasource Connection + // Tested`. Tenant rides tokenPayload; the datasource is the subject (a + // Connection) via ABOUT keyed by its id. + const tokenPayload = securityContext?.tokenPayload || {}; + const tenant = { + accountId: tokenPayload.accountId ?? null, + partition: tokenPayload.partition ?? null, + userId: securityContext?.userId ?? null, + }; + const dataSource = securityContext?.userScope?.dataSource; + const dataSourceId = dataSource?.dataSourceId ?? null; + const dbType = dataSource?.dbType ?? null; + const startedAt = Date.now(); + + const emitTested = (status, extra) => + emitQueryEvent({ + event: "Datasource Connection Tested", + ...tenant, + status, + about: dataSourceId + ? { entity_type: "Connection", id: dataSourceId } + : null, + dimensions: dbType ? { datasource_type: dbType } : null, + metrics: { duration_ms: Date.now() - startedAt }, + properties: extra || null, + }); + try { const driver = await cubejs.options.driverFactory({ securityContext }); await driver.testConnection(); + // Fire-and-forget; never blocks the response (FR-007). + emitTested("ok"); + res.json({ code: "ok", message: "Connection is OK", @@ -23,6 +55,8 @@ export default async (req, res, cubejs) => { } catch (err) { console.error(err); + emitTested("error", { error_message: err?.message || String(err) }); + res.status(500).json({ code: "test_connection_failed", message: err.message || err, diff --git a/services/cubejs/src/routes/validateInBranch.js b/services/cubejs/src/routes/validateInBranch.js index cf2f9e21..197c7bce 100644 --- a/services/cubejs/src/routes/validateInBranch.js +++ b/services/cubejs/src/routes/validateInBranch.js @@ -1,6 +1,7 @@ import { prepareCompiler } from "@cubejs-backend/schema-compiler"; import { verifyAndProvision } from "../utils/directVerifyAuth.js"; +import { emitModelEvent } from "../utils/eventEmitter.js"; import { findUser, findDataSchemas, @@ -364,6 +365,24 @@ export default async function validateInBranch(req, res) { } } + // 099 T087 (FR-091): the agent compile gate is a model lifecycle fact. + // Fire-and-forget; never blocks the validation response (FR-007). + emitModelEvent({ + event: "Model Draft Validated", + accountId: payload?.accountId ?? null, + partition: payload?.partition ?? null, + userId, + modelId: targetDataschemaId || branchId, + status: result.valid ? "ok" : "error", + properties: { + mode, + valid: result.valid, + error_count: result.errors.length, + warning_count: result.warnings.length, + branch_id: branchId, + }, + }); + return res.json(result); } catch (err) { return respondJson(res, 500, { diff --git a/services/cubejs/src/routes/versionRollback.js b/services/cubejs/src/routes/versionRollback.js index aeee62a8..05399247 100644 --- a/services/cubejs/src/routes/versionRollback.js +++ b/services/cubejs/src/routes/versionRollback.js @@ -1,4 +1,5 @@ import { verifyAndProvision } from "../utils/directVerifyAuth.js"; +import { emitModelEvent } from "../utils/eventEmitter.js"; import { findUser, findVersionBranch, @@ -195,6 +196,23 @@ export default async function versionRollback(req, res, cubejs) { ); } + // 099 T087 (FR-091): a successful rollback minted a new current version. + // Fire-and-forget; never blocks the response (FR-007). + emitModelEvent({ + event: "Model Version Rolled Back", + accountId: payload?.accountId ?? null, + partition: payload?.partition ?? null, + userId, + modelId: result.newVersionId, + status: "ok", + metrics: { record_count: result.clonedDataschemaCount }, + properties: { + branch_id: branchId, + to_version_id: toVersionId, + cloned_dataschema_count: result.clonedDataschemaCount, + }, + }); + return res.json({ newVersionId: result.newVersionId, clonedDataschemaCount: result.clonedDataschemaCount, diff --git a/services/cubejs/src/utils/checkAuth.js b/services/cubejs/src/utils/checkAuth.js index 3a6d638c..0e152df0 100644 --- a/services/cubejs/src/utils/checkAuth.js +++ b/services/cubejs/src/utils/checkAuth.js @@ -48,14 +48,24 @@ const checkAuth = async (req) => { let userId; const tokenType = detectTokenType(authToken); + // 099 T086 (FR-091): retain the FraiOS tenant attribution alongside the + // resolved userId so downstream semantic-event emitters can attribute events + // to the right tenant. Purely ADDITIVE — the auth validation below is + // unchanged; accountId/partition populate only on the token paths that carry + // them (FraiOS always; WorkOS partition when present), null otherwise. + const tokenPayload = { accountId: null, partition: null, tokenType }; + if (tokenType === "workos") { // WorkOS RS256 path const payload = await verifyWorkOSToken(authToken); userId = await provisionUserFromWorkOS(payload); + tokenPayload.partition = payload?.partition ?? null; } else if (tokenType === "fraios") { // FraiOS HS256 path const payload = await verifyFraiOSToken(authToken); userId = await provisionUserFromFraiOS(payload); + tokenPayload.accountId = payload?.accountId ?? null; + tokenPayload.partition = payload?.partition ?? null; } else { // Hasura HS256 path (existing) let jwtDecoded; @@ -120,6 +130,7 @@ const checkAuth = async (req) => { authToken, userId, userScope, + tokenPayload, }; }; diff --git a/services/cubejs/src/utils/checkSqlAuth.js b/services/cubejs/src/utils/checkSqlAuth.js index 43c3cab8..debb7963 100644 --- a/services/cubejs/src/utils/checkSqlAuth.js +++ b/services/cubejs/src/utils/checkSqlAuth.js @@ -10,6 +10,7 @@ import buildSecurityContext from "./buildSecurityContext.js"; import defineUserScope, { getDataSourceAccessList, } from "./defineUserScope.js"; +import { emitQueryEvent } from "./eventEmitter.js"; const buildSqlSecurityContext = (sqlCredentials) => { if (!sqlCredentials) { @@ -116,6 +117,19 @@ const checkSqlAuth = async (request, userArg, passwordArg) => { datasourceId ); + // 099 T089 (FR-091): a SQL-API session was authenticated. WorkOS tokens + // carry a partition (tenant) but no accountId. Fire-and-forget; never + // throws / never blocks the login (FR-007). Skips when no tenant. + emitQueryEvent({ + event: "SQL Session Opened", + accountId: null, + partition: payload?.partition ?? null, + userId, + status: "ok", + dimensions: { surface: "sql-api", credential_source: "workos" }, + properties: { datasource_id: datasourceId }, + }); + return { password, securityContext: { @@ -155,6 +169,19 @@ const checkSqlAuth = async (request, userArg, passwordArg) => { datasourceId ); + // 099 T089 (FR-091): a SQL-API session was authenticated. FraiOS tokens + // carry both accountId and partition. Fire-and-forget; never throws / + // never blocks the login (FR-007). Skips when no tenant. + emitQueryEvent({ + event: "SQL Session Opened", + accountId: payload?.accountId ?? null, + partition: payload?.partition ?? null, + userId, + status: "ok", + dimensions: { surface: "sql-api", credential_source: "fraios" }, + properties: { datasource_id: datasourceId }, + }); + return { password, securityContext: { diff --git a/services/cubejs/src/utils/dataSourceHelpers.js b/services/cubejs/src/utils/dataSourceHelpers.js index cc5b08f6..ae887eea 100644 --- a/services/cubejs/src/utils/dataSourceHelpers.js +++ b/services/cubejs/src/utils/dataSourceHelpers.js @@ -2,6 +2,7 @@ import { createHash } from "crypto"; import { fetchGraphQL } from "./graphql.js"; import { fetchWorkOSUserProfile } from "./workosAuth.js"; +import { emitModelEvent } from "./eventEmitter.js"; // --- User scope cache: keyed by userId, 30s TTL --- const userCache = new Map(); @@ -284,7 +285,10 @@ export const getDataSources = async () => { }; export const createDataSchema = async (object) => { - const { authToken, ...version } = object; + // `emit` (optional) carries tenant attribution for the 099 T087 `Model Saved` + // lifecycle event — destructured OUT here alongside `authToken` so it never + // reaches the `versions_insert_input` mutation variable (unknown column). + const { authToken, emit, ...version } = object; let res = await fetchGraphQL( upsertVersionMutation, @@ -293,6 +297,24 @@ export const createDataSchema = async (object) => { ); res = res?.data?.insert_versions_one; + // 099 T087 (FR-091): this is THE server-side persistence chokepoint for a + // model version. Emit `Model Saved` (persistence fact) when a caller supplied + // tenant attribution and a version was actually created. Fire-and-forget; + // never throws / never blocks (FR-007). The pure editor save (raw GraphQL + // straight to Hasura) does not pass through here — it is covered by a Hasura + // event trigger (T087, tables.yaml), a separate mechanism. + if (emit && res?.id) { + emitModelEvent({ + event: "Model Saved", + accountId: emit.accountId ?? null, + partition: emit.partition ?? null, + userId: emit.userId ?? null, + modelId: res.id, + status: "ok", + properties: { branch_id: version.branch_id ?? null, origin: version.origin ?? "save" }, + }); + } + return res; }; diff --git a/services/cubejs/src/utils/eventEmitter.js b/services/cubejs/src/utils/eventEmitter.js new file mode 100644 index 00000000..f1ec5e1c --- /dev/null +++ b/services/cubejs/src/utils/eventEmitter.js @@ -0,0 +1,816 @@ +import fetch from "node-fetch"; +import { SignJWT } from "jose"; +import { v5 as uuidv5 } from "uuid"; +import { hostname } from "os"; + +/** + * eventEmitter — never-throw, fire-and-forget semantic-event emitter for + * synmetrix (099 T086, FR-091). This is the SUBSTRATE only: it builds canonical + * envelopes and POSTs them to the FraiOS ingression endpoint. It is NOT yet + * wired into any handler — that is T087–T090 (llmEnricher / modelAdvisor emit + * `Connection Called` per OpenAI call). + * + * Contract mirror: modelled on `auditWriter.js` — best-effort with N attempts + + * exponential backoff, and NEVER throws / never blocks the caller's response + * (FR-007). Every failure branch drops a single structured stderr line as a + * last-resort observation and returns `{ ok: false }`. + * + * Credential (FR-074 / A4): synmetrix is a background/service caller. If the + * caller forwards its own token (`emitSemanticEvent(env, { token })`) that is + * used verbatim as the ingression `writekey`. Otherwise a short-lived + * FraiOS-shaped service token is minted from the shared `TOKEN_SECRET` + * (HS256 via `jose`, mirroring `mintHasuraToken.js`) so the record still bills + * to the operating tenant. A credential is never invented to fill a gap: with + * no forwarded token and no `TOKEN_SECRET`, the event is skipped + counted. + * + * Endpoint: `{INGRESSION_HOST}/api/s/{envelope.type||'log'}` — `INGRESSION_HOST` + * defaults to the FraiOS inbox (matches ai-service `config.py`). + */ + +const { TOKEN_SECRET, INGRESSION_HOST, CXS_EVENT_SOURCE, CXS_ENVIRONMENT } = + process.env; + +// Retry policy — mirrors auditWriter.js (3 attempts, 50ms initial backoff). +const MAX_ATTEMPTS = 3; +const INITIAL_BACKOFF_MS = 50; + +// FraiOS inbox default (parity with ai-service `INGRESSION_HOST`). +const DEFAULT_INGRESSION_HOST = "https://inbox.fraios.dev"; + +// Short-lived minted service token (minutes) + its provider marker so the +// ingression side can attribute the writer. +const SERVICE_TOKEN_TTL_MIN = 5; +const SERVICE_PROVIDER = "synmetrix-service"; + +// Origin tagging — the shared vocabulary tags `source` = producer identity. +const DEFAULT_SOURCE = "synmetrix"; + +// Canonical DNS namespace (== Python `uuid.NAMESPACE_DNS`) so message_id / +// event_gid are byte-identical to the other producers' deterministic ids. +const DNS_NAMESPACE = "6ba7b810-9dad-11d1-80b4-00c04fd430c8"; + +// Issuing system for every involve id_type (required field, always "FraiOS"). +const ID_TYPE = "FraiOS"; + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/* ------------------------------------------------------------------ * + * entity_gid derivation — ported verbatim from the normative shared + * implementation (fraios `apps/rule-engine/src/events/entity-gid.js`, + * pinned by `shared/schemas/events/entity-gid-vectors.json`). A well-formed + * RFC-4122 UUID passes through UNCHANGED (original case preserved); any other + * stable id — including "" — maps via four parallel 32-bit FNV-1a lanes over + * UTF-16 code units, forced to a v4-shaped UUID. This is NOT uuid5 and must not + * be replaced with uuid5 (that would silently split the identity space). + * ------------------------------------------------------------------ */ + +const UUID_REGEX = + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +/** Loose UUID shape — matches Python `_is_uuid` (used only to decide whether to + * also keep the raw `id` on an involve entry). */ +const UUID_LOOSE_REGEX = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +/** + * Four parallel 32-bit FNV-1a lanes over the input's UTF-16 code units. + * `Math.imul(...) >>> 0` is load-bearing at EVERY step — it is what keeps each + * lane a wrapping 32-bit multiply. + */ +function hashToBytes(input) { + let h1 = 0x811c9dc5; + let h2 = 0x811c9dc5 ^ 0xdeadbeef; + let h3 = 0x811c9dc5 ^ 0x41c6ce57; + let h4 = 0x811c9dc5 ^ 0x9e3779b9; + for (let i = 0; i < input.length; i += 1) { + const c = input.charCodeAt(i); + h1 = Math.imul(h1 ^ c, 0x01000193) >>> 0; + h2 = Math.imul(h2 ^ c, 0x01000193) >>> 0; + h3 = Math.imul(h3 ^ c, 0x01000193) >>> 0; + h4 = Math.imul(h4 ^ c, 0x01000193) >>> 0; + } + const bytes = new Uint8Array(16); + const write = (h, offset) => { + bytes[offset] = (h >>> 24) & 0xff; + bytes[offset + 1] = (h >>> 16) & 0xff; + bytes[offset + 2] = (h >>> 8) & 0xff; + bytes[offset + 3] = h & 0xff; + }; + write(h1, 0); + write(h2, 4); + write(h3, 8); + write(h4, 12); + return bytes; +} + +function hashToUuid(input) { + const bytes = hashToBytes(input); + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + const toHex = (b) => b.toString(16).padStart(2, "0"); + const hex = Array.from(bytes, toHex).join(""); + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice( + 12, + 16 + )}-${hex.slice(16, 20)}-${hex.slice(20)}`; +} + +/** + * Canonical graph UUID for an entity, placed in `involves[].entity_gid` and the + * envelope `entity_gid`. RFC-4122 passthrough (original case), else hashToUuid. + * + * Callers must reject an EMPTY *account* identifier before emitting: "" derives + * to a well-formed UUID, which would file every credential-less event under one + * shared synthetic tenant. + * + * @param {string} raw the identifier as minted + * @returns {string} a canonical UUID + */ +export function normalizeEntityGid(raw) { + const s = String(raw ?? ""); + return UUID_REGEX.test(s) ? s : hashToUuid(s); +} + +/* ------------------------------------------------------------------ * + * Envelope helpers — snake_case SemanticEvent conventions, ported from the + * canonical shared builder (fraios `libs/python/fraios-core/.../envelope.py`). + * ------------------------------------------------------------------ */ + +/** Deterministic uuid5 message_id from stable parts joined by ":". */ +export function messageIdFor(...parts) { + const name = parts.map((p) => String(p ?? "")).join(":"); + return uuidv5(name, DNS_NAMESPACE); +} + +/** Deterministic, replay-stable event_gid derived from the message_id. */ +function eventGidFor(messageId) { + return uuidv5(String(messageId).toLowerCase(), DNS_NAMESPACE); +} + +/** (source, environment, host) for origin tagging on every event. */ +function sourceInfo() { + const source = (CXS_EVENT_SOURCE || "").trim() || DEFAULT_SOURCE; + let env = (CXS_ENVIRONMENT || "").trim(); + if (!env) env = process.env.KUBERNETES_SERVICE_HOST ? "cluster" : "local"; + const host = (process.env.HOSTNAME || "").trim() || hostname(); + return { source, environment: env, host }; +} + +/** + * One involve entry (fraios id model): graph UUID in `entity_gid`; a non-UUID + * source id also kept in `id`; `id_type` always "FraiOS". + * + * @param {string} role + * @param {string} entity_type + * @param {string} value the entity's stable id as minted + * @param {string} [label] + */ +export function involve(role, entity_type, value, label) { + const entry = { + role, + entity_type, + entity_gid: normalizeEntityGid(value), + id_type: ID_TYPE, + }; + const s = String(value ?? ""); + if (s && !UUID_LOOSE_REGEX.test(s)) entry.id = s; + if (label) entry.label = label; + return entry; +} + +/** + * Canonical snake_case SemanticEvent skeleton. Origin info on every event + * (`source`; `environment` in dimensions; `producer_host`/`producer_env` in + * properties); deterministic `event_gid`. + */ +export function buildEnvelope({ + event, + abstract_event, + message_id, + timestamp, + partition, + entity_gid, + involves, + type = "log", + customer_facing = 0, + dimensions = null, + metrics = null, + analysis = null, + properties = null, +}) { + const { source, environment, host } = sourceInfo(); + const dims = { ...(dimensions || {}), environment }; + const props = { + ...(properties || {}), + producer_host: host, + producer_env: environment, + }; + const envelope = { + type, + event, + abstract_event, + message_id, + event_gid: eventGidFor(message_id), + timestamp, + partition, + entity_gid: normalizeEntityGid(entity_gid), + customer_facing, + source, + dimensions: dims, + involves, + properties: props, + }; + if (metrics) envelope.metrics = metrics; + if (analysis) envelope.analysis = analysis; + return envelope; +} + +/* ------------------------------------------------------------------ * + * Connection Called — the billable record (contract: + * specs/099-semantic-events/contracts/billable-record.md). Exactly one per + * provider/vendor call; ungated; `amount` + `currency` ALWAYS present. + * ------------------------------------------------------------------ */ + +/** + * Build the compliant `Connection Called` envelope for a single provider call. + * + * `amount` + `currency` are ALWAYS present (contract: absence is a violation). + * A free/priced call ships `amount, pricing: "known"`; an unpriced model ships + * `amount: 0.0, pricing: "unknown"`. + * + * @param {object} args + * @param {string} args.partition tenant partition (required) + * @param {string} args.accountId OWNED_BY account id as minted (required) + * @param {string} [args.userId] REQUESTED_BY person id, when a human is attributable + * @param {string} args.provider vendor (e.g. "openai") + * @param {string} [args.model] model / variant id + * @param {string} args.item `feature:operation` slug (never a vendor name) + * @param {number} [args.durationMs] wall-clock call duration + * @param {number} [args.cost] USD cost; null/undefined ⇒ unpriced ⇒ amount 0.0 + pricing "unknown" + * @param {string} [args.connectionId] USES_CONNECTION connection id, when known + * @param {"ok"|"error"} [args.status] call outcome (default "ok") + * @param {string} [args.timestamp] ISO timestamp (default now) + * @returns {object} canonical Connection Called envelope + */ +export function buildConnectionCalled({ + partition, + accountId, + userId = null, + provider, + model = null, + item, + durationMs = null, + cost = null, + connectionId = null, + status = "ok", + timestamp = null, +} = {}) { + const ts = timestamp || new Date().toISOString(); + const priced = cost != null && Number.isFinite(Number(cost)); + const amount = priced ? Number(cost) : 0.0; // ALWAYS present + const currency = "USD"; // ALWAYS present + const pricing = priced ? "known" : "unknown"; + + // Owner id for the envelope entity_gid / OWNED_BY — fall back to partition so + // a record is never filed under the derived-empty synthetic tenant. + const ownerId = String(accountId ?? "") || String(partition ?? ""); + + // Deterministic-ish id: tenant + event + provider/model/item + status + ts. + // (Handlers T087–T090 should prefer the provider call id when available.) + const message_id = messageIdFor( + partition, + "Connection Called", + provider, + model, + item, + status, + ts + ); + + const involves = [involve("OWNED_BY", "Account", ownerId)]; + if (userId) involves.push(involve("REQUESTED_BY", "Person", String(userId))); + if (connectionId) + involves.push(involve("USES_CONNECTION", "Connection", String(connectionId))); + + const dimensions = { provider, status, item }; + if (model) dimensions.model = model; + + const metrics = {}; + if (durationMs != null && Number.isFinite(Number(durationMs))) { + metrics.duration_ms = Number(durationMs); + } + + const analysisEntry = { + item, + provider, + variant: model, + amount, + currency, + }; + if (durationMs != null && Number.isFinite(Number(durationMs))) { + analysisEntry.processing_time = Number(durationMs) / 1000; + } + + return buildEnvelope({ + event: "Connection Called", + abstract_event: "Connection Called", + message_id, + timestamp: ts, + partition: String(partition ?? ""), + entity_gid: ownerId, + involves, + type: "log", + customer_facing: 0, + dimensions, + metrics: Object.keys(metrics).length ? metrics : null, + analysis: [analysisEntry], + // FR-043 "unknown pricing" marker rides the top-level `properties` JSON + // column — the analysis[] Nested has fixed subcolumns and the persister + // drops unknown keys like `extras` (review P1-4). + properties: { pricing }, + }); +} + +/** + * Build + fire-and-forget a billable `Connection Called` record for a single + * provider call. + * + * Fully guarded like {@link emitModelEvent}: NEVER throws (FR-007) and NEVER + * blocks the caller — the ingression POST is detached, so a slow inbox can + * never delay the provider call's own path. Skips silently when there is no + * tenant to attribute (neither accountId nor partition): a credential-less + * record would file under the derived-empty synthetic tenant, and a credential + * is never invented to fill the gap (A4). + * + * `properties` (optional) is merged onto the envelope's free-form properties + * slot — the smart-generation callers (099 T088) use it to carry `attempts` / + * `pass` context so a silent LLM degradation stays auditable. The billing + * contract itself lives in dimensions/analysis (amount+currency always) and is + * untouched by this merge. + * + * @param {object} args same shape as {@link buildConnectionCalled}, plus: + * @param {object} [args.properties] free-form context merged onto the envelope + */ +export function emitConnectionCalled(args = {}) { + try { + const { + accountId = null, + partition = null, + userId = null, + properties = null, + } = args; + if (!accountId && !partition) return; // no tenant → skip, never invent one + const envelope = buildConnectionCalled(args); + if (properties && typeof properties === "object") { + // Producer proof is IMMUTABLE: re-stamp it AFTER merging caller-supplied + // properties so a caller can never override it (review P1-6). + const builtProducer = envelope.properties?.producer; + envelope.properties = { ...(envelope.properties || {}), ...properties }; + if (builtProducer != null) envelope.properties.producer = builtProducer; + } + // emitSemanticEvent is itself never-throw; the detached .catch is belt-and- + // braces so an unexpected rejection can never surface as unhandled. + emitSemanticEvent(envelope, { accountId, partition, userId }).catch( + () => {} + ); + } catch { + // absolute never-throw guard (FR-007) + } +} + +/* ------------------------------------------------------------------ * + * Model-management lifecycle events (099 T087, FR-091). Business-class, + * category `lifecycle`, `type: "track"`. Every model event carries the + * canonical involves grammar: + * OWNED_BY / Account — the owning tenant (from tokenPayload.accountId, + * falling back to the partition so a record is never + * filed under the derived-empty synthetic tenant), + * ACTED_BY / Person — the human who performed the action, when known, + * ABOUT / Data Model — the dataschema / version / branch the event concerns. + * Dimensions stay inside the declared key dictionary (`status` + the + * auto-added `environment`); ids NEVER appear in dimensions (FR-031) — counts, + * modes and other free-form context ride the un-keyed `properties` slot. + * ------------------------------------------------------------------ */ + +/** + * Build a canonical model-management lifecycle envelope. + * + * @param {object} args + * @param {string} args.event past-tense event name (e.g. "Model Saved") + * @param {string} args.partition tenant partition (required for attribution) + * @param {string} [args.accountId] OWNED_BY account id as minted (id_type FraiOS) + * @param {string} [args.userId] ACTED_BY person id, when a human is attributable + * @param {string} args.modelId ABOUT Data Model id (dataschema/version/branch) + * @param {string} [args.modelLabel] optional human label for the model + * @param {"ok"|"error"} [args.status] outcome (default "ok") + * @param {object} [args.dimensions] extra LowCardinality dimensions (declared keys only) + * @param {object} [args.metrics] declared metric keys (e.g. record_count) + * @param {object} [args.properties] free-form context (counts, modes, ids-of-record) + * @param {string} [args.timestamp] ISO timestamp (default now) + * @returns {object} canonical SemanticEvent envelope + */ +export function buildModelEvent({ + event, + partition, + accountId = null, + userId = null, + modelId, + modelLabel = null, + status = "ok", + dimensions = null, + metrics = null, + properties = null, + timestamp = null, +} = {}) { + const ts = timestamp || new Date().toISOString(); + // Owner id for OWNED_BY / entity_gid — fall back to partition so the record + // is never filed under the derived-empty synthetic tenant (normalizeEntityGid). + const ownerId = String(accountId ?? "") || String(partition ?? ""); + const aboutId = String(modelId ?? "") || ownerId; + + const message_id = messageIdFor(partition, event, aboutId, status, ts); + + const involves = [involve("OWNED_BY", "Account", ownerId)]; + if (userId) involves.push(involve("ACTED_BY", "Person", String(userId))); + involves.push(involve("ABOUT", "Data Model", aboutId, modelLabel)); + + return buildEnvelope({ + event, + abstract_event: event, + message_id, + timestamp: ts, + partition: String(partition ?? ""), + entity_gid: aboutId, + involves, + type: "track", + customer_facing: 0, + dimensions: { status, ...(dimensions || {}) }, + metrics, + properties, + }); +} + +/** + * Build + fire-and-forget a model-management lifecycle event. + * + * Fully guarded: NEVER throws (FR-007) and NEVER blocks the caller — the + * ingression POST is detached, so a slow inbox can never delay the model + * operation's response. Skips silently when there is no tenant to attribute + * (neither accountId nor partition): a credential-less event would otherwise + * file under the derived-empty synthetic tenant, and a credential is never + * invented to fill the gap (A4). + * + * @param {object} args same shape as {@link buildModelEvent} + */ +export function emitModelEvent(args = {}) { + try { + const { accountId = null, partition = null, userId = null } = args; + if (!accountId && !partition) return; // no tenant → skip, never invent one + const envelope = buildModelEvent(args); + // emitSemanticEvent is itself never-throw; the detached .catch is belt-and- + // braces so an unexpected rejection can never surface as unhandled. + emitSemanticEvent(envelope, { accountId, partition, userId }).catch( + () => {} + ); + } catch { + // absolute never-throw guard (FR-007) + } +} + +/* ------------------------------------------------------------------ * + * Query-shoulder + connection-test events (099 T089, US7 / FR-091). + * `type: "track"`. These close the query/execution/session/export/profiling/ + * connection-test coverage gaps. Same canonical grammar as the model events: + * OWNED_BY / Account — the owning tenant (tokenPayload.accountId, id_type + * FraiOS; falls back to the partition so a record is + * never filed under the derived-empty synthetic tenant), + * ACTED_BY / Person — the human who performed the action, when known, + * ABOUT / — OPTIONAL subject, using ONLY a vocab entity_type + * (e.g. "Connection" for a datasource). Omitted where no + * first-class subject id exists — the house involves + * (OWNED_BY + ACTED_BY) still anchor the event. + * Dimensions stay inside the declared key dictionary (`status` + the auto-added + * `environment`, plus any caller-declared keys); ids NEVER appear in dimensions + * (FR-031) — they ride ABOUT / the free-form `properties` slot. + * ------------------------------------------------------------------ */ + +/** + * Build a canonical query-shoulder / connection-test envelope. + * + * @param {object} args + * @param {string} args.event past-tense event name (e.g. "SQL Executed") + * @param {string} args.partition tenant partition (required for attribution) + * @param {string} [args.accountId] OWNED_BY account id as minted (id_type FraiOS) + * @param {string} [args.userId] ACTED_BY person id, when a human is attributable + * @param {object} [args.about] OPTIONAL subject: { entity_type, id, label } — entity_type + * must be a vocab kind; omitted when id is null/absent + * @param {"ok"|"error"} [args.status] outcome (default "ok") + * @param {string} [args.type] envelope type (default "track") + * @param {object} [args.dimensions] extra LowCardinality dimensions (declared keys only) + * @param {object} [args.metrics] declared metric keys (e.g. duration_ms, record_count) + * @param {object} [args.properties] free-form context (schema/table/format/ids-of-record) + * @param {string} [args.timestamp] ISO timestamp (default now) + * @returns {object} canonical SemanticEvent envelope + */ +export function buildQueryEvent({ + event, + partition, + accountId = null, + userId = null, + about = null, + status = "ok", + type = "track", + customer_facing = 0, + dimensions = null, + metrics = null, + properties = null, + timestamp = null, +} = {}) { + const ts = timestamp || new Date().toISOString(); + // Owner id for OWNED_BY / entity_gid — fall back to partition so the record + // is never filed under the derived-empty synthetic tenant (normalizeEntityGid). + const ownerId = String(accountId ?? "") || String(partition ?? ""); + + const aboutId = + about && about.id != null && String(about.id) !== "" + ? String(about.id) + : null; + const aboutType = aboutId ? String(about.entity_type ?? "") : null; + // entity_gid anchors on the subject when there is one, else on the tenant. + const entityGid = aboutId || ownerId; + + const message_id = messageIdFor( + partition, + event, + aboutId ?? "", + status, + ts + ); + + const involves = [involve("OWNED_BY", "Account", ownerId)]; + if (userId) involves.push(involve("ACTED_BY", "Person", String(userId))); + if (aboutId && aboutType) { + involves.push(involve("ABOUT", aboutType, aboutId, about.label)); + } + + return buildEnvelope({ + event, + abstract_event: event, + message_id, + timestamp: ts, + partition: String(partition ?? ""), + entity_gid: entityGid, + involves, + type, + customer_facing, + dimensions: { status, ...(dimensions || {}) }, + metrics, + properties, + }); +} + +/** + * Build + fire-and-forget a query-shoulder / connection-test event. + * + * Fully guarded like {@link emitModelEvent}: NEVER throws (FR-007) and NEVER + * blocks the caller — the ingression POST is detached. Skips silently when there + * is no tenant to attribute (neither accountId nor partition): a credential-less + * event would file under the derived-empty synthetic tenant, and a credential is + * never invented to fill the gap (A4). Credential is service-minted (the + * registry declares `credential: service-minted`); the caller token is NOT + * forwarded, mirroring the model-event emitter. + * + * @param {object} args same shape as {@link buildQueryEvent} + */ +export function emitQueryEvent(args = {}) { + try { + const { accountId = null, partition = null, userId = null } = args; + if (!accountId && !partition) return; // no tenant → skip, never invent one + const envelope = buildQueryEvent(args); + // emitSemanticEvent is itself never-throw; the detached .catch is belt-and- + // braces so an unexpected rejection can never surface as unhandled. + emitSemanticEvent(envelope, { accountId, partition, userId }).catch( + () => {} + ); + } catch { + // absolute never-throw guard (FR-007) + } +} + +/* ------------------------------------------------------------------ * + * Buffered query-log emission (type='log'). The query path only ENQUEUES + * (O(1), no network, never throws); a detached background flusher batches the + * ingression POSTs OFF the query path, so semantic-event creation/emission can + * NEVER affect query performance. The buffer is bounded (drops the oldest under + * backpressure) and the flush timer is unref'd (never keeps the process alive). + * ------------------------------------------------------------------ */ + +const LOG_BUFFER_MAX = Number(process.env.LOG_EVENT_BUFFER_MAX || 2000); +const LOG_FLUSH_MS = Number(process.env.LOG_EVENT_FLUSH_MS || 2000); +const LOG_FLUSH_BATCH = Number(process.env.LOG_EVENT_FLUSH_BATCH || 200); +const _logBuffer = []; +let _logFlusher = null; + +function _drainLogBuffer() { + const batch = _logBuffer.splice(0, LOG_FLUSH_BATCH); + for (const item of batch) { + // detached — a slow inbox can never delay the query path + emitSemanticEvent(item.envelope, item.ctx).catch(() => {}); + } +} + +function _startLogFlusher() { + if (_logFlusher) return; + _logFlusher = setInterval(_drainLogBuffer, LOG_FLUSH_MS); + if (_logFlusher && _logFlusher.unref) _logFlusher.unref(); +} + +/** + * Enqueue a pre-built log envelope for buffered, off-path delivery. Synchronous, + * O(1), never throws. Drops the oldest buffered event once the bound is hit so a + * slow/unreachable inbox can never grow memory without limit. + */ +export function enqueueLogEvent(envelope, ctx = {}) { + try { + if (_logBuffer.length >= LOG_BUFFER_MAX) _logBuffer.shift(); + _logBuffer.push({ envelope, ctx }); + _startLogFlusher(); + } catch { + // never throw into the caller (the cube query logger) + } +} + +/** + * Build + BUFFER-emit a `Query Executed` log event (type='log') for one completed + * cube analytical query. Called from the cube `logger` hook (src/utils/logging.js); + * it ENQUEUES only, so query performance is never affected. Skips silently when + * there is no tenant to attribute (A4 — a credential is never invented). + * + * @param {object} args same shape as {@link buildQueryEvent} (minus event/type) + */ +export function emitQueryLog(args = {}) { + try { + const { accountId = null, partition = null, userId = null } = args; + if (!accountId && !partition) return; // no tenant → skip, never invent one + const envelope = buildQueryEvent({ + ...args, + event: "Query Executed", + type: "log", + }); + enqueueLogEvent(envelope, { accountId, partition, userId }); + } catch { + // absolute never-throw guard (FR-007) + } +} + +/* ------------------------------------------------------------------ * + * Credential + transport. + * ------------------------------------------------------------------ */ + +/** + * Mint a short-lived FraiOS-shaped service token (HS256 over `TOKEN_SECRET`), + * mirroring `mintHasuraToken.js`. Carries `accountId`/`partition`/`userId` so + * the ingression side attributes the write to the operating tenant. + * + * Never throws. Returns the signed JWT, or `null` when no `TOKEN_SECRET` is + * configured (the caller then skips + counts — a credential is never invented). + * + * @returns {Promise} + */ +export async function mintServiceToken({ + accountId = null, + partition = null, + userId = null, +} = {}) { + try { + if (!TOKEN_SECRET) return null; + const secret = new TextEncoder().encode(TOKEN_SECRET); + const claims = { provider: SERVICE_PROVIDER }; + if (accountId != null) claims.accountId = String(accountId); + if (partition != null) claims.partition = String(partition); + if (userId != null) claims.userId = String(userId); + return await new SignJWT(claims) + .setProtectedHeader({ alg: "HS256" }) + .setIssuedAt() + .setIssuer("services:cubejs") + .setAudience("fraios:ingression") + .setExpirationTime(`${SERVICE_TOKEN_TTL_MIN}m`) + .sign(secret); + } catch { + return null; + } +} + +function logSkip(reason, envelope, extra = {}) { + try { + console.error( + JSON.stringify({ + level: "error", + event: "semantic_event_emit_failed", + reason, + event_name: envelope?.event ?? null, + partition: envelope?.partition ?? null, + message_id: envelope?.message_id ?? null, + ...extra, + ts: new Date().toISOString(), + }) + ); + } catch { + // stderr must never itself throw — swallow. + } +} + +/** + * emitSemanticEvent — best-effort, fire-and-forget POST of a canonical envelope + * to the ingression endpoint. NEVER throws (FR-007); the caller is never + * informed of a transport failure — observability bookkeeping must not block a + * user response. + * + * `POST {INGRESSION_HOST}/api/s/{envelope.type||'log'}` with header + * `writekey: `. Best-effort with 3 attempts + exponential backoff + * (mirrors auditWriter.js). On exhausted attempts a structured stderr line is + * emitted and `{ ok: false }` returned. + * + * Credential: `opts.token` (forwarded caller token) is used verbatim; otherwise + * a short-lived service token is minted from `TOKEN_SECRET` using the envelope's + * tenant (or `opts.accountId`/`opts.partition`/`opts.userId`). + * + * @param {object} envelope canonical SemanticEvent (e.g. from buildConnectionCalled) + * @param {object} [opts] + * @param {string} [opts.token] forwarded caller token → used as writekey + * @param {string} [opts.accountId] tenant account id for the minted token + * @param {string} [opts.partition] tenant partition for the minted token + * @param {string} [opts.userId] person id for the minted token + * @returns {Promise<{ok: true, status: number} | {ok: false}>} + */ +export async function emitSemanticEvent( + envelope, + { token = null, accountId = null, partition = null, userId = null } = {} +) { + try { + if (!envelope || typeof envelope !== "object") { + logSkip("missing_envelope", envelope); + return { ok: false }; + } + + const ownerInvolve = Array.isArray(envelope.involves) + ? envelope.involves.find((i) => i?.role === "OWNED_BY") + : null; + + let writekey = token; + if (!writekey) { + writekey = await mintServiceToken({ + accountId: accountId ?? ownerInvolve?.id ?? null, + partition: partition ?? envelope.partition ?? null, + userId, + }); + } + if (!writekey) { + // A4: no forwarded credential and none can be minted — skip + count, + // never invent one. + logSkip("no_writekey", envelope); + return { ok: false }; + } + + const host = (INGRESSION_HOST || DEFAULT_INGRESSION_HOST).replace( + /\/+$/, + "" + ); + const type = envelope.type || "log"; + const url = `${host}/api/s/${encodeURIComponent(type)}`; + const body = JSON.stringify(envelope); + + let lastErr; + for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt += 1) { + try { + const res = await fetch(url, { + method: "POST", + headers: { + "content-type": "application/json", + writekey, + }, + body, + }); + if (res.ok) return { ok: true, status: res.status }; + lastErr = new Error(`ingression responded ${res.status}`); + } catch (err) { + lastErr = err; + } + if (attempt < MAX_ATTEMPTS) { + await sleep(INITIAL_BACKOFF_MS * 2 ** (attempt - 1)); + } + } + + logSkip("transport_failed", envelope, { + detail: lastErr?.message || "unknown", + }); + return { ok: false }; + } catch (err) { + // Absolute never-throw guard — even an unexpected failure (bad env, JSON + // serialisation, etc.) must not propagate to the caller. + logSkip("unexpected", envelope, { detail: err?.message || "unknown" }); + return { ok: false }; + } +} diff --git a/services/cubejs/src/utils/logging.js b/services/cubejs/src/utils/logging.js index 924e79a4..7d22b18f 100644 --- a/services/cubejs/src/utils/logging.js +++ b/services/cubejs/src/utils/logging.js @@ -1,4 +1,5 @@ import { devLogger } from "@cubejs-backend/server-core/dist/src/core/logger.js"; +import { emitQueryLog } from "./eventEmitter.js"; import redisClient from "./redis.js"; /** @@ -43,9 +44,35 @@ export const logging = async (message, event) => { data.timestamp = new Date().toISOString(); if (data?.securityContext) { - data.userId = data.securityContext?.userId; - data.dataSourceId = - data.securityContext?.userScope?.dataSource?.dataSourceId; + const sc = data.securityContext; + data.userId = sc?.userId; + data.dataSourceId = sc?.userScope?.dataSource?.dataSourceId; + + // 099 FR-091: mirror a COMPLETED cube analytical query into a buffered + // `type='log'` `Query Executed` semantic event. ENQUEUE-ONLY — never awaited, + // never throws — so query performance is untouched; a background flusher does + // the ingression POSTs off the query path (see eventEmitter.emitQueryLog). + if (message === "Load Request Success") { + const ds = sc?.userScope?.dataSource; + emitQueryLog({ + accountId: sc?.accountId ?? null, + partition: sc?.partition ?? null, + userId: sc?.userId ?? null, + status: "ok", + dimensions: { + surface: "load", + ...(ds?.dbType ? { datasource_type: ds.dbType } : {}), + }, + metrics: Number.isFinite(Number(data?.duration)) + ? { duration_ms: Number(data.duration) } + : {}, + properties: { + ...(data.dataSourceId ? { datasource_id: String(data.dataSourceId) } : {}), + ...(requestId ? { request_id: String(requestId) } : {}), + ...(data.path ? { path: String(data.path) } : {}), + }, + }); + } delete data.securityContext; } diff --git a/services/cubejs/src/utils/smart-generation/llmEnricher.js b/services/cubejs/src/utils/smart-generation/llmEnricher.js index 1e3c4486..0fa51bd7 100644 --- a/services/cubejs/src/utils/smart-generation/llmEnricher.js +++ b/services/cubejs/src/utils/smart-generation/llmEnricher.js @@ -10,6 +10,7 @@ import { z } from 'zod'; import { validateAIMetrics } from './llmValidator.js'; +import { emitConnectionCalled } from '../eventEmitter.js'; // --------------------------------------------------------------------------- // Constants @@ -265,7 +266,9 @@ function buildCorrectionPrompt(rejected) { * @param {object} profiledTable - { table, database, columns (Map), row_count } * @param {object[]} existingCubes - Cube definitions from cubeBuilder * @param {object[]} existingAIMetrics - Previously generated AI metrics - * @param {object} [options] - { timeout, existingMeasureNames, profilerFields, profiledTableColumns } + * @param {object} [options] - { timeout, existingMeasureNames, profilerFields, profiledTableColumns, accountId, partition, userId } + * accountId/partition/userId (099 T088) attribute the billable `Connection + * Called` record emitted per OpenAI call; absent tenant ⇒ the emit self-skips. * @returns {Promise<{ metrics: object[], status: string, model: string, error: string|null, rejected?: Array<{ metric: object, reasons: string[] }> }>} */ export async function enrichWithAIMetrics( @@ -311,15 +314,42 @@ export async function enrichWithAIMetrics( let allValid = []; let lastRejected = []; + // 099 T088 (FR-040/FR-091): emit one billable `Connection Called` per OpenAI + // call. cost=null ⇒ compliant unpriced fallback (amount 0.0 / USD / pricing + // unknown). Fire-and-forget + never-throw; `attempts` (1-based) rides + // properties so the previously-silent LLM path stays auditable — including + // on failure, where the record is still emitted with status="error". + const emitEnrichCall = (status, startedAt, attempt) => + emitConnectionCalled({ + partition: options.partition, + accountId: options.accountId, + userId: options.userId, + provider: 'openai', + model: MODEL, + item: 'smart-generation:enrich', + durationMs: Date.now() - startedAt, + cost: null, + status, + properties: { attempts: attempt + 1 }, + }); + for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { - const completion = await client.chat.completions.parse( - { - model: MODEL, - messages, - response_format: zodResponseFormat(ResponseSchema, 'ai_metrics'), - }, - { signal: AbortSignal.timeout(timeout) } - ); + const startedAt = Date.now(); + let completion; + try { + completion = await client.chat.completions.parse( + { + model: MODEL, + messages, + response_format: zodResponseFormat(ResponseSchema, 'ai_metrics'), + }, + { signal: AbortSignal.timeout(timeout) } + ); + } catch (callErr) { + emitEnrichCall('error', startedAt, attempt); + throw callErr; // preserve behavior — outer catch records result.error + } + emitEnrichCall('ok', startedAt, attempt); const parsed = completion.choices[0].message.parsed; const metricsToValidate = parsed.metrics; diff --git a/services/cubejs/src/utils/smart-generation/modelAdvisor.js b/services/cubejs/src/utils/smart-generation/modelAdvisor.js index 141dee0a..98547c34 100644 --- a/services/cubejs/src/utils/smart-generation/modelAdvisor.js +++ b/services/cubejs/src/utils/smart-generation/modelAdvisor.js @@ -9,6 +9,7 @@ import fs from 'fs'; import { validateModelSyntax } from './modelValidator.js'; import { generateJs } from './yamlGenerator.js'; +import { emitConnectionCalled } from '../eventEmitter.js'; const MODEL = 'gpt-5.4'; const PASS_TIMEOUT = 45_000; // 45s per micro-prompt (plenty for small output) @@ -59,7 +60,7 @@ function buildModelContext(generatedCode, profileSummary, cubes) { // -- Individual passes -------------------------------------------------------- -async function runPass(client, zodResponseFormat, z, modelContext, passName, principles, question, schema, timeout) { +async function runPass(client, zodResponseFormat, z, modelContext, passName, principles, question, schema, timeout, emitCtx) { const systemPrompt = [ 'You are a Cube.js data modeling advisor. You review auto-generated models and suggest improvements.', 'You have deep expertise in Cube.js, ClickHouse, and semantic layer best practices.', @@ -74,6 +75,24 @@ async function runPass(client, zodResponseFormat, z, modelContext, passName, pri '- Return ONLY what needs changing. Empty arrays are fine if no changes needed.', ].join('\n'); + // 099 T088 (FR-040/FR-091): emit one billable `Connection Called` per OpenAI + // call (one per advisory pass). cost=null ⇒ compliant unpriced fallback + // (amount 0.0 / USD / pricing unknown). Fire-and-forget + never-throw; the + // pass name rides properties so a silently-swallowed pass failure stays + // auditable — the record is still emitted with status="error". + const emitAdviseCall = (status, startedAt) => + emitConnectionCalled({ + ...(emitCtx || {}), + provider: 'openai', + model: MODEL, + item: 'smart-generation:advise', + durationMs: Date.now() - startedAt, + cost: null, + status, + properties: { pass: passName, attempts: 1 }, + }); + + const startedAt = Date.now(); try { const completion = await client.chat.completions.parse( { @@ -88,8 +107,10 @@ async function runPass(client, zodResponseFormat, z, modelContext, passName, pri { signal: AbortSignal.timeout(timeout) } ); + emitAdviseCall('ok', startedAt); return completion.choices[0]?.message?.parsed || null; } catch (err) { + emitAdviseCall('error', startedAt); console.warn('[modelAdvisor] Pass "' + passName + '" failed: ' + err.message); return null; } @@ -103,7 +124,10 @@ async function runPass(client, zodResponseFormat, z, modelContext, passName, pri * @param {string} generatedCode - JS model code * @param {object} profileSummary - { table, schema, row_count, columns } * @param {object[]} cubes - Parsed cube definitions - * @param {object} [options] + * @param {object} [options] - { timeout, accountId, partition, userId } + * accountId/partition/userId (099 T088) attribute the billable `Connection + * Called` record emitted per advisory pass; absent tenant ⇒ the emit + * self-skips (a credential is never invented). * @returns {Promise<{ passes: object[], status: string, error: string|null }>} */ export async function adviseModel(generatedCode, profileSummary, cubes, options = {}) { @@ -119,6 +143,12 @@ export async function adviseModel(generatedCode, profileSummary, cubes, options const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY }); const modelContext = buildModelContext(generatedCode, profileSummary, cubes); + // 099 T088: tenant attribution threaded to each per-pass `Connection Called`. + const emitCtx = { + accountId: options.accountId ?? null, + partition: options.partition ?? null, + userId: options.userId ?? null, + }; const passes = []; // -- Pass 1: Descriptions & Titles -- @@ -134,7 +164,7 @@ export async function adviseModel(generatedCode, profileSummary, cubes, options const descResult = await runPass(client, zodResponseFormat, z, modelContext, 'descriptions', PRINCIPLES_DESCRIPTIONS, 'Review all dimensions and measures. For any with a poor title or missing description, provide corrections. Focus on titles that would confuse a non-technical analyst.', - DescSchema, timeout); + DescSchema, timeout, emitCtx); if (descResult) passes.push({ pass: 'descriptions', result: descResult }); // -- Pass 2: Segments -- @@ -150,7 +180,7 @@ export async function adviseModel(generatedCode, profileSummary, cubes, options const segResult = await runPass(client, zodResponseFormat, z, modelContext, 'segments', PRINCIPLES_SEGMENTS, 'What meaningful analyst-facing segments should this cube have? Consider common filter patterns analysts would reuse. Return 0-5 segments.', - SegSchema, timeout); + SegSchema, timeout, emitCtx); if (segResult) passes.push({ pass: 'segments', result: segResult }); // -- Pass 3: Derived Metrics -- @@ -168,7 +198,7 @@ export async function adviseModel(generatedCode, profileSummary, cubes, options const metricsResult = await runPass(client, zodResponseFormat, z, modelContext, 'derived_metrics', PRINCIPLES_METRICS, 'What calculated metrics (rates, ratios, decomposed averages) would support analysis of this data? Reference existing measures using {measure_name} syntax. Return 0-10 metrics.', - MetricsSchema, timeout); + MetricsSchema, timeout, emitCtx); if (metricsResult) passes.push({ pass: 'derived_metrics', result: metricsResult }); // -- Pass 4: Pre-aggregation Review -- @@ -188,7 +218,7 @@ export async function adviseModel(generatedCode, profileSummary, cubes, options const preAggResult = await runPass(client, zodResponseFormat, z, modelContext, 'pre_aggregations', PRINCIPLES_PREAGGS, 'Review the pre-aggregations. Are they appropriate for this data and likely dashboard patterns? Return the complete recommended set (it replaces existing pre-aggs).', - PreAggSchema, timeout); + PreAggSchema, timeout, emitCtx); if (preAggResult) passes.push({ pass: 'pre_aggregations', result: preAggResult }); return { diff --git a/services/hasura/metadata/tables.yaml b/services/hasura/metadata/tables.yaml index f4edc72c..92ee7e60 100644 --- a/services/hasura/metadata/tables.yaml +++ b/services/hasura/metadata/tables.yaml @@ -426,6 +426,31 @@ _in: - owner - admin + # 099 Semantic Events (US7, FR-091, T087): emit Branch Created / Branch Deleted + # lifecycle events. Raw-GraphQL branch mutations have no JS chokepoint, so this + # event trigger is the emission point. Handler resolves the tenant (partition == + # team.settings.partition) from the branch's datasource → team and posts to the + # ingress; emission is fire-and-forget and never fails the mutation (FR-007). + event_triggers: + - name: emit_branch_lifecycle + definition: + enable_manual: false + insert: + columns: '*' + delete: + columns: '*' + retry_conf: + interval_sec: 10 + num_retries: 3 + timeout_sec: 60 + webhook: '{{ACTIONS_URL}}/rpc/emit_branch_lifecycle' + cleanup_config: + batch_size: 10000 + clean_invocation_logs: false + clear_older_than: 168 + paused: false + schedule: 0 0 * * * + timeout: 60 - table: name: dashboards schema: public @@ -1377,6 +1402,31 @@ members: user_id: _eq: X-Hasura-User-Id + # 099 Semantic Events (US7, FR-091, T089): emit SQL Credential Created / Deleted. + # Raw-GraphQL credential mutations have no JS chokepoint. Handler resolves the + # tenant (partition == team.settings.partition) from the credential's datasource + # → team and posts to the ingress; fire-and-forget, never fails the mutation + # (FR-007). Only the credential id + username ride the event — never the secret. + event_triggers: + - name: emit_sql_credential_lifecycle + definition: + enable_manual: false + insert: + columns: '*' + delete: + columns: '*' + retry_conf: + interval_sec: 10 + num_retries: 3 + timeout_sec: 60 + webhook: '{{ACTIONS_URL}}/rpc/emit_sql_credential_lifecycle' + cleanup_config: + batch_size: 10000 + clean_invocation_logs: false + clear_older_than: 168 + paused: false + schedule: 0 0 * * * + timeout: 60 - table: name: query_rewrite_rules schema: public @@ -1733,6 +1783,29 @@ paused: false schedule: 0 0 * * * timeout: 60 + # 099 Semantic Events (US7, FR-091, T087): emit Model Version Created on every + # versions INSERT (the editor's pure-save + other raw-GraphQL version creations + # have no JS chokepoint). Handler resolves the tenant (partition == + # team.settings.partition) from branch → datasource → team and posts to the + # ingress; fire-and-forget, never fails the mutation (FR-007). Distinct from + # version_rollback_audit (DB audit) — this emits the semantic lifecycle event. + - name: emit_model_version_created + definition: + enable_manual: false + insert: + columns: '*' + retry_conf: + interval_sec: 10 + num_retries: 3 + timeout_sec: 60 + webhook: '{{ACTIONS_URL}}/rpc/emit_model_version_created' + cleanup_config: + batch_size: 10000 + clean_invocation_logs: false + clear_older_than: 168 + paused: false + schedule: 0 0 * * * + timeout: 60 - table: name: audit_logs schema: public