Subchapter 45.3

Bi Sink

Number: 45.3
Position: 3 of 5
Type: JavaScript
Size: 12 KB
Lines: 273

Nearby

Source

lib/bi-sink.js

JavaScript273 lines12 KB

1'use strict';
2
3// Wix BI sink for the telemetry recorder — the one transport path.
4//
5// Wire format (proven end-to-end 2026-07-26):
6// POST https://frog.wix.com/migration-data-validator with a JSON batch envelope
7//   {"dt":0,"g":{"src":10,"evid":<evid>},"e":[{"dt":0,"f":{...fields}}]}
8// capped at 100 events / ~240KB per request. No transport auth — identity is the
9// claim-based `logged_user_id` field, routing every row to events.dbo.users_10.
10//
11// Three hard rules, each a silent-drop trap (2xx with no row, or NULL columns):
12//   1. Schema fields go up in camelCase (`runId`, `errorType`, ...) — the catcher
13//      matches registered param names, not the snake_case Trino column names.
14//      Standard identity fields stay snake_case (`logged_user_id`).

bi-sink.js on GitHub

// 2. Every GUID-typed field must be a real GUID; a non-hex value fails

// validation before routing and drops the whole event.

// 3. Every row carries `logged_user_id` = the operator's wix_user_id (GUID).

//

// frog's HTTP status is not an ingestion receipt — only a Trino read-back proves

// arrival. The recorder treats any non-2xx or network error as a push failure

// (journaled into telemetry_health); a 2xx merely means "handed to frog".

//

// Env switches (tests / offline dev only — production runs always emit):

// RP_TELEMETRY_BI_ENDPOINT override the frog endpoint (contract-test mock)

// RP_TELEMETRY_BI_DISABLED=1 skip all pushes entirely

const SRC = 10;

const EVID_RUN = 5012; // replatform_run — run-grain lifecycle, phase started|finalized

const EVID_RUN_EVENT = 5013; // replatform_run_event — one row per observation class

const DEFAULT_ENDPOINT = 'https://frog.wix.com/migration-data-validator';

const MAX_EVENTS_PER_REQUEST = 100;

const MAX_BODY_BYTES = 240 * 1024;

const PUSH_TIMEOUT_MS = 8000;

const GUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

function isGuid(value) {

return typeof value === 'string' && GUID_RE.test(value);

}

function disabled() {

return process.env.RP_TELEMETRY_BI_DISABLED === '1';

}

function endpoint() {

return process.env.RP_TELEMETRY_BI_ENDPOINT || DEFAULT_ENDPOINT;

}

// Empty-value convention: unknown fields are omitted, never sent as null/"".

// JSON-typed schema fields (arrays/objects) are JSON-serialized into their value.

// GUID-typed fields are dropped unless they hold a real GUID (hard rule 2 —

// sending a non-GUID would silently drop the whole event, not just the field).

function put(fields, key, value, { json = false, guid = false } = {}) {

if (value === undefined || value === null) return;

if (guid && !isGuid(value)) return;

fields[key] = json ? JSON.stringify(value) : value;

}

// `replatform_run` phase:started — built from the journal's run_start record, so

// its content is a pure function of run start and re-pushes are byte-identical

// (dedup key: (run_id, phase), latest row wins). `wixUserId` is the one late

// addition: identity may arrive via a later dims call, and the row is only

// sendable once it has (rule 3), so it rides along when known.

function startedRunRow(runStart, wixUserId) {

const dims = runStart.dims || {};

const f = { phase: 'started' };

put(f, 'runId', runStart.run_id, { guid: true });

put(f, 'projectId', runStart.project_id);

put(f, 'attempt', runStart.attempt);

put(f, 'sessionCount', 1);

put(f, 'schemaVersions', runStart.schema_version);

put(f, 'skillsVersion', runStart.skills_version);

put(f, 'runtimeEnv', runStart.runtime_env, { json: true });

put(f, 'wixUserId', wixUserId, { guid: true });

put(f, 'sourcePlatform', dims.source_platform);

put(f, 'sourcePlatformVersion', dims.source_platform_version);

put(f, 'sourceExtensions', dims.source_extensions, { json: true });

put(f, 'sourceSiteUrl', dims.source_site_url);

put(f, 'sourceAcquisition', dims.source_acquisition);

put(f, 'deliveryMode', dims.delivery_mode);

put(f, 'destinationStrategy', dims.destination_strategy);

put(f, 'runStarted', runStart.ts);

return { evid: EVID_RUN, fields: f };

}

// `replatform_run` phase:finalized — the complete rollup. Local names map to the

// as-registered BI names here and nowhere else (see the Registration notes):

// schema_version→schemaVersions, verification→verificationType. skills_commit and

// discovered_entity_types are local-only (not registered BI columns) and

// deliberately not sent — the entity-class demand signal reaches BI through the

// `volumes` column; the dim exists for the finalize cross-check, whose flags

// travel in telemetryHealth.

function finalizedRunRow(rollup) {

const f = { phase: 'finalized' };

put(f, 'runId', rollup.run_id, { guid: true });

put(f, 'projectId', rollup.project_id);

put(f, 'attempt', rollup.attempt);

put(f, 'sessionCount', rollup.session_count);

put(f, 'schemaVersions', rollup.schema_version);

put(f, 'skillsVersion', rollup.skills_version);

put(f, 'runtimeEnv', rollup.runtime_env, { json: true });

put(f, 'wixUserId', rollup.wix_user_id, { guid: true });

put(f, 'siteId', rollup.site_id, { guid: true });

put(f, 'sourcePlatform', rollup.source_platform);

put(f, 'sourcePlatformVersion', rollup.source_platform_version);

put(f, 'sourceExtensions', rollup.source_extensions, { json: true });

put(f, 'sourceSiteUrl', rollup.source_site_url);

put(f, 'sourceAcquisition', rollup.source_acquisition);

put(f, 'deliveryMode', rollup.delivery_mode);

put(f, 'destinationStrategy', rollup.destination_strategy);

put(f, 'terminalState', rollup.terminal_state);

put(f, 'stoppedAtStage', rollup.stopped_at_stage);

put(f, 'stages', rollup.stages, { json: true });

put(f, 'volumes', rollup.volumes, { json: true });

put(f, 'verificationType', rollup.verification, { json: true });

put(f, 'operatorAcceptance', rollup.operator_acceptance);

put(f, 'telemetryHealth', rollup.telemetry_health, { json: true });

put(f, 'eventCountByType', rollup.event_count_by_type, { json: true });

put(f, 'activeMs', rollup.active_ms);

put(f, 'waitingMs', rollup.waiting_ms);

put(f, 'runStarted', rollup.run_started);

put(f, 'runEnded', rollup.run_ended);

put(f, 'bundleManifest', rollup.bundle_manifest, { json: true });

// spec 0039 §5: the transcript_digest is a run-grain fact, so it rides the

// existing finalized row rather than opening a new BI call — one JSON field,

// matching how `stages`/`volumes` already travel, rather than one registered

// column per sub-field. A run with no digest (offline recompute skipped, or

// a run that predates 1.5.0) simply omits the field, same as any unknown.

//

// Known limitation (verified live, spec 0039 §5): as of 2026-08-17, `bi_events_raw`

// shows evid 5012 rows with an EMPTY dimensions/measures map for every one of them —

// this field will land alongside `terminalState`/`volumes`/every other field already

// on this row and be equally unqueryable until whoever owns the BI schema resolves

// why 5012/5013 aren't materializing their payload. Nothing to fix here: this `put()`

// call matches the pattern every other field on the row already uses correctly.

put(f, 'transcriptDigest', rollup.transcript_digest, { json: true });

return { evid: EVID_RUN, fields: f };

}

// `replatform_run_event` — one row per folded observation class. Re-pushing a

// class after more occurrences folded in is by design: the dedup key is

// (run_id, seq) and the latest row carries the cumulative count. Renames:

// event_type→eventTypeName, error_code→errorType, expected/actual→

// expectedSkill/actualSkill.

function eventRow(folded) {

const f = {};

put(f, 'runId', folded.run_id, { guid: true });

put(f, 'seq', folded.seq);

put(f, 'count', folded.count);

put(f, 'stage', folded.stage);

put(f, 'skill', folded.skill);

put(f, 'eventTypeName', folded.event_type);

put(f, 'subtype', folded.subtype);

put(f, 'entityType', folded.entity_type);

put(f, 'severity', folded.severity);

put(f, 'wixApiSurface', folded.wix_api_surface);

put(f, 'sourceApiSurface', folded.source_api_surface);

put(f, 'wixAppId', folded.wix_app_id, { guid: true });

put(f, 'errorType', folded.error_code);

put(f, 'decisionPoint', folded.decision_point);

put(f, 'retryCount', folded.retry_count);

put(f, 'recovered', folded.recovered);

put(f, 'whatHappened', folded.what_happened);

put(f, 'expectedSkill', folded.expected);

put(f, 'actualSkill', folded.actual);

put(f, 'observedShapes', folded.observed_shapes, { json: true });

put(f, 'shapesSeen', folded.shapes_seen);

put(f, 'evidenceRefs', folded.evidence_refs, { json: true });

return { evid: EVID_RUN_EVENT, fields: f };

}

function envelopeBody(evid, events) {

return JSON.stringify({ dt: 0, g: { src: SRC, evid }, e: events });

}

// A single event over the body cap sheds its unbounded-ish JSON fields before

// being declared unsendable. Truncation is surfaced (telemetry_health flag),

// never silent.

const SHEDDABLE_FIELDS = ['observedShapes', 'evidenceRefs', 'bundleManifest', 'stages', 'volumes'];

function shedOversize(event) {

const f = { ...event.f };

let shed = false;

for (const key of SHEDDABLE_FIELDS) {

if (f[key] === undefined) continue;

delete f[key];

shed = true;

}

return shed ? { ...event, f } : null;

}

async function sendBatch(evid, events) {

const body = envelopeBody(evid, events);

try {

const res = await fetch(endpoint(), {

method: 'POST',

headers: { 'content-type': 'application/json' },

body,

signal: AbortSignal.timeout(PUSH_TIMEOUT_MS),

});

if (!res.ok) return `http_${res.status}`;

return null;

} catch (error) {

return error && error.name === 'TimeoutError' ? 'timeout' : 'network_error';

}

// Push rows (from the builders above) as the given operator. Never throws:

// returns { sent, failures: [{evid, rows, detail}], truncated, skipped } and

// the caller journals the failures — telemetry must never block a migration.

// A disabled sink reports the rows it swallowed in `skipped` instead of

// masquerading as a successful push — the caller journals those too, so a

// kill-switch leaking into a production session is visible, never silent.

async function pushRows(rows, loggedUserId) {

const out = { sent: 0, failures: [], truncated: 0, skipped: 0 };

if (rows.length === 0) return out;

if (disabled()) {

out.skipped = rows.length;

return out;

}

if (!isGuid(loggedUserId)) {

out.failures.push({ evid: null, rows: rows.length, detail: 'no_operator_identity' });

return out;

}

const byEvid = new Map();

for (const row of rows) {

if (!byEvid.has(row.evid)) byEvid.set(row.evid, []);

byEvid.get(row.evid).push({ dt: 0, f: { ...row.fields, logged_user_id: loggedUserId } });

}

for (const [evid, events] of byEvid) {

const queue = [events];

while (queue.length > 0) {

let chunk = queue.shift();

if (chunk.length > MAX_EVENTS_PER_REQUEST) {

const mid = Math.ceil(chunk.length / 2);

queue.unshift(chunk.slice(0, mid), chunk.slice(mid));

continue;

}

if (Buffer.byteLength(envelopeBody(evid, chunk)) > MAX_BODY_BYTES) {

if (chunk.length > 1) {

const mid = Math.ceil(chunk.length / 2);

queue.unshift(chunk.slice(0, mid), chunk.slice(mid));

continue;

}

const slimmed = shedOversize(chunk[0]);

if (!slimmed || Buffer.byteLength(envelopeBody(evid, [slimmed])) > MAX_BODY_BYTES) {

out.failures.push({ evid, rows: 1, detail: 'oversized_event' });

continue;

}

out.truncated += 1;

chunk = [slimmed];

}

const failureDetail = await sendBatch(evid, chunk);

if (failureDetail) out.failures.push({ evid, rows: chunk.length, detail: failureDetail });

else out.sent += chunk.length;

}

return out;

}

module.exports = {

SRC,

EVID_RUN,

EVID_RUN_EVENT,

MAX_EVENTS_PER_REQUEST,

MAX_BODY_BYTES,

isGuid,

disabled,

startedRunRow,

finalizedRunRow,

eventRow,

pushRows,

};