The Cartographer
What It Is
The Cartographer is a runnable demo: a real browser-executed DAG application, not a decorative diagram. It is a deterministic data-orchestration pipeline powered by Dagonizer: multi-source fan-in, branching conditional routing, offline country-coder geo-resolution, GDPR redaction, and continent insights. It uses the same engine as the Archivist, applied to ETL instead of LLM agents.
Use it to see data-pipeline work stay inspectable, resumable, and honest about skipped work. The graph shows which branches run, which branches are skipped, and which embedded DAGs own each transformation.
How It Works
The runner wires real node classes, real DAG documents, and browser UI observers together. The visual panes listen to dispatcher lifecycle events, so the page shows execution rather than replaying a canned animation.
Architecture
One top-level scatter and a tree of per-type embedded pipeline DAGs:
cartographer (top-level)
entrypoints: position-ping | facility-scan | sensor-reading | customs-event | delivery-confirmation
gather('intake-gather', source-intake) ← multi-input source intake; writes merged state.sources stream
scatter('process-stream', 'sources', ← one run of stream-event per source payload
{ dag: 'stream-event' },
gather: insights-fold, ← O(1) fold into state.insights / state.journeys / state.sampleRecords
container: 'cpu', ← browser demo: WebWorkerContainer role
execution: { mode: 'reservoir', concurrency: 16, reservoir: { keyField: 'eventType', capacity } })
└─ stream-event ← decode-payload → route-event-type-variant
├─ position-ping ──► pipeline-position-ping (parse → geo-pipeline → enrich-leg → aggregate)
├─ sensor-reading ──► pipeline-sensor-reading (parse → geo-pipeline → cold-chain → enrich-leg → aggregate)
├─ customs-event ──► pipeline-customs-event (parse → geo-pipeline → customs-dwell → enrich-leg → aggregate)
├─ facility-scan ──► pipeline-facility-scan (parse → geo-pipeline → canonicalize-facility
│ → order-enrichment → gdpr-compliance → aggregate)
└─ delivery-confirmation ► pipeline-delivery-confirmation (parse → geo-pipeline → canonicalize-recipient
→ confirm-delivery → gdpr-compliance → aggregate)
Each per-type pipeline embeds:
geo-pipeline ← route-geo → validate-coords → geo-source-resolve (six embedded resolver DAG entrypoints → geo-weighted-fusion GatherNode) | apply-geo
gdpr-compliance ← consent-gate → classify-pii → redact-pii
embed('summarize-insights', 'insights-summary',
container: 'io') ← browser demo: separate WebWorkerContainer role
└─ insights-summary
summarize → done
doneThe insights-fold gather accumulates each clone's state.enriched into three bounded accumulators (state.insights, state.journeys, state.sampleRecords) as clones complete. Memory is O(1) regardless of event count — the parent state never holds a full copy of every record at once.
The browser demo runs the process-stream scatter body through container role cpu and the summarize-insights embedded DAG through container role io. CartographerWorkerContainer extends WebWorkerContainer to spawn a statically-bundled worker entry so Vite can chunk the registry. The reservoir capacity is a UI-controlled knob: the runner calls CartographerWorkersDag.bundle(clampedBatchCapacity) on each run so the batch size tracks the visitor's setting without mutating shared constants.
Diagrams, Examples, and Outputs
The live demo is the main diagram. Its graph, state panes, traces, memory views, backend selectors, and outputs are all evidence from the running system.
What this proves
The Cartographer proves Dagonizer is not only an agent framework. The same JSON-LD DAG model, scatter/gather machinery, embedded DAGs, worker containers, checkpoint semantics, and visualization surfaces run deterministic ETL/data-orchestration workloads in the browser.
It runs on the same @studnicky/dagonizer engine as The Archivist. Only the node domain differs: agent reasoning vs data enrichment. The DAG topology, lifecycle hooks, observer pattern, streaming scatter, and embedded-DAG composition are identical.
Try it live below. Click Run to stream 21 synthetic tracking events (the default: 6 position-pings, 5 facility-scans, 4 sensor-readings, 3 customs-events, 3 delivery-confirmations) through the full pipeline. Watch the DAG pane: nodes light cyan while executing, edges flash when traversed, and branching skips are visible as edges that never fire.
Watch the Panels tab after the run: the before/after panel shows raw GPS coordinates resolved to a real continent/country, and raw PII fields redacted to their pseudonymised forms. The routing savings table shows how many node executions the conditional branching avoided.
What It Lets You Do
Use the Cartographer when you want to see Dagonizer without an LLM anywhere in the loop. It is a streaming data pipeline with typed inputs, bounded fan-out, conditional routing, worker-backed processing, and aggregate outputs.
For application teams, this page answers a different practical question than the Archivist: can the same graph engine handle ETL-shaped work with real branching, backpressure, and data-quality decisions? Yes, and the panels show the skipped work as clearly as the completed work.
What to try
Click Run and watch the stream, DAG, and panels while synthetic shipment events move through parsing, geo-resolution, GDPR compliance, worker-backed stream processing, and insight aggregation. Compare the routing-savings table with the highlighted DAG path.
Code Samples
The Cartographer source shows the data-pipeline side of the same engine. Start with the top-level DAGs, then inspect the routing nodes, state/services, entity shapes, and CLI runner that make the browser demo deterministic.
Branching conditional routing
Each per-type pipeline DAG routes the event only through the nodes it needs. Two skip conditions are the headline:
route-geo: a position-ping that already carries resolved geo (country, continent, region from the JSON source) routes toapply-geoand never enters thegeo-source-resolvesub-DAG. The entire source-model geo lookup — offline coords/locale/code resolution and the IP geolocation network call — is skipped.route-redaction: an event with no PII fields, or one whose consent/jurisdiction does not require processing, routes toskip-redactionand never enters thegdpr-compliancesub-DAG.
Each routing node records its decision on the clone's state.routing object (a EnrichedShipment.routing value). The parent delegates summarize-insights to the insights-summary DAG, which folds these across all records to produce the savings tally when the streaming gather has not already produced bounded aggregates.
ts
export class RouteGeoNode extends MonadicNode<CartographerState, 'has-geo' | 'needs-geo'> {
readonly '@id' = 'urn:noocodec:node:route-geo';
readonly 'name' = 'route-geo';
readonly 'outputs' = ['has-geo', 'needs-geo'] as const;
override get outputSchema(): Record<'has-geo' | 'needs-geo', SchemaObjectType> {
return {
'has-geo': { 'type': 'object' },
'needs-geo': { 'type': 'object' },
};
}
override async execute(
batch: Batch<CartographerState>,
_context: NodeContextType,
): Promise<RoutedBatchType<'has-geo' | 'needs-geo', CartographerState>> {
const acc = new Map<'has-geo' | 'needs-geo', ItemType<CartographerState>[]>();
for (const item of batch) {
const result = this.routeItem(item.state);
for (const error of result.errors) {
item.state.collectError(error);
}
const bucket = acc.get(result.output);
if (bucket === undefined) {
acc.set(result.output, [item]);
} else {
bucket.push(item);
}
}
const routed = new Map<'has-geo' | 'needs-geo', Batch<CartographerState>>();
for (const [output, items] of acc) {
routed.set(output, Batch.from(items));
}
return routed;
}
private routeItem(state: CartographerState): NodeOutputType<'has-geo' | 'needs-geo'> {
const geo = state.canonical.geo;
// A source's pre-resolved geo only lets us skip the lookup when it actually
// resolved a location — an 'UNK'/'Unmapped' placeholder (e.g. a ping whose
// coords were out of range at the source) is NOT resolved, so it must run
// the lookup path where validate-coords can reject the bad coords.
const hasResolvedGeo =
geo !== undefined &&
geo.country.length > 0 &&
geo.country !== 'UNK' &&
geo.region.length > 0 &&
geo.region !== 'Unmapped';
if (hasResolvedGeo) {
state.routing = { ...state.routing, 'geoLookupSkipped': true, 'geoLookupRun': false };
return NodeOutput.create('has-geo');
}
state.routing = { ...state.routing, 'geoLookupRun': true, 'geoLookupSkipped': false };
return NodeOutput.create('needs-geo');
}
}ts
export class RouteRedactionNode extends MonadicNode<CartographerState, 'needs-redaction' | 'skip-redaction'> {
readonly '@id' = 'urn:noocodec:node:route-redaction';
readonly 'name' = 'route-redaction';
readonly 'outputs' = ['needs-redaction', 'skip-redaction'] as const;
override get outputSchema(): Record<'needs-redaction' | 'skip-redaction', SchemaObjectType> {
return {
'needs-redaction': { 'type': 'object' },
'skip-redaction': { 'type': 'object' },
};
}
override async execute(
batch: Batch<CartographerState>,
_context: NodeContextType,
): Promise<RoutedBatchType<'needs-redaction' | 'skip-redaction', CartographerState>> {
const acc = new Map<'needs-redaction' | 'skip-redaction', ItemType<CartographerState>[]>();
for (const item of batch) {
const result = this.routeItem(item.state);
for (const error of result.errors) {
item.state.collectError(error);
}
const bucket = acc.get(result.output);
if (bucket === undefined) {
acc.set(result.output, [item]);
} else {
bucket.push(item);
}
}
const routed = new Map<'needs-redaction' | 'skip-redaction', Batch<CartographerState>>();
for (const [output, items] of acc) {
routed.set(output, Batch.from(items));
}
return routed;
}
private routeItem(state: CartographerState): NodeOutputType<'needs-redaction' | 'skip-redaction'> {
const ev = state.currentEvent;
const hasPii =
state.canonical.pii === true ||
ev.recipientName.length > 0 ||
ev.recipientEmail.length > 0;
const alreadyHandled = state.canonical.consentHandled === true;
const consentStatus = Consent.statusFor(ev.shipmentId, ev.marketingConsent);
const juris = state.geoContext.jurisdiction;
const lightRegime = juris === 'baseline' || juris === 'international-waters';
// Light regime + valid consent imposes no redaction obligation.
const notRequired = lightRegime && consentStatus === 'valid';
const skip = !hasPii || alreadyHandled || notRequired;
if (skip) {
state.routing = { ...state.routing, 'redactionSkipped': true, 'redactionRun': false };
// Set a minimal no-op GdprResult: redaction NOT applied, precise coords
// retained. Marketing analytics eligibility still tracks valid consent.
state.gdprResult = {
...state.gdprResult,
'consentStatus': consentStatus,
'lawfulBasis': state.raw.lawfulBasis,
'jurisdiction': state.geoContext.jurisdiction,
'redactionApplied': false,
'coordsCoarsened': false,
'marketingAnalyticsEligible': consentStatus === 'valid',
};
return NodeOutput.create('skip-redaction');
}
state.routing = { ...state.routing, 'redactionRun': true, 'redactionSkipped': false };
return NodeOutput.create('needs-redaction');
}
}
export const routeRedaction = new RouteRedactionNode();The DAGs
Top-level: cartographer
ts
/**
* cartographerDAG: multi-entry source intake gather over raw source payloads.
*
* Five canonical entrypoint IRIs target intake-gather directly. The
* source-intake gather opens those per-type streams and merges them into
* state.sources.
* The processing scatter reads that merged stream at concurrency 16, runs
* stream-event per item, and folds completed
* clone state through insights-fold. Memory is O(1) regardless of event count.
*
* Topology:
* 5 data-type entrypoints → gather('intake-gather', source-intake)
* → scatter('process-stream', 'sources', { dag: 'stream-event' }, concurrency: 16)
* → gather('fold-insights', strategy: insights-fold)
* → summarize → done
*/
export const cartographerDAG: DAGType = new DAGBuilder(CARTOGRAPHER_DAG_IRI, '1.0')
.gather(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
CARTOGRAPHER_IRIS.intakeSources(CARTOGRAPHER_DAG_IRI),
{ 'strategy': 'source-intake' },
{
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
},
)
.scatter(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'sources',
{ 'dag': STREAM_EVENT_DAG_IRI },
{
'all-success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'partial': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'all-error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
},
{
'itemKey': 'source-payload',
'execution': { 'mode': 'reservoir', 'concurrency': 16, 'reservoir': { 'keyField': 'eventType', 'capacity': 1000 } },
},
)
.gather(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'), {
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream')]: {},
}, { 'strategy': 'insights-fold' }, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
})
// Pass-through in the streaming path (insights-fold already populated
// state.insights, state.journeys, and state.sampleRecords). Falls back
// to the records-based fold for non-streaming callers.
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'), summarizeInsights, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
})
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'), { outcome: 'completed' })
.entrypoints({
'position-ping': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'facility-scan': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'sensor-reading': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'customs-event': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'delivery-confirmation': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
})
.build();Worker-role top-level: cartographer
ts
/** Default reservoir capacity for the process-stream scatter in the workers DAG. */
export const DEFAULT_RESERVOIR_CAPACITY = 1000;
/**
* CartographerWorkersDag: static factory for the cartographer-workers DAG and
* its associated dispatcher bundle. Consumers call CartographerWorkersDag.build()
* (DAG only) or CartographerWorkersDag.bundle() (full dispatcher bundle) with an
* optional reservoir capacity override.
*
* DAG topology — identical to cartographerDAG with containerized boundaries:
* - container: 'cpu' so each stream-event body runs inside a
* WorkerThreadContainer/WebWorkerContainer rather than in-process.
* - container: 'io' so the final summary runs through the same embedded-DAG
* interface used by plugins and nested flows.
* - reservoir.capacity is parameterised; callers pass their UI-controlled
* batch size rather than relying on the compile-time default.
*
* 5 data-type entrypoints → gather('intake-gather', source-intake)
* → scatter('process-stream', 'sources', { dag: 'stream-event' },
* concurrency: 16, container: 'cpu', reservoir: { capacity })
* → gather('fold-insights', strategy: insights-fold)
* → embed('summarize-insights', 'insights-summary', container: 'io')
* → done
*/
export class CartographerWorkersDag {
private constructor() { /* static-only */ }
/**
* Build the cartographer-workers DAG with the given reservoir capacity.
* CLI, smoke tests, and dag-validate consumers use cartographerWorkersDAG
* (the pre-built constant); the browser demo calls this with a UI-controlled value.
*/
static build(capacity: number = DEFAULT_RESERVOIR_CAPACITY): DAGType {
return new DAGBuilder(CARTOGRAPHER_DAG_IRI, '1.0')
.gather(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
CARTOGRAPHER_IRIS.intakeSources(CARTOGRAPHER_DAG_IRI),
{ 'strategy': 'source-intake' },
{
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'),
},
)
.scatter(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'sources',
{ 'dag': STREAM_EVENT_DAG_IRI },
{
'all-success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'partial': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'all-error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
},
{
'itemKey': 'source-payload',
'container': 'cpu',
'execution': { 'mode': 'reservoir', 'concurrency': 16, 'reservoir': { 'keyField': 'eventType', 'capacity': capacity } },
},
)
.gather(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'), {
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream')]: {},
}, { 'strategy': 'insights-fold' }, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
})
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'), INSIGHTS_SUMMARY_DAG_IRI, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'),
}, {
'container': 'io',
})
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'), { outcome: 'completed' })
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'), { outcome: 'failed' })
.entrypoints({
'position-ping': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'facility-scan': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'sensor-reading': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'customs-event': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'delivery-confirmation': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
})
.build();
}
/**
* Build the workers bundle with a configurable reservoir capacity. The returned
* bundle is identical to cartographerWorkersBundle except that its cartographer
* DAG is built with CartographerWorkersDag.build(capacity) so the process-stream
* scatter uses the caller-supplied batch size.
*
* Used by the browser demo to wire UI-controlled knobs into each run() without
* mutating the shared default-capacity constants.
*/
static bundle(
capacity: number = DEFAULT_RESERVOIR_CAPACITY,
): DispatcherBundleType<CartographerState> {
return {
'nodes': [
...cartographerWorkerRuntimeBundle.nodes,
],
'dags': [
...cartographerWorkerRuntimeBundle.dags,
CartographerWorkersDag.build(capacity),
],
};
}
}
/**
* cartographerWorkersDAG: pre-built workers DAG at DEFAULT_RESERVOIR_CAPACITY.
* CLI, smoke tests, and dag-validate consumers use this constant; the browser
* demo uses CartographerWorkersDag.build(capacity) with a UI-controlled value.
*/
export const cartographerWorkersDAG: DAGType = CartographerWorkersDag.build();Summary body: insights-summary
ts
/**
* insights-summary: embedded summary body for the browser workers topology.
*
* The top-level workers DAG delegates this single-cardinality stage to the
* `io` container role after the `cpu` scatter finishes. The body is the same
* summarizeInsights node used by the in-process cartographer DAG, packaged as a
* registered DAG so the worker registry and JSON-LD assembly use the same
* canonical embed/plugin surface.
*/
export const insightsSummaryDAG: DAGType = new DAGBuilder(INSIGHTS_SUMMARY_DAG_IRI, '1.0')
.node(CARTOGRAPHER_IRIS.placementIri(INSIGHTS_SUMMARY_DAG_IRI, 'summarize'), summarizeInsights, {
'success': CARTOGRAPHER_IRIS.placementIri(INSIGHTS_SUMMARY_DAG_IRI, 'done'),
})
.terminal(CARTOGRAPHER_IRIS.placementIri(INSIGHTS_SUMMARY_DAG_IRI, 'done'), { outcome: 'completed' })
.build();Branching enrichment: event-pipeline-typed
ts
/**
* event-pipeline-typed: the live enrichment scatter body for process-events.
*
* Reads the scattered CanonicalEventVariant from metadata key 'canonical-event'
* and routes it to one of five per-type embedded DAGs via route-event-type-variant.
* Each per-type DAG starts with parse-variant (which also reads from metadata),
* embeds geo-pipeline for geo resolution, runs type-specific enrichment nodes,
* and converges on aggregate-event → done.
*
* route-event-type-variant
* ├─position-ping──────────► pipeline-position-ping (embedded)
* ├─sensor-reading─────────► pipeline-sensor-reading (embedded)
* ├─customs-event──────────► pipeline-customs-event (embedded)
* ├─facility-scan──────────► pipeline-facility-scan (embedded)
* └─delivery-confirmation──► pipeline-delivery-confirmation (embedded)
* Each per-type DAG:
* parse-variant → geo-pipeline → canonicalize-core → [type-specific] → aggregate-event → done
*
* Metadata propagation: the scatter sets 'canonical-event' on each clone's
* metadata. NodeStateBase.clone() copies _metadata, so metadata propagates
* to embedded child clones. Both route-event-type-variant and parse-variant
* read 'canonical-event' from metadata.
*/
export const eventPipelineTypedDAG: DAGType = new DAGBuilder(EVENT_PIPELINE_TYPED_DAG_IRI, '1.0')
// 1. route-event-type-variant: read eventType from 'canonical-event' metadata
// and dispatch to the corresponding per-type sub-DAG.
.node(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'route-event-type-variant'), routeEventType, {
'position-ping': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-position-ping'),
'sensor-reading': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-sensor-reading'),
'customs-event': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-customs-event'),
'facility-scan': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-facility-scan'),
'delivery-confirmation': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-delivery-confirmation'),
})
// 2a. pipeline-position-ping: geo + leg measurement.
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-position-ping'), CARTOGRAPHER_IRIS.dag.pipelinePositionPing, {
'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
'error': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
}, {
'outputs': {
'canonicalVariant': 'canonicalVariant',
'raw': 'raw',
'normalized': 'normalized',
'currentEvent': 'currentEvent',
'geoContext': 'geoContext',
'resolvedGeo': 'resolvedGeo',
'legKm': 'legKm',
'routing': 'routing',
'enriched': 'enriched',
'capturedErrors': 'capturedErrors',
},
})
// 2b. pipeline-sensor-reading: geo + cold-chain + leg measurement.
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-sensor-reading'), CARTOGRAPHER_IRIS.dag.pipelineSensorReading, {
'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
'error': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
}, {
'outputs': {
'canonicalVariant': 'canonicalVariant',
'raw': 'raw',
'normalized': 'normalized',
'currentEvent': 'currentEvent',
'geoContext': 'geoContext',
'resolvedGeo': 'resolvedGeo',
'coldChainBreach': 'coldChainBreach',
'legKm': 'legKm',
'routing': 'routing',
'enriched': 'enriched',
'capturedErrors': 'capturedErrors',
},
})
// 2c. pipeline-customs-event: geo + customs-dwell + leg measurement.
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-customs-event'), CARTOGRAPHER_IRIS.dag.pipelineCustomsEvent, {
'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
'error': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
}, {
'outputs': {
'canonicalVariant': 'canonicalVariant',
'raw': 'raw',
'normalized': 'normalized',
'currentEvent': 'currentEvent',
'geoContext': 'geoContext',
'resolvedGeo': 'resolvedGeo',
'customsDwellHours': 'customsDwellHours',
'legKm': 'legKm',
'routing': 'routing',
'enriched': 'enriched',
'capturedErrors': 'capturedErrors',
},
})
// 2d. pipeline-facility-scan: geo + facility canonicalization + order enrichment
// + GDPR-gated redaction.
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-facility-scan'), CARTOGRAPHER_IRIS.dag.pipelineFacilityScan, {
'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
'error': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
}, {
'outputs': {
'canonicalVariant': 'canonicalVariant',
'raw': 'raw',
'normalized': 'normalized',
'currentEvent': 'currentEvent',
'geoContext': 'geoContext',
'resolvedGeo': 'resolvedGeo',
'pricedOrder': 'pricedOrder',
'shippingQuote': 'shippingQuote',
'deliveryEstimate': 'deliveryEstimate',
'legKm': 'legKm',
'gdprResult': 'gdprResult',
'routing': 'routing',
'enriched': 'enriched',
'capturedErrors': 'capturedErrors',
},
})
// 2e. pipeline-delivery-confirmation: geo + recipient canonicalization +
// delivery confirmation + GDPR-gated redaction.
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'pipeline-delivery-confirmation'), CARTOGRAPHER_IRIS.dag.pipelineDeliveryConfirmation, {
'success': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'),
'error': CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'),
}, {
'outputs': {
'canonicalVariant': 'canonicalVariant',
'raw': 'raw',
'normalized': 'normalized',
'currentEvent': 'currentEvent',
'geoContext': 'geoContext',
'resolvedGeo': 'resolvedGeo',
'legKm': 'legKm',
'gdprResult': 'gdprResult',
'routing': 'routing',
'enriched': 'enriched',
'capturedErrors': 'capturedErrors',
},
})
.terminal(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'done'), { outcome: 'completed' })
.terminal(CARTOGRAPHER_IRIS.placementIri(EVENT_PIPELINE_TYPED_DAG_IRI, 'rejected'), { outcome: 'failed' })
.build();Ingestion sub-DAG: ingest-source
The shared transform node chain. Only the subset each format needs runs; the rest is skipped by the select-source routing node.
ts
/**
* IngestSourceDAG: orthogonal-compression, per-format-normalize ingestion DAG.
*
* Compression is a pre-step independent of format; each format has its own
* parse node and normalization sub-DAG; all converge on the shared coerce/validate tail.
*
* select-source ─compressed──► decompress ──► route-format
* ─plain─────────────────────► route-format
* ─invalid──────────────────────────────────► rejected
*
* route-format ─csv────► parse-csv ► [normalize-csv] ─┐
* ─json───► parse-json ► [normalize-json] ─┤
* ─ndjson─► parse-ndjson ► [normalize-ndjson] ─┤─► coerce-types ─► validate-event ─► ingested
* ─yaml───► parse-yaml ► [normalize-yaml] ─┘
* ─invalid──────────────────────────────────────► rejected
*
* Terminals: ingested (completed), rejected (failed — unselectable / bad payload).
*/
// #region ingest-source-dag
import { selectSource } from '../nodes/ingest/selectSource.ts';
import { decompress } from '../nodes/ingest/decompress.ts';
import { routeFormat } from '../nodes/ingest/routeFormat.ts';
import { parseCsv } from '../nodes/ingest/parseCsv.ts';
import { parseJson } from '../nodes/ingest/parseJson.ts';
import { parseNdjson } from '../nodes/ingest/parseNdjson.ts';
import { parseYaml } from '../nodes/ingest/parseYaml.ts';
import { normalizeCsv } from '../nodes/ingest/normalizeCsv.ts';
import { normalizeJson } from '../nodes/ingest/normalizeJson.ts';
import { normalizeNdjson } from '../nodes/ingest/normalizeNdjson.ts';
import { normalizeYaml } from '../nodes/ingest/normalizeYaml.ts';
import { coerceTypes } from '../nodes/ingest/coercion.ts';
import { validateEvent } from '../nodes/ingest/validateEvent.ts';
import { normalizeCsvDAG } from './NormalizeCsvDAG.ts';
import { normalizeJsonDAG } from './NormalizeJsonDAG.ts';
import { normalizeNdjsonDAG } from './NormalizeNdjsonDAG.ts';
import { normalizeYamlDAG } from './NormalizeYamlDAG.ts';
import type { CartographerState } from '../CartographerState.ts';
import type { DAGType, DispatcherBundleType } from '@studnicky/dagonizer';
import { DAGBuilder, DAGIdentity } from '@studnicky/dagonizer';
const ingestSourceDagIri = 'urn:noocodec:dag:ingest-source' as const;
const normalizeCsvDagIri = 'urn:noocodec:dag:normalize-csv' as const;
const normalizeJsonDagIri = 'urn:noocodec:dag:normalize-json' as const;
const normalizeNdjsonDagIri = 'urn:noocodec:dag:normalize-ndjson' as const;
const normalizeYamlDagIri = 'urn:noocodec:dag:normalize-yaml' as const;
const placement = (placementIdentifier: string): string =>
DAGIdentity.placementId(ingestSourceDagIri, placementIdentifier);
export const ingestSourceDAG: DAGType = new DAGBuilder(ingestSourceDagIri, '1.0')
// 1. select-source: read source from scatter metadata; route by compression.
.node(placement('select-source'), selectSource, {
'compressed': placement('decompress'),
'plain': placement('route-format'),
'invalid': placement('rejected'),
})
// 2. decompress: base64-decode + gunzip → plain text in state.decodedText.
// Format-agnostic: any gzipped format passes through here to route-format.
.node(placement('decompress'), decompress, {
'route-format': placement('route-format'),
'invalid': placement('rejected'),
})
// 3. route-format: dispatch to the format-specific parser.
.node(placement('route-format'), routeFormat, {
'csv': placement('parse-csv'),
'json': placement('parse-json'),
'ndjson': placement('parse-ndjson'),
'yaml': placement('parse-yaml'),
'invalid': placement('rejected'),
})
// 4a. parse-csv: CSV text → state.parsedRecords.
.node(placement('parse-csv'), parseCsv, {
'normalized': placement('normalize-csv'),
'invalid': placement('rejected'),
})
// 4b. parse-json: JSON array → state.parsedRecords.
.node(placement('parse-json'), parseJson, {
'normalized': placement('normalize-json'),
'invalid': placement('rejected'),
})
// 4c. parse-ndjson: NDJSON text → state.parsedRecords.
.node(placement('parse-ndjson'), parseNdjson, {
'normalized': placement('normalize-ndjson'),
'invalid': placement('rejected'),
})
// 4d. parse-yaml: YAML sequence → state.parsedRecords.
.node(placement('parse-yaml'), parseYaml, {
'normalized': placement('normalize-yaml'),
'invalid': placement('rejected'),
})
// 5a. normalize-csv: embedded sub-DAG — apply FieldMap by header name.
// The FieldMap is name-keyed, so shuffled CSV column order aligns correctly.
.embed<CartographerState, CartographerState>(placement('normalize-csv'), normalizeCsvDagIri, {
'success': placement('coerce-types'),
'error': placement('rejected'),
}, {
// Embedded DAGs run in an isolated state clone: thread the parsed records
// and the current source (for its mappingKey) IN, and the aligned records
// back OUT, or the normalize node sees empty input and nothing merges back.
'inputs': { 'parsedRecords': 'parsedRecords', 'currentSource': 'currentSource' },
'outputs': { 'mappedRecords': 'mappedRecords' },
})
// 5b. normalize-json: embedded sub-DAG — apply FieldMap by key name.
.embed<CartographerState, CartographerState>(placement('normalize-json'), normalizeJsonDagIri, {
'success': placement('coerce-types'),
'error': placement('rejected'),
}, {
// Embedded DAGs run in an isolated state clone: thread the parsed records
// and the current source (for its mappingKey) IN, and the aligned records
// back OUT, or the normalize node sees empty input and nothing merges back.
'inputs': { 'parsedRecords': 'parsedRecords', 'currentSource': 'currentSource' },
'outputs': { 'mappedRecords': 'mappedRecords' },
})
// 5c. normalize-ndjson: embedded sub-DAG — apply FieldMap by key name.
.embed<CartographerState, CartographerState>(placement('normalize-ndjson'), normalizeNdjsonDagIri, {
'success': placement('coerce-types'),
'error': placement('rejected'),
}, {
// Embedded DAGs run in an isolated state clone: thread the parsed records
// and the current source (for its mappingKey) IN, and the aligned records
// back OUT, or the normalize node sees empty input and nothing merges back.
'inputs': { 'parsedRecords': 'parsedRecords', 'currentSource': 'currentSource' },
'outputs': { 'mappedRecords': 'mappedRecords' },
})
// 5d. normalize-yaml: embedded sub-DAG — apply FieldMap by key name.
.embed<CartographerState, CartographerState>(placement('normalize-yaml'), normalizeYamlDagIri, {
'success': placement('coerce-types'),
'error': placement('rejected'),
}, {
// Embedded DAGs run in an isolated state clone: thread the parsed records
// and the current source (for its mappingKey) IN, and the aligned records
// back OUT, or the normalize node sees empty input and nothing merges back.
'inputs': { 'parsedRecords': 'parsedRecords', 'currentSource': 'currentSource' },
'outputs': { 'mappedRecords': 'mappedRecords' },
})
// 6. coerce-types: string cells → number / bool / epoch. Shared tail.
.node(placement('coerce-types'), coerceTypes, {
'validate-event': placement('validate-event'),
})
// 7. validate-event: build CanonicalEvents → state.ingestedEvents. Shared tail.
.node(placement('validate-event'), validateEvent, {
'validated': placement('ingested'),
})
// Terminals
.terminal(placement('ingested'), { outcome: 'completed' })
.terminal(placement('rejected'), { outcome: 'failed' })
.build();
export const ingestSourceBundle: DispatcherBundleType<CartographerState> = {
// Normalize DAGs registered FIRST so the embedded-DAG placements above resolve.
'nodes': [
selectSource, decompress, routeFormat,
parseCsv, parseJson, parseNdjson, parseYaml,
normalizeCsv, normalizeJson, normalizeNdjson, normalizeYaml,
coerceTypes, validateEvent,
],
'dags': [
normalizeCsvDAG, normalizeJsonDAG, normalizeNdjsonDAG, normalizeYamlDAG,
ingestSourceDAG,
],
};
// #endregion ingest-source-dagSource-model geo-resolution sub-DAG: geo-source-resolve
geo-source-resolve has six labeled entrypoints: coords, address, ip, code, phone, and locale. Each entrypoint embeds a small resolver DAG. The resolver DAG prepares a GeoSignalDescriptor when its modality is present, runs the dedicated resolver node, and projects state.candidate into the parent gather record. The parent geo-weighted-fusion GatherNode waits for all six producer labels and folds the candidate records by weight into state.resolvedGeo, state.geoContext, and state.routing.{geoConfidence,geoModalities}. When no candidate resolves, the gather writes the same baseline values directly.
Coords resolution uses GeohashTzMap (a base64-embedded binary geohash→timezone table) as the fast offline path, with CoordTimezone (tz-lookup + @rapideditor/country-coder) as the browser-safe border/gap default path. Address resolution calls the injected AddressGeocoder transport (Nominatim live; deterministic no-answer in the smoke). Both IP and address transports are injected per-call so worker threads own independent instances.
ts
/**
* GeoSourceResolveDAG: validity-gated weighted multi-entry geo-resolution sub-DAG.
*
* The parent DAG has one entrypoint per geo modality. Each entrypoint embeds a
* modality-specific resolver DAG that prepares its descriptor, resolves the
* candidate when present, and projects `state.candidate` into a gather record.
* The first-class `geo-weighted-fusion` gather node is the graph-visible barrier
* that folds candidates by weight into `state.resolvedGeo`, `state.geoContext`,
* and `state.routing.{geoConfidence,geoModalities}`. When no signal resolves,
* the gather writes the baseline values directly.
*
* Topology:
* entrypoints:
* coords ─► resolve-coords-source ─success/error─┐
* address ─► resolve-address-source ─success/error─┤
* ip ─► resolve-ip-source ─success/error─┤
* code ─► resolve-code-source ─success/error─┤─► geo-weighted-fusion
* phone ─► resolve-phone-source ─success/error─┤
* locale ─► resolve-locale-source ─success/error─┘
* geo-weighted-fusion (gather) ─success/error/empty──► resolved
* resolved: terminal (completed)
*
* DI: `ipGeolocator` and `addressGeocoder` are injected per-call so each
* dispatcher (main thread, worker thread) owns its own transport instances.
*/
// #region geo-source-resolve-dag
import { resolveCoords } from '../nodes/geo/resolveCoords.ts';
import { resolveLocale } from '../nodes/geo/resolveLocale.ts';
import { resolveCode } from '../nodes/geo/resolveCode.ts';
import { resolvePhone } from '../nodes/geo/resolvePhone.ts';
import { ResolveIpNode } from '../nodes/geo/resolveIp.ts';
import { ResolveAddressNode } from '../nodes/geo/resolveAddress.ts';
import {
prepareGeoAddress,
prepareGeoCode,
prepareGeoCoords,
prepareGeoIp,
prepareGeoLocale,
prepareGeoPhone,
} from '../nodes/geo/prepareGeoSignal.ts';
import { CARTOGRAPHER_IRIS } from '../cartographerIds.ts';
import type { IpGeolocator } from '../contracts/IpGeolocator.ts';
import type { AddressGeocoder } from '../contracts/AddressGeocoder.ts';
import type { CartographerState } from '../CartographerState.ts';
// Side-effect import: registers 'geo-weighted-fusion' at module load so the
// scatter placement can resolve the strategy name at dispatcher construction.
import '../core/GeoWeightedFusionGather.ts';
import type { DispatcherBundleType } from '@studnicky/dagonizer';
import { DAGBuilder } from '@studnicky/dagonizer';
export class GeoSourceResolveDAG {
private constructor() { /* static-only */ }
static build(
ipGeolocator: IpGeolocator,
addressGeocoder: AddressGeocoder,
): DispatcherBundleType<CartographerState> {
const resolveIp = new ResolveIpNode(ipGeolocator);
const resolveAddress = new ResolveAddressNode(addressGeocoder);
const resolveCoordsDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveCoords, '1.0')
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'prepare-geo-coords'), prepareGeoCoords, {
'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'resolve-coords'),
'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'done'),
})
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'resolve-coords'), resolveCoords, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'done') })
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'done'), { 'outcome': 'completed' })
.build();
const resolveAddressDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveAddress, '1.0')
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'prepare-geo-address'), prepareGeoAddress, {
'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'resolve-address'),
'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'done'),
})
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'resolve-address'), resolveAddress, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'done') })
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'done'), { 'outcome': 'completed' })
.build();
const resolveIpDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveIp, '1.0')
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'prepare-geo-ip'), prepareGeoIp, {
'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'resolve-ip'),
'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'done'),
})
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'resolve-ip'), resolveIp, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'done') })
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'done'), { 'outcome': 'completed' })
.build();
const resolveCodeDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveCode, '1.0')
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'prepare-geo-code'), prepareGeoCode, {
'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'resolve-code'),
'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'done'),
})
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'resolve-code'), resolveCode, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'done') })
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'done'), { 'outcome': 'completed' })
.build();
const resolvePhoneDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolvePhone, '1.0')
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'prepare-geo-phone'), prepareGeoPhone, {
'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'resolve-phone'),
'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'done'),
})
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'resolve-phone'), resolvePhone, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'done') })
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'done'), { 'outcome': 'completed' })
.build();
const resolveLocaleDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveLocale, '1.0')
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'prepare-geo-locale'), prepareGeoLocale, {
'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'resolve-locale'),
'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'done'),
})
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'resolve-locale'), resolveLocale, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'done') })
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'done'), { 'outcome': 'completed' })
.build();
const dag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoSourceResolve, '1.0')
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-coords-source'), CARTOGRAPHER_IRIS.dag.geoResolveCoords, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
}, {
'gatherResult': { 'resultField': 'candidate' },
})
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-address-source'), CARTOGRAPHER_IRIS.dag.geoResolveAddress, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
}, {
'gatherResult': { 'resultField': 'candidate' },
})
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-ip-source'), CARTOGRAPHER_IRIS.dag.geoResolveIp, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
}, {
'gatherResult': { 'resultField': 'candidate' },
})
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-code-source'), CARTOGRAPHER_IRIS.dag.geoResolveCode, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
}, {
'gatherResult': { 'resultField': 'candidate' },
})
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-phone-source'), CARTOGRAPHER_IRIS.dag.geoResolvePhone, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
}, {
'gatherResult': { 'resultField': 'candidate' },
})
.embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-locale-source'), CARTOGRAPHER_IRIS.dag.geoResolveLocale, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
}, {
'gatherResult': { 'resultField': 'candidate' },
})
.entrypoints({
'coords': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-coords-source'),
'address': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-address-source'),
'ip': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-ip-source'),
'code': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-code-source'),
'phone': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-phone-source'),
'locale': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-locale-source'),
})
// First-class gather barrier over all embedded resolver producers.
.gather(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
{
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-coords-source')]: {},
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-address-source')]: {},
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-ip-source')]: {},
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-code-source')]: {},
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-phone-source')]: {},
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-locale-source')]: {},
},
{ 'strategy': 'geo-weighted-fusion' },
{
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'),
},
)
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'), { 'outcome': 'completed' })
.build();
return {
'nodes': [
prepareGeoCoords,
prepareGeoAddress,
prepareGeoIp,
prepareGeoCode,
prepareGeoPhone,
prepareGeoLocale,
resolveCoords,
resolveAddress,
resolveIp,
resolveCode,
resolvePhone,
resolveLocale,
],
'dags': [
resolveCoordsDag,
resolveAddressDag,
resolveIpDag,
resolveCodeDag,
resolvePhoneDag,
resolveLocaleDag,
dag,
],
};
}
}
// #endregion geo-source-resolve-dagts
export class PrepareGeoSignalNode extends MonadicNode<CartographerState, PrepareGeoSignalOutput> {
readonly '@id': string;
readonly 'outputs' = ['present', 'missing'] as const;
readonly #kind: GeoSignalKind;
readonly #name: string;
constructor(id: string, name: string, kind: GeoSignalKind) {
super();
this['@id'] = id;
this.#name = name;
this.#kind = kind;
}
get name(): string {
return this.#name;
}
override get outputSchema(): Record<PrepareGeoSignalOutput, SchemaObjectType> {
return {
'present': { 'type': 'object' },
'missing': { 'type': 'object' },
};
}
override async execute(
batch: Batch<CartographerState>,
_context: NodeContextType,
): Promise<RoutedBatchType<PrepareGeoSignalOutput, CartographerState>> {
const present: Array<{ readonly id: string; readonly state: CartographerState }> = [];
const missing: Array<{ readonly id: string; readonly state: CartographerState }> = [];
for (const item of batch) {
const descriptor = this.#descriptor(item.state);
if (descriptor === null) {
item.state.candidate = GeoResolutionBuilder.from({ 'source': this.#kind, 'weight': 0 });
missing.push(item);
} else {
item.state.setMetadata('geo-signal', descriptor);
present.push(item);
}
}
const routed = new Map<PrepareGeoSignalOutput, Batch<CartographerState>>();
if (present.length > 0) routed.set('present', Batch.from(present));
if (missing.length > 0) routed.set('missing', Batch.from(missing));
return routed;
}
#descriptor(state: CartographerState): GeoSignalDescriptor | null {
const body = state.canonical.body;
switch (this.#kind) {
case 'coords':
return Number.isFinite(body.latitude)
&& Number.isFinite(body.longitude)
&& (body.latitude !== 0 || body.longitude !== 0)
&& Math.abs(body.latitude) <= 90
&& Math.abs(body.longitude) <= 180
? GeoSignalDescriptorBuilder.from({
'kind': 'coords',
'weight': SignalWeight.for('coords'),
'lat': body.latitude,
'lng': body.longitude,
})
: null;
case 'address':
return body.address.length > 0
? GeoSignalDescriptorBuilder.from({
'kind': 'address',
'weight': SignalWeight.for('address'),
'address': body.address,
})
: null;
case 'ip':
return body.ipAddress.length > 0
? GeoSignalDescriptorBuilder.from({
'kind': 'ip',
'weight': SignalWeight.for('ip'),
'ipAddress': body.ipAddress,
})
: null;
case 'code':
return body.countryCode.length > 0
? GeoSignalDescriptorBuilder.from({
'kind': 'code',
'weight': SignalWeight.for('code'),
'countryCode': body.countryCode,
})
: null;
case 'phone':
return body.phone.length > 0 && CallingCode.countryFor(body.phone).length > 0
? GeoSignalDescriptorBuilder.from({
'kind': 'phone',
'weight': SignalWeight.for('phone'),
'phone': body.phone,
})
: null;
case 'locale':
return body.localeTag.length > 0
? GeoSignalDescriptorBuilder.from({
'kind': 'locale',
'weight': SignalWeight.for('locale'),
'localeTag': body.localeTag,
})
: null;
}
}
}
export const prepareGeoCoords = new PrepareGeoSignalNode('urn:noocodec:node:prepare-geo-coords', 'prepare-geo-coords', 'coords');
export const prepareGeoAddress = new PrepareGeoSignalNode('urn:noocodec:node:prepare-geo-address', 'prepare-geo-address', 'address');
export const prepareGeoIp = new PrepareGeoSignalNode('urn:noocodec:node:prepare-geo-ip', 'prepare-geo-ip', 'ip');
export const prepareGeoCode = new PrepareGeoSignalNode('urn:noocodec:node:prepare-geo-code', 'prepare-geo-code', 'code');
export const prepareGeoPhone = new PrepareGeoSignalNode('urn:noocodec:node:prepare-geo-phone', 'prepare-geo-phone', 'phone');
export const prepareGeoLocale = new PrepareGeoSignalNode('urn:noocodec:node:prepare-geo-locale', 'prepare-geo-locale', 'locale');GDPR compliance sub-DAG: gdpr-compliance
ts
/**
* GdprComplianceDAG: reusable GDPR compliance sub-pipeline.
*
* Internal flow:
*
* consent-gate
* └─ classify ──► classify-pii
* classify-pii
* └─ redact ──► redact-pii
* redact-pii
* ├─ ok ──► compliant (TerminalNode completed → parent routes 'success')
* └─ violation ──► violation (TerminalNode failed → parent routes 'error')
*
* Embedded via:
* .embed(gdprPlacementIri, gdprComplianceDagIri,
* { 'success': aggregateEventPlacementIri, 'error': gdprViolationPlacementIri },
* { 'outputs': { 'currentEvent':'currentEvent', 'gdprResult':'gdprResult' } })
*
* The embedded DAG runs in a CLONED child state. Its nodes redact PII on
* state.currentEvent and write state.gdprResult; the `outputs` mapping copies
* those two fields back into the parent shipment-pipeline clone when the
* sub-DAG completes (child-key → parent-path orientation).
*/
// #region gdpr-compliance-dag
import { consentGate, classifyPii, redactPii } from '../nodes/gdprNodes.ts';
import { CARTOGRAPHER_IRIS } from '../cartographerIds.ts';
import type { CartographerState } from '../CartographerState.ts';
import type { DAGType, DispatcherBundleType } from '@studnicky/dagonizer';
import { DAGBuilder } from '@studnicky/dagonizer';
const GDPR_COMPLIANCE_DAG_IRI = CARTOGRAPHER_IRIS.dag.gdprCompliance;
export const gdprComplianceDAG: DAGType = new DAGBuilder(GDPR_COMPLIANCE_DAG_IRI, '1.0')
// ── 1. consent-gate ──────────────────────────────────────────────────────
// Resolves the consent status from marketingConsent + simulated expiry.
// Always routes 'classify' (both consented and non-consented proceed;
// the consent status drives redaction rules downstream).
.node(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'consent-gate'), consentGate, {
'classify': CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'classify-pii'),
})
// ── 2. classify-pii ──────────────────────────────────────────────────────
// Records which fields are personal/sensitive; no routing decision yet.
.node(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'classify-pii'), classifyPii, {
'redact': CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'redact-pii'),
})
// ── 3. redact-pii ────────────────────────────────────────────────────────
// Applies GdprRedactor.redact. Routes to 'compliant' (ok) or 'violation'.
.node(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'redact-pii'), redactPii, {
'ok': CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'compliant'),
'violation': CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'violation'),
})
// ── Terminals ─────────────────────────────────────────────────────────────
.terminal(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'compliant'), { outcome: 'completed' })
.terminal(CARTOGRAPHER_IRIS.placementIri(GDPR_COMPLIANCE_DAG_IRI, 'violation'), { outcome: 'failed' })
.build();
// #endregion gdpr-compliance-dag
export const gdprComplianceBundle: DispatcherBundleType<CartographerState> = {
'nodes': [consentGate, classifyPii, redactPii],
'dags': [gdprComplianceDAG],
};State and services
CartographerState
The mutable clipboard threaded through every node. Top-level fields hold the source feeds, ingested events, gathered records, and insights aggregates. Clone fields hold the per-event enrichment pipeline's intermediate values.
ts
export class CartographerState extends NodeStateBase {
/** Number of synthetic journeys to generate; part of the checkpoint/resume serialized state. */
eventCount: number = 200;
/**
* Per-event-type feed configuration driving buildTypedFeed / streamTyped. Each
* entry generates entry.count typed scans of its eventType, encoded across the
* formats in its formatMix.
*/
eventConfig: EventTypeConfig = [
{ 'eventType': 'position-ping', 'count': 6, 'formatMix': [{ 'format': 'json', 'compression': 'none', 'weight': 2 }, { 'format': 'yaml', 'compression': 'gzip', 'weight': 1 }] },
{ 'eventType': 'facility-scan', 'count': 5, 'formatMix': [{ 'format': 'csv', 'compression': 'none', 'weight': 2 }, { 'format': 'json', 'compression': 'gzip', 'weight': 1 }] },
{ 'eventType': 'sensor-reading', 'count': 4, 'formatMix': [{ 'format': 'ndjson', 'compression': 'gzip', 'weight': 2 }, { 'format': 'ndjson', 'compression': 'none', 'weight': 1 }] },
{ 'eventType': 'customs-event', 'count': 3, 'formatMix': [{ 'format': 'json', 'compression': 'none', 'weight': 2 }, { 'format': 'csv', 'compression': 'none', 'weight': 1 }] },
{ 'eventType': 'delivery-confirmation', 'count': 3, 'formatMix': [{ 'format': 'json', 'compression': 'none', 'weight': 2 }, { 'format': 'csv', 'compression': 'gzip', 'weight': 1 }] },
];
/**
* When true, host tooling reports the Cartographer run as streaming. The DAG
* always assembles source streams through its multi-entry intake gather; this
* flag remains a UI/CLI display knob.
*/
useStreamingSource: boolean = false;
/**
* Override for the total source-payload count. When > 0, the source-intake
* gather scales eventConfig before opening per-type streams.
*/
streamCount: number = 0;
/**
* StreamChannel buffer capacity for the streaming source channel.
* When 0 (default), the channel uses its built-in default capacity (256).
* Set to 1 to enable genuine per-item backpressure (useful for abort-and-resume
* scenarios where the producer must not pre-fill the buffer).
*/
streamChannelCapacity: number = 0;
/**
* The multi-format source feeds, assembled by intake-gather. Each is a
* `{ sourceId, format, mappingKey, eventType, payload }` — a different on-the-wire
* encoding (JSON / CSV / gzip NDJSON) of a typed scan from the event feed.
*
* This field normally holds the merged `AsyncIterable<SourcePayload>` built
* from five per-type source streams. The engine's scatter accepts that stream
* directly. Snapshot/restore serialises the array path only; resume rebuilds
* the async iterable from eventConfig and the scatter cursor.
*/
sources: SourcePayload[] | AsyncIterable<SourcePayload> = [];
/**
* Ingestion fan-in buckets: the `append` gather of the ingestion scatter
* appends each source clone's `ingestedEvents` array as one element here, so
* this is one bucket per source. The `merge-events` node flattens it into the
* unified `canonicalEvents` collection.
*/
ingestBuckets: CanonicalEventVariant[][] = [];
/**
* The unified canonical event collection. Every source's decoded events are
* flattened into this one array (from `ingestBuckets`); the enrichment scatter
* then reads it.
*/
canonicalEvents: CanonicalEventVariant[] = [];
// ── Per-source ingest slots (used inside a source's ingest sub-DAG clone) ──
/** The source feed currently being ingested (set from `sources` by select). */
currentSource: SourcePayload = {
'sourceId': '',
'format': 'json',
'compression': 'none',
'mappingKey': 'json-position',
'eventType': 'position-ping',
'payload': '',
};
/** Decompressed/raw text of the current source (after `decompress`). */
decodedText: string = '';
/** Records parsed from the decoded text (after parse-csv/json/ndjson). */
parsedRecords: Array<Record<string, unknown>> = [];
/** Records with source field names mapped to canonical fields (after map-fields). */
mappedRecords: Array<Record<string, unknown>> = [];
/** Canonical events validated from this source (after coerce-types + validate-event). */
ingestedEvents: CanonicalEventVariant[] = [];
/** The single canonical event (geo/consent mirror) under enrichment in a scatter clone; set/updated by parseVariant. */
canonical: CanonicalEventVariant = CanonicalEventVariantBuilder.from({});
/** The discriminated per-type variant under enrichment (typed path; set by parseVariant). The old fat path uses `canonical`. */
canonicalVariant: CanonicalEventVariant = CanonicalEventVariantBuilder.from({});
/** Enriched shipment records gathered from scatter clones. */
records: EnrichedShipment[] = [];
/**
* Bounded FIFO sample of enriched scans (cap 200) produced by the
* insights-fold gather strategy. The gather writes to this field
* incrementally as scatter clones complete; memory does not grow
* with event count. In the streaming path state.records stays empty;
* this field holds the representative sample the UI consumes.
*/
sampleRecords: EnrichedShipment[] = [];
/** Fixed-size regional insights aggregate produced by summarizeInsights. */
insights: Map<string, RegionInsights> = new Map();
/** Per-journey aggregate (grouped by shipmentId) produced by summarizeInsights. */
journeys: Map<string, JourneyInsights> = new Map();
/**
* In-progress per-journey accumulator map, maintained by the insights-fold gather strategy.
* Part of durable checkpoint state so accumulations survive process restart on resume.
*/
journeyAccumulators: Map<string, JourneyAccumulator> = new Map();
/**
* Parent-side bounded rollup of captured exceptions, folded by the
* insights-fold gather from each clone's `state.capturedErrors`. Errors flow
* scatter→gather as first-class data; the run prints this distribution for
* analysis. Reset per execution by the gather's `initial`.
*/
errorRollup: ErrorRollupType = ErrorRollup.empty();
/** Raw scan from scatter metadata (set by parseEvent). */
raw: RawShipmentEvent = {
'shipmentId': '',
'scanSeq': 0,
'rawTimestamp': '',
'rawDispatchAt': '',
'rawStatus': '',
'carrier': '',
'ipAddress': '',
'localeTag': '',
'countryCode': '',
'latitude': 0,
'longitude': 0,
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'weight': 0,
'weightUnit': 'kg',
'recipientName': '',
'recipientEmail': '',
'recipientPhone': '',
'recipientAddress': '',
'recipientCountry': '',
'marketingConsent': false,
'rawPromisedDeliveryAt': '',
'lineItems': [{ 'productId': '', 'quantity': 1 }],
'facilityId': '',
'lawfulBasis': 'contract',
'specialCategory': 'none',
'disruptionReason': '',
};
/** Normalised canonical form (set by normalize node). */
normalized: NormalizedShipment = {
'shipmentId': '',
'scanSeq': 0,
'epochMs': 0,
'dispatchEpochMs': 0,
'isoTimestamp': '',
'localIso': '',
'utcOffset': '',
'carrierId': '',
'carrierName': '',
'countryIso3': 'UNK',
'weightGrams': 0,
'status': 'SCAN',
'serviceTier': 'standard',
'sizeTier': 'small',
'lineItems': [{ 'productId': '', 'quantity': 1 }],
'facilityId': '',
'latitude': 0,
'longitude': 0,
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'recipientName': '',
'recipientEmail': '',
'recipientPhone': '',
'recipientAddress': '',
'recipientCountry': '',
'marketingConsent': false,
'promisedEpochMs': 0,
'disruptionHours': 0,
'disruptionReason': '',
};
/**
* ShipmentEvent-shaped current event used by geo and GDPR nodes.
* Populated from normalized by the classify node.
*/
currentEvent: ShipmentEvent = {
'shipmentId': '',
'timestamp': '',
'eventType': 'SCAN',
'latitude': 0,
'longitude': 0,
'carrier': '',
'facilityId': '',
'recipientName': '',
'recipientEmail': '',
'recipientPhone': '',
'recipientAddress': '',
'recipientCountry': '',
'marketingConsent': false,
'promisedDeliveryAt': '',
};
/** Geo-enrichment result for the current scan (incl. timezone + jurisdiction). */
geoContext: GeoContext = {
'gridZone': '',
'country': '',
'continent': 'Unmapped',
'countries': [],
'region': '',
'hub': '',
'status': 'unmapped',
'waterBodies': [],
'timezone': 'UTC',
'jurisdiction': 'baseline',
};
/** Basket pricing result (set by enrich-pricing node). */
pricedOrder: PricedOrder = {
'lines': [],
'subtotalMinor': 0,
'currency': 'USD',
'subtotalUsdMinor': 0,
'fxRate': 1.0,
};
/** Shipping cost + distance (set by enrich-shipping node). */
shippingQuote: ShippingQuote = {
'distanceKm': 0,
'costUsdMinor': 0,
'breakdown': {
'baseMinor': 0,
'perKmMinor': 0,
'perKgMinor': 0,
'tierMultiplier': 1.0,
},
};
/** ETA calculation (set by enrich-eta node). */
deliveryEstimate: DeliveryEstimate = {
'transitHours': 0,
'etaEpochMs': 0,
'etaIso': '',
'promisedEpochMs': 0,
'onTime': false,
'delayHours': 0,
};
/** Leg distance (legFrom → this scan) in km, set by enrich-leg node. */
legKm: number = 0;
/** Cold-chain breach flag (sensor lane only; set by cold-chain-check). */
coldChainBreach: boolean = false;
/** Customs clearance dwell hours (customs lane only; set by customs-dwell). */
customsDwellHours: number = 0;
/**
* This scan's conditional-routing decisions (the branching headline). Each
* routing node records what RAN vs was SKIPPED here; aggregate-event copies it
* onto the enriched record so the parent's summarize totals the savings (no
* shared mutable counters across scatter clones).
*/
routing: EnrichedShipment['routing'] = CartographerState.defaultRouting();
/**
* Ephemeral per-clone accumulator of captured exceptions (like ipCandidate:
* non-serialized, recomputed each dispatch). The geo / ingest nodes append a
* `GeoErrorRecordType` here whenever their transport reports a captured error;
* the gather folds these into the parent's `errorRollup`. The node still
* routes its normal output — the error rides alongside as data.
*
* Named `capturedErrors` (not `errors`) to stay distinct from the framework
* `NodeStateInterface.errors` channel — these are the example's own captured
* geo/ingest faults, a different concern.
*/
capturedErrors: readonly GeoErrorRecordType[] = [];
/** IP-modality candidate from ip-geolocate (unresolved when that node skipped). */
ipCandidate: GeoCandidate = CartographerState.unresolvedCandidate('ip');
/** The fused multi-modal location (set by fuse-source-geo). */
resolvedGeo: ResolvedGeo = {
'country': '',
'countryName': '',
'continent': 'Unmapped',
'region': '',
'locality': '',
'locale': '',
'lat': 0,
'lng': 0,
'status': 'land',
'jurisdiction': 'baseline',
'confidence': 0,
'modalities': [],
'provenance': [],
};
/** Source-model routing signal built by classify-geo-source (which path to take). */
geoSignal: GeoSignal = { ...DEFAULT_GEO_SIGNAL };
/** Resolved geo from the selected source-model path (set by the source-resolve nodes). */
geoResolution: GeoResolution = { ...DEFAULT_GEO_RESOLUTION };
/** Scored geo signals for the scatter-gather resolution path (Wave 1+). */
geoSignals: GeoSignalDescriptor[] = [];
/** Current best candidate resolution (Wave 1+). */
candidate: GeoResolution = { ...DEFAULT_GEO_RESOLUTION };
/** All scored resolution candidates (Wave 1+). */
geoCandidates: GeoResolution[] = [];
/** GDPR processing result for the current scan (location + consent driven). */
gdprResult: GdprResult = {
'personalDataFields': [],
'sensitiveDataFields': [],
'consentStatus': 'missing',
'lawfulBasis': 'contract',
'jurisdiction': 'baseline',
'strictness': 'light',
'complianceScore': 0,
'retention': { 'retainUntil': '', 'autoDelete': false },
'redactionApplied': false,
'marketingAnalyticsEligible': false,
'coordsCoarsened': false,
};
/** Compact enriched per-scan record written by aggregate-event; parent gather appends it. */
enriched: EnrichedShipment = {
'shipmentId': '',
'scanSeq': 0,
'epochMs': 0,
'localIso': '',
'utcOffset': '',
'timezone': 'UTC',
'jurisdiction': 'baseline',
'continent': 'Unmapped',
'region': '',
'country': '',
'hub': '',
'geoStatus': 'unmapped',
'lat': 0,
'lng': 0,
'coordsCoarsened': false,
'legKm': 0,
'status': 'SCAN',
'serviceTier': 'standard',
'sizeTier': 'small',
'onTime': false,
'exception': false,
'consentStatus': 'missing',
'disruptionReason': '',
'subtotalUsdMinor': 0,
'currency': 'USD',
'shippingUsdMinor': 0,
'distanceKm': 0,
'transitHours': 0,
'delayHours': 0,
'redactionApplied': false,
'redactedSample': { 'recipientName': '', 'recipientEmail': '', 'recipientPhone': '' },
'routing': CartographerState.defaultRouting(),
};
override clone(): this {
const copy = super.clone(); // new Constructor() + _metadata copy from base
copy.eventCount = this.eventCount;
copy.eventConfig = this.eventConfig.map((e) => ({ 'eventType': e.eventType, 'count': e.count, 'formatMix': e.formatMix.map((m) => ({ ...m })) }));
if (Array.isArray(this.sources)) {
copy.sources = this.sources.map((s) => ({ ...s }));
} else {
// AsyncIterable — shared by reference
copy.sources = this.sources;
}
copy.useStreamingSource = this.useStreamingSource;
copy.streamCount = this.streamCount;
copy.streamChannelCapacity = this.streamChannelCapacity;
// Parent-level accumulators: reset to defaults in child clones.
//
// ingestBuckets, canonicalEvents, records, sampleRecords, insights, and
// journeys are scatter-gather accumulators written by the parent DAG's
// gather strategy (InsightsFoldGather) or by post-scatter summary nodes.
// Scatter body clones (stream-event, ingestion) never read these fields —
// they only read the item placed on metadata by the engine. Copying them
// into clones would send up to 200 EnrichedShipment JSON objects per clone
// over the worker channel (60 KB × 16,000 in-flight clones = ~960 MB at
// concurrencyLimit=16 / capacity=1000), producing the O(peak-concurrency)
// heap spike. Resetting to defaults eliminates that overhead with no loss
// of correctness: the child never reads them, and the parent retains its
// own live copies.
copy.ingestBuckets = [];
copy.canonicalEvents = [];
copy.records = [];
copy.sampleRecords = [];
copy.insights = new Map();
copy.journeys = new Map();
copy.journeyAccumulators = new Map();
copy.errorRollup = ErrorRollup.empty();
copy.currentSource = { ...this.currentSource };
copy.decodedText = this.decodedText;
copy.parsedRecords = this.parsedRecords.map((r) => ({ ...r }));
copy.mappedRecords = this.mappedRecords.map((r) => ({ ...r }));
copy.ingestedEvents = this.ingestedEvents.map((e) => CartographerState.cloneVariant(e));
copy.canonical = CartographerState.cloneVariant(this.canonical);
copy.canonicalVariant = CartographerState.cloneVariant(this.canonicalVariant);
copy.raw = {
...this.raw,
'lineItems': this.raw.lineItems.map((li) => ({ ...li })),
};
copy.normalized = {
...this.normalized,
'lineItems': this.normalized.lineItems.map((li) => ({ ...li })),
};
copy.currentEvent = { ...this.currentEvent };
copy.geoContext = {
...this.geoContext,
'countries': [...this.geoContext.countries],
'waterBodies': [...this.geoContext.waterBodies],
};
copy.pricedOrder = {
...this.pricedOrder,
'lines': this.pricedOrder.lines.map((l) => ({ ...l })),
};
copy.shippingQuote = {
...this.shippingQuote,
'breakdown': { ...this.shippingQuote.breakdown },
};
copy.deliveryEstimate = { ...this.deliveryEstimate };
copy.legKm = this.legKm;
copy.coldChainBreach = this.coldChainBreach;
copy.customsDwellHours = this.customsDwellHours;
copy.capturedErrors = this.capturedErrors.map((e) => ({ ...e }));
copy.ipCandidate = { ...this.ipCandidate };
copy.resolvedGeo = { ...this.resolvedGeo, 'modalities': [...this.resolvedGeo.modalities], 'provenance': [...this.resolvedGeo.provenance] };
copy.geoSignal = { ...this.geoSignal };
copy.geoResolution = { ...this.geoResolution };
copy.geoSignals = this.geoSignals.map((s) => ({ ...s }));
copy.candidate = { ...this.candidate };
copy.geoCandidates = this.geoCandidates.map((c) => ({ ...c }));
copy.routing = { ...this.routing, 'geoModalities': [...this.routing.geoModalities] };
copy.gdprResult = {
...this.gdprResult,
'personalDataFields': [...this.gdprResult.personalDataFields],
'sensitiveDataFields': [...this.gdprResult.sensitiveDataFields],
'retention': { ...this.gdprResult.retention },
};
copy.enriched = {
...this.enriched,
'redactedSample': { ...this.enriched.redactedSample },
};
return copy;
}
protected override snapshotData(): JsonObjectType {
return {
'eventCount': this.eventCount,
'eventConfig': this.eventConfig.map((e) => ({ 'eventType': e.eventType, 'count': e.count, 'formatMix': e.formatMix.map((m) => ({ 'format': m.format, 'compression': m.compression, 'weight': m.weight })) })),
// AsyncIterable sources are not checkpointable. Resume rebuilds them via
// CartographerSourceIntake using eventConfig + streamCount + cursor.
// Snapshot as an empty array so restoreData leaves sources = [].
'sources': Array.isArray(this.sources)
? this.sources.map((s) => CartographerState.sourceToJson(s))
: [],
'useStreamingSource': this.useStreamingSource,
'streamCount': this.streamCount,
'streamChannelCapacity': this.streamChannelCapacity,
'ingestBuckets': this.ingestBuckets.map((bucket) => bucket.map((e) => CartographerState.variantToJson(e))),
'canonicalEvents': this.canonicalEvents.map((e) => CartographerState.variantToJson(e)),
'records': this.records.map((r) => CartographerState.enrichedToJson(r)),
'sampleRecords': this.sampleRecords.map((r) => CartographerState.enrichedToJson(r)),
'enriched': CartographerState.enrichedToJson(this.enriched),
'insights': [...this.insights.entries()].map(([key, r]) => ({
'key': key,
'region': r.region, 'country': r.country, 'hub': r.hub,
'deliveries': r.deliveries, 'exceptions': r.exceptions,
'onTimeCount': r.onTimeCount, 'lateCount': r.lateCount,
'totalSubtotalUsdMinor': r.totalSubtotalUsdMinor,
'totalShippingUsdMinor': r.totalShippingUsdMinor,
'totalDistanceKm': r.totalDistanceKm,
'totalDelayHours': r.totalDelayHours,
'consentValid': r.consentValid, 'consentMissing': r.consentMissing, 'consentExpired': r.consentExpired,
'sizeTierEnvelope': r.sizeTierEnvelope, 'sizeTierSmall': r.sizeTierSmall,
'sizeTierMedium': r.sizeTierMedium, 'sizeTierLarge': r.sizeTierLarge, 'sizeTierFreight': r.sizeTierFreight,
'shipmentCount': r.shipmentCount,
})),
'journeyAccumulators': [...this.journeyAccumulators.entries()].map(([id, acc]) => ({
'id': id,
'scans': acc.scans.map((s) => ({
'scanSeq': s.scanSeq, 'epochMs': s.epochMs, 'localIso': s.localIso,
'utcOffset': s.utcOffset, 'timezone': s.timezone, 'jurisdiction': s.jurisdiction,
'status': s.status, 'hub': s.hub, 'region': s.region, 'country': s.country,
'lat': s.lat, 'lng': s.lng, 'legKm': s.legKm, 'disruptionReason': s.disruptionReason,
})),
'scanCount': acc.scanCount, 'pathKm': acc.pathKm,
'minEpoch': acc.minEpoch, 'maxEpoch': acc.maxEpoch,
'offsets': [...acc.offsets], 'timezones': [...acc.timezones], 'jurisdictions': [...acc.jurisdictions],
'statusProgression': [...acc.statusProgression],
'delivered': acc.delivered, 'etaCaptured': acc.etaCaptured, 'onTime': acc.onTime,
'delayHours': acc.delayHours, 'subtotalUsdMinor': acc.subtotalUsdMinor, 'shippingUsdMinor': acc.shippingUsdMinor,
})),
'errorRollup': {
'total': this.errorRollup.total,
'groups': [...this.errorRollup.groups.entries()].map(([key, g]) => ({
'key': key,
'source': g.source, 'variant': g.variant, 'count': g.count,
'samples': [...g.samples], 'sampleInput': g.sampleInput,
})),
},
};
}
protected override restoreData(snap: JsonObjectType): void {
if (typeof snap['eventCount'] === 'number') this.eventCount = snap['eventCount'];
if (typeof snap['useStreamingSource'] === 'boolean') this.useStreamingSource = snap['useStreamingSource'];
if (typeof snap['streamCount'] === 'number') this.streamCount = snap['streamCount'];
if (typeof snap['streamChannelCapacity'] === 'number') this.streamChannelCapacity = snap['streamChannelCapacity'];
if (Array.isArray(snap['sources'])) {
this.sources = snap['sources'].map((s) => CartographerState.sourceFromJson(CartographerState.asObject(s) ?? {}));
}
if (Array.isArray(snap['ingestBuckets'])) {
this.ingestBuckets = snap['ingestBuckets'].map((bucket) =>
Array.isArray(bucket)
? bucket.map((e) => CartographerState.variantFromJson(CartographerState.asObject(e) ?? {}))
: [],
);
}
if (Array.isArray(snap['canonicalEvents'])) {
this.canonicalEvents = snap['canonicalEvents'].map((e) => CartographerState.variantFromJson(CartographerState.asObject(e) ?? {}));
}
if (Array.isArray(snap['eventConfig'])) {
const loadedEvtCfg: EventTypeConfig = snap['eventConfig']
.map((e) => CartographerState.asObject(e))
.filter((e): e is JsonObjectType => e !== null)
.map((e) => {
const mixRaw = Array.isArray(e['formatMix']) ? e['formatMix'] : [];
const formatMix = mixRaw
.map((m) => CartographerState.asObject(m))
.filter((m): m is JsonObjectType => m !== null)
.map((m): { readonly format: 'csv' | 'json' | 'ndjson' | 'yaml'; readonly compression: 'none' | 'gzip'; readonly weight: number } => ({
'format': CartographerState.sourceFormat(m['format']),
'compression': m['compression'] === 'gzip' ? 'gzip' : 'none',
'weight': CartographerState.num(m['weight'], 1),
}));
return {
'eventType': CartographerState.canonicalEventType(e['eventType']),
'count': CartographerState.num(e['count']),
'formatMix': formatMix,
};
});
if (loadedEvtCfg.length > 0) this.eventConfig = loadedEvtCfg;
}
if (Array.isArray(snap['records'])) {
this.records = snap['records'].map((r) => CartographerState.enrichedFromJson(CartographerState.asObject(r) ?? {}));
}
if (Array.isArray(snap['sampleRecords'])) {
this.sampleRecords = snap['sampleRecords'].map((r) => CartographerState.enrichedFromJson(CartographerState.asObject(r) ?? {}));
}
const enObj = CartographerState.asObject(snap['enriched']);
if (enObj !== null) this.enriched = CartographerState.enrichedFromJson(enObj);
if (Array.isArray(snap['insights'])) {
this.insights = new Map(
snap['insights']
.map((e) => CartographerState.asObject(e))
.filter((e): e is JsonObjectType => e !== null)
.map((e): [string, RegionInsights] => [
CartographerState.str(e['key']),
{
'region': CartographerState.str(e['region']),
'country': CartographerState.str(e['country']),
'hub': CartographerState.str(e['hub']),
'deliveries': CartographerState.num(e['deliveries']),
'exceptions': CartographerState.num(e['exceptions']),
'onTimeCount': CartographerState.num(e['onTimeCount']),
'lateCount': CartographerState.num(e['lateCount']),
'totalSubtotalUsdMinor': CartographerState.num(e['totalSubtotalUsdMinor']),
'totalShippingUsdMinor': CartographerState.num(e['totalShippingUsdMinor']),
'totalDistanceKm': CartographerState.num(e['totalDistanceKm']),
'totalDelayHours': CartographerState.num(e['totalDelayHours']),
'consentValid': CartographerState.num(e['consentValid']),
'consentMissing': CartographerState.num(e['consentMissing']),
'consentExpired': CartographerState.num(e['consentExpired']),
'sizeTierEnvelope': CartographerState.num(e['sizeTierEnvelope']),
'sizeTierSmall': CartographerState.num(e['sizeTierSmall']),
'sizeTierMedium': CartographerState.num(e['sizeTierMedium']),
'sizeTierLarge': CartographerState.num(e['sizeTierLarge']),
'sizeTierFreight': CartographerState.num(e['sizeTierFreight']),
'shipmentCount': CartographerState.num(e['shipmentCount']),
},
])
);
}
if (Array.isArray(snap['journeyAccumulators'])) {
this.journeyAccumulators = new Map(
snap['journeyAccumulators']
.map((e) => CartographerState.asObject(e))
.filter((e): e is JsonObjectType => e !== null)
.map((e): [string, JourneyAccumulator] => [
CartographerState.str(e['id']),
{
'scans': Array.isArray(e['scans'])
? e['scans']
.map((s) => CartographerState.asObject(s))
.filter((s): s is JsonObjectType => s !== null)
.map((s): JourneyScan => ({
'scanSeq': CartographerState.num(s['scanSeq']),
'epochMs': CartographerState.num(s['epochMs']),
'localIso': CartographerState.str(s['localIso']),
'utcOffset': CartographerState.str(s['utcOffset']),
'timezone': CartographerState.str(s['timezone']),
'jurisdiction': CartographerState.str(s['jurisdiction']),
'status': CartographerState.str(s['status']),
'hub': CartographerState.str(s['hub']),
'region': CartographerState.str(s['region']),
'country': CartographerState.str(s['country']),
'lat': CartographerState.num(s['lat']),
'lng': CartographerState.num(s['lng']),
'legKm': CartographerState.num(s['legKm']),
'disruptionReason': CartographerState.str(s['disruptionReason']),
}))
: [],
'scanCount': CartographerState.num(e['scanCount']),
'pathKm': CartographerState.num(e['pathKm']),
'minEpoch': CartographerState.num(e['minEpoch']),
'maxEpoch': CartographerState.num(e['maxEpoch']),
'offsets': CartographerState.strArr(e['offsets']),
'timezones': CartographerState.strArr(e['timezones']),
'jurisdictions': CartographerState.strArr(e['jurisdictions']),
'statusProgression': CartographerState.strArr(e['statusProgression']),
'delivered': CartographerState.bool(e['delivered']),
'etaCaptured': CartographerState.bool(e['etaCaptured']),
'onTime': CartographerState.bool(e['onTime']),
'delayHours': CartographerState.num(e['delayHours']),
'subtotalUsdMinor': CartographerState.num(e['subtotalUsdMinor']),
'shippingUsdMinor': CartographerState.num(e['shippingUsdMinor']),
},
])
);
}
const rollupRaw = CartographerState.asObject(snap['errorRollup']);
if (rollupRaw !== null) {
const groupsRaw = Array.isArray(rollupRaw['groups']) ? rollupRaw['groups'] : [];
const groups = new Map<string, ErrorGroupType>(
groupsRaw
.map((g) => CartographerState.asObject(g))
.filter((g): g is JsonObjectType => g !== null)
.map((g): [string, ErrorGroupType] => [
CartographerState.str(g['key']),
{
'source': CartographerState.str(g['source']),
'variant': CartographerState.str(g['variant']),
'count': CartographerState.num(g['count']),
'samples': CartographerState.strArr(g['samples']),
'sampleInput': CartographerState.str(g['sampleInput']),
},
])
);
this.errorRollup = { 'total': CartographerState.num(rollupRaw['total']), 'groups': groups };
}
}
// ── Scalar narrowing helpers (no blanket `as unknown as` casts) ────────────
/** Type predicate: narrows `unknown` to `JsonObjectType` via structural runtime checks. */
private static isJsonObject(value: unknown): value is JsonObjectType {
return value !== null && value !== undefined && typeof value === 'object' && !Array.isArray(value);
}
private static asObject(value: unknown): JsonObjectType | null {
return CartographerState.isJsonObject(value) ? value : null;
}
private static str(value: unknown, defaultValue: string = ''): string {
return typeof value === 'string' ? value : defaultValue;
}
private static num(value: unknown, defaultValue: number = 0): number {
return typeof value === 'number' ? value : defaultValue;
}
private static bool(value: unknown, defaultValue: boolean = false): boolean {
return typeof value === 'boolean' ? value : defaultValue;
}
private static strArr(value: unknown): string[] {
return Array.isArray(value) ? value.filter((v): v is string => typeof v === 'string') : [];
}
// ── CanonicalEventVariant / SourcePayload narrowers + reconstruction ──────────────
private static canonicalEventType(value: unknown): CanonicalEventVariant['eventType'] {
return value === 'position-ping' || value === 'facility-scan' || value === 'sensor-reading'
|| value === 'customs-event' || value === 'delivery-confirmation'
? value
: 'position-ping';
}
private static sourceFormat(value: unknown): SourcePayload['format'] {
return value === 'csv' || value === 'json' || value === 'ndjson' || value === 'yaml' ? value : 'json';
}
private static canonicalSourceFormat(value: unknown): CanonicalEventVariant['sourceFormat'] {
return value === 'csv' || value === 'json' || value === 'ndjson' || value === 'yaml' ? value : 'json';
}
private static canonicalSourceCompression(value: unknown): CanonicalEventVariant['sourceCompression'] {
return value === 'none' || value === 'gzip' ? value : 'none';
}
// ── Dispatch maps (replace switches on eventType) ──────────────────────────
private static readonly cloneVariantDispatch: Readonly<Record<string, (v: CanonicalEventVariant) => CanonicalEventVariant>> = {
'position-ping': (v) => {
if (v.eventType !== 'position-ping') return v;
const copy: PositionPingEvent = { ...CartographerState.variantEnvelope(v), 'eventType': 'position-ping', 'body': { ...v.body } };
if (v.geo !== undefined) copy.geo = { ...v.geo };
if (v.consentHandled !== undefined) copy.consentHandled = v.consentHandled;
if (v.pii !== undefined) copy.pii = v.pii;
return copy;
},
'facility-scan': (v) => {
if (v.eventType !== 'facility-scan') return v;
const copy: FacilityScanEvent = { ...CartographerState.variantEnvelope(v), 'eventType': 'facility-scan', 'body': { ...v.body, 'lineItems': v.body.lineItems.map((li) => ({ ...li })) } };
if (v.geo !== undefined) copy.geo = { ...v.geo };
if (v.consentHandled !== undefined) copy.consentHandled = v.consentHandled;
if (v.pii !== undefined) copy.pii = v.pii;
return copy;
},
'sensor-reading': (v) => {
if (v.eventType !== 'sensor-reading') return v;
const copy: SensorReadingEvent = { ...CartographerState.variantEnvelope(v), 'eventType': 'sensor-reading', 'body': { ...v.body } };
if (v.geo !== undefined) copy.geo = { ...v.geo };
if (v.consentHandled !== undefined) copy.consentHandled = v.consentHandled;
if (v.pii !== undefined) copy.pii = v.pii;
return copy;
},
'customs-event': (v) => {
if (v.eventType !== 'customs-event') return v;
const copy: CustomsEvent = { ...CartographerState.variantEnvelope(v), 'eventType': 'customs-event', 'body': { ...v.body } };
if (v.geo !== undefined) copy.geo = { ...v.geo };
if (v.consentHandled !== undefined) copy.consentHandled = v.consentHandled;
if (v.pii !== undefined) copy.pii = v.pii;
return copy;
},
'delivery-confirmation': (v) => {
if (v.eventType !== 'delivery-confirmation') return v;
const copy: DeliveryConfirmationEvent = { ...CartographerState.variantEnvelope(v), 'eventType': 'delivery-confirmation', 'body': { ...v.body } };
if (v.geo !== undefined) copy.geo = { ...v.geo };
if (v.consentHandled !== undefined) copy.consentHandled = v.consentHandled;
if (v.pii !== undefined) copy.pii = v.pii;
return copy;
},
};
private static readonly variantToJsonBodyDispatch: Readonly<Record<string, (v: CanonicalEventVariant) => JsonObjectType>> = {
'position-ping': (v) => {
if (v.eventType !== 'position-ping') return {};
return {
'scanSeq': v.body.scanSeq, 'latitude': v.body.latitude, 'longitude': v.body.longitude,
'ipAddress': v.body.ipAddress, 'localeTag': v.body.localeTag, 'countryCode': v.body.countryCode,
'legFromLat': v.body.legFromLat, 'legFromLng': v.body.legFromLng,
'originLat': v.body.originLat, 'originLng': v.body.originLng, 'destLat': v.body.destLat, 'destLng': v.body.destLng,
'carrier': v.body.carrier, 'status': v.body.status, 'rawTimestamp': v.body.rawTimestamp,
'address': v.body.address, 'phone': v.body.phone,
};
},
'facility-scan': (v) => {
if (v.eventType !== 'facility-scan') return {};
return {
'scanSeq': v.body.scanSeq, 'latitude': v.body.latitude, 'longitude': v.body.longitude,
'ipAddress': v.body.ipAddress, 'localeTag': v.body.localeTag, 'countryCode': v.body.countryCode,
'legFromLat': v.body.legFromLat, 'legFromLng': v.body.legFromLng,
'originLat': v.body.originLat, 'originLng': v.body.originLng, 'destLat': v.body.destLat, 'destLng': v.body.destLng,
'carrier': v.body.carrier, 'status': v.body.status, 'rawTimestamp': v.body.rawTimestamp,
'facilityId': v.body.facilityId, 'weight': v.body.weight, 'weightUnit': v.body.weightUnit,
'lineItems': v.body.lineItems.map((li) => ({ 'productId': li.productId, 'quantity': li.quantity })),
'rawDispatchAt': v.body.rawDispatchAt, 'rawPromisedDeliveryAt': v.body.rawPromisedDeliveryAt,
'disruptionReason': v.body.disruptionReason,
'recipientName': v.body.recipientName, 'recipientEmail': v.body.recipientEmail,
'recipientPhone': v.body.recipientPhone, 'recipientAddress': v.body.recipientAddress,
'recipientCountry': v.body.recipientCountry, 'marketingConsent': v.body.marketingConsent,
'lawfulBasis': v.body.lawfulBasis, 'specialCategory': v.body.specialCategory,
'address': v.body.address, 'phone': v.body.phone,
};
},
'sensor-reading': (v) => {
if (v.eventType !== 'sensor-reading') return {};
return {
'scanSeq': v.body.scanSeq, 'latitude': v.body.latitude, 'longitude': v.body.longitude,
'ipAddress': v.body.ipAddress, 'localeTag': v.body.localeTag, 'countryCode': v.body.countryCode,
'legFromLat': v.body.legFromLat, 'legFromLng': v.body.legFromLng,
'originLat': v.body.originLat, 'originLng': v.body.originLng, 'destLat': v.body.destLat, 'destLng': v.body.destLng,
'carrier': v.body.carrier, 'status': v.body.status, 'rawTimestamp': v.body.rawTimestamp,
'tempC': v.body.tempC, 'humidityPct': v.body.humidityPct, 'shockG': v.body.shockG,
'address': v.body.address, 'phone': v.body.phone,
};
},
'customs-event': (v) => {
if (v.eventType !== 'customs-event') return {};
return {
'scanSeq': v.body.scanSeq, 'latitude': v.body.latitude, 'longitude': v.body.longitude,
'ipAddress': v.body.ipAddress, 'localeTag': v.body.localeTag, 'countryCode': v.body.countryCode,
'legFromLat': v.body.legFromLat, 'legFromLng': v.body.legFromLng,
'originLat': v.body.originLat, 'originLng': v.body.originLng, 'destLat': v.body.destLat, 'destLng': v.body.destLng,
'carrier': v.body.carrier, 'status': v.body.status, 'rawTimestamp': v.body.rawTimestamp,
'customsStatus': v.body.customsStatus,
'address': v.body.address, 'phone': v.body.phone,
};
},
'delivery-confirmation': (v) => {
if (v.eventType !== 'delivery-confirmation') return {};
return {
'scanSeq': v.body.scanSeq, 'latitude': v.body.latitude, 'longitude': v.body.longitude,
'ipAddress': v.body.ipAddress, 'localeTag': v.body.localeTag, 'countryCode': v.body.countryCode,
'legFromLat': v.body.legFromLat, 'legFromLng': v.body.legFromLng,
'originLat': v.body.originLat, 'originLng': v.body.originLng, 'destLat': v.body.destLat, 'destLng': v.body.destLng,
'carrier': v.body.carrier, 'status': v.body.status, 'rawTimestamp': v.body.rawTimestamp,
'delivered': v.body.delivered, 'rawPromisedDeliveryAt': v.body.rawPromisedDeliveryAt,
'disruptionReason': v.body.disruptionReason,
'recipientName': v.body.recipientName, 'recipientEmail': v.body.recipientEmail,
'recipientPhone': v.body.recipientPhone, 'recipientAddress': v.body.recipientAddress,
'recipientCountry': v.body.recipientCountry, 'marketingConsent': v.body.marketingConsent,
'lawfulBasis': v.body.lawfulBasis, 'specialCategory': v.body.specialCategory,
'address': v.body.address, 'phone': v.body.phone,
};
},
};
private static readonly variantFromJsonDispatch: Readonly<Record<string, (
envelope: { shipmentId: string; eventId: string; epochMs: number; sourceId: string; sourceFormat: CanonicalEventVariant['sourceFormat']; sourceCompression: CanonicalEventVariant['sourceCompression'] },
sharedBody: { scanSeq: number; latitude: number; longitude: number; ipAddress: string; localeTag: string; countryCode: string; legFromLat: number; legFromLng: number; originLat: number; originLng: number; destLat: number; destLng: number; carrier: string; status: string; rawTimestamp: string; address: string; phone: string },
b: Record<string, unknown>,
o: Record<string, unknown>,
) => CanonicalEventVariant>> = {
'facility-scan': (envelope, sharedBody, b, o) => {
const variant: FacilityScanEvent = {
...envelope, 'eventType': 'facility-scan',
'body': {
...sharedBody,
'facilityId': CartographerState.str(b['facilityId']),
'weight': CartographerState.num(b['weight']),
'weightUnit': CartographerState.weightUnit(b['weightUnit']),
'lineItems': CartographerState.lineItemsFromJson(b['lineItems']),
'rawDispatchAt': CartographerState.str(b['rawDispatchAt']),
'rawPromisedDeliveryAt': CartographerState.str(b['rawPromisedDeliveryAt']),
'disruptionReason': CartographerState.str(b['disruptionReason']),
'recipientName': CartographerState.str(b['recipientName']),
'recipientEmail': CartographerState.str(b['recipientEmail']),
'recipientPhone': CartographerState.str(b['recipientPhone']),
'recipientAddress': CartographerState.str(b['recipientAddress']),
'recipientCountry': CartographerState.str(b['recipientCountry']),
'marketingConsent': CartographerState.bool(b['marketingConsent']),
'lawfulBasis': CartographerState.lawfulBasis(b['lawfulBasis']),
'specialCategory': CartographerState.specialCategory(b['specialCategory']),
},
};
const geoObj = CartographerState.asObject(o['geo']);
if (geoObj !== null) variant.geo = { 'country': CartographerState.str(geoObj['country']), 'continent': CartographerState.str(geoObj['continent']), 'region': CartographerState.str(geoObj['region']) };
if (typeof o['consentHandled'] === 'boolean') variant.consentHandled = o['consentHandled'];
if (typeof o['pii'] === 'boolean') variant.pii = o['pii'];
return variant;
},
'sensor-reading': (envelope, sharedBody, b, o) => {
const variant: SensorReadingEvent = {
...envelope, 'eventType': 'sensor-reading',
'body': {
...sharedBody,
'tempC': CartographerState.num(b['tempC']),
'humidityPct': CartographerState.num(b['humidityPct']),
'shockG': CartographerState.num(b['shockG']),
},
};
const geoObj = CartographerState.asObject(o['geo']);
if (geoObj !== null) variant.geo = { 'country': CartographerState.str(geoObj['country']), 'continent': CartographerState.str(geoObj['continent']), 'region': CartographerState.str(geoObj['region']) };
if (typeof o['consentHandled'] === 'boolean') variant.consentHandled = o['consentHandled'];
if (typeof o['pii'] === 'boolean') variant.pii = o['pii'];
return variant;
},
'customs-event': (envelope, sharedBody, b, o) => {
const variant: CustomsEvent = {
...envelope, 'eventType': 'customs-event',
'body': { ...sharedBody, 'customsStatus': CartographerState.str(b['customsStatus']) },
};
const geoObj = CartographerState.asObject(o['geo']);
if (geoObj !== null) variant.geo = { 'country': CartographerState.str(geoObj['country']), 'continent': CartographerState.str(geoObj['continent']), 'region': CartographerState.str(geoObj['region']) };
if (typeof o['consentHandled'] === 'boolean') variant.consentHandled = o['consentHandled'];
if (typeof o['pii'] === 'boolean') variant.pii = o['pii'];
return variant;
},
'delivery-confirmation': (envelope, sharedBody, b, o) => {
const variant: DeliveryConfirmationEvent = {
...envelope, 'eventType': 'delivery-confirmation',
'body': {
...sharedBody,
'delivered': CartographerState.bool(b['delivered']),
'rawPromisedDeliveryAt': CartographerState.str(b['rawPromisedDeliveryAt']),
'disruptionReason': CartographerState.str(b['disruptionReason']),
'recipientName': CartographerState.str(b['recipientName']),
'recipientEmail': CartographerState.str(b['recipientEmail']),
'recipientPhone': CartographerState.str(b['recipientPhone']),
'recipientAddress': CartographerState.str(b['recipientAddress']),
'recipientCountry': CartographerState.str(b['recipientCountry']),
'marketingConsent': CartographerState.bool(b['marketingConsent']),
'lawfulBasis': CartographerState.lawfulBasis(b['lawfulBasis']),
'specialCategory': CartographerState.specialCategory(b['specialCategory']),
},
};
const geoObj = CartographerState.asObject(o['geo']);
if (geoObj !== null) variant.geo = { 'country': CartographerState.str(geoObj['country']), 'continent': CartographerState.str(geoObj['continent']), 'region': CartographerState.str(geoObj['region']) };
if (typeof o['consentHandled'] === 'boolean') variant.consentHandled = o['consentHandled'];
if (typeof o['pii'] === 'boolean') variant.pii = o['pii'];
return variant;
},
'position-ping': (envelope, sharedBody, _b, o) => {
const variant: PositionPingEvent = { ...envelope, 'eventType': 'position-ping', 'body': { ...sharedBody } };
const geoObj = CartographerState.asObject(o['geo']);
if (geoObj !== null) variant.geo = { 'country': CartographerState.str(geoObj['country']), 'continent': CartographerState.str(geoObj['continent']), 'region': CartographerState.str(geoObj['region']) };
if (typeof o['consentHandled'] === 'boolean') variant.consentHandled = o['consentHandled'];
if (typeof o['pii'] === 'boolean') variant.pii = o['pii'];
return variant;
},
};
/** Extract the shared envelope fields from a CanonicalEventVariant (used by dispatch maps). */
private static variantEnvelope(v: CanonicalEventVariant): {
shipmentId: string; eventId: string; epochMs: number;
sourceId: string; sourceFormat: CanonicalEventVariant['sourceFormat']; sourceCompression: CanonicalEventVariant['sourceCompression'];
} {
return {
'shipmentId': v.shipmentId,
'eventId': v.eventId,
'epochMs': v.epochMs,
'sourceId': v.sourceId,
'sourceFormat': v.sourceFormat,
'sourceCompression': v.sourceCompression,
};
}
private static resolveCloneHandler(eventType: string): (v: CanonicalEventVariant) => CanonicalEventVariant {
return CartographerState.cloneVariantDispatch[eventType]
?? CartographerState.cloneVariantDispatch['position-ping']
?? ((vv: CanonicalEventVariant): CanonicalEventVariant => ({ ...vv }));
}
/** Deep-clone a CanonicalEventVariant (dispatch map on eventType to keep each member's exact shape). */
private static cloneVariant(v: CanonicalEventVariant): CanonicalEventVariant {
return CartographerState.resolveCloneHandler(v.eventType)(v);
}
private static resolveToJsonBodyHandler(eventType: string): (v: CanonicalEventVariant) => JsonObjectType {
return CartographerState.variantToJsonBodyDispatch[eventType]
?? CartographerState.variantToJsonBodyDispatch['position-ping']
?? (() => ({}));
}
/** Serialize a CanonicalEventVariant to a JSON-safe object (dispatch map on eventType for exact body fields). */
private static variantToJson(v: CanonicalEventVariant): JsonObjectType {
const bodyHandler = CartographerState.resolveToJsonBodyHandler(v.eventType);
return {
'shipmentId': v.shipmentId,
'eventId': v.eventId,
'epochMs': v.epochMs,
'eventType': v.eventType,
'sourceId': v.sourceId,
'sourceFormat': v.sourceFormat,
'sourceCompression': v.sourceCompression,
'geo': v.geo !== undefined ? { 'country': v.geo.country, 'continent': v.geo.continent, 'region': v.geo.region } : null,
'consentHandled': v.consentHandled !== undefined ? v.consentHandled : null,
'pii': v.pii !== undefined ? v.pii : null,
'body': bodyHandler(v),
};
}
private static resolveFromJsonHandler(eventType: string): (
envelope: { shipmentId: string; eventId: string; epochMs: number; sourceId: string; sourceFormat: CanonicalEventVariant['sourceFormat']; sourceCompression: CanonicalEventVariant['sourceCompression'] },
sharedBody: { scanSeq: number; latitude: number; longitude: number; ipAddress: string; localeTag: string; countryCode: string; legFromLat: number; legFromLng: number; originLat: number; originLng: number; destLat: number; destLng: number; carrier: string; status: string; rawTimestamp: string; address: string; phone: string },
b: Record<string, unknown>,
o: Record<string, unknown>,
) => CanonicalEventVariant {
const pp = (
envelope: { shipmentId: string; eventId: string; epochMs: number; sourceId: string; sourceFormat: CanonicalEventVariant['sourceFormat']; sourceCompression: CanonicalEventVariant['sourceCompression'] },
sharedBody: { scanSeq: number; latitude: number; longitude: number; ipAddress: string; localeTag: string; countryCode: string; legFromLat: number; legFromLng: number; originLat: number; originLng: number; destLat: number; destLng: number; carrier: string; status: string; rawTimestamp: string; address: string; phone: string },
_b: Record<string, unknown>,
o: Record<string, unknown>,
): CanonicalEventVariant => {
const variant: PositionPingEvent = { ...envelope, 'eventType': 'position-ping', 'body': { ...sharedBody } };
const geoObj = CartographerState.asObject(o['geo']);
if (geoObj !== null) variant.geo = { 'country': CartographerState.str(geoObj['country']), 'continent': CartographerState.str(geoObj['continent']), 'region': CartographerState.str(geoObj['region']) };
if (typeof o['consentHandled'] === 'boolean') variant.consentHandled = o['consentHandled'];
if (typeof o['pii'] === 'boolean') variant.pii = o['pii'];
return variant;
};
return CartographerState.variantFromJsonDispatch[eventType] ?? pp;
}
/** Reconstruct a CanonicalEventVariant from a deserialized JSON object (dispatch map on eventType). */
private static variantFromJson(o: Record<string, unknown>): CanonicalEventVariant {
const eventType = typeof o['eventType'] === 'string' ? o['eventType'] : 'position-ping';
const b = CartographerState.asObject(o['body']) ?? {};
const envelope = {
'shipmentId': CartographerState.str(o['shipmentId']),
'eventId': CartographerState.str(o['eventId']),
'epochMs': CartographerState.num(o['epochMs']),
'sourceId': CartographerState.str(o['sourceId']),
'sourceFormat': CartographerState.canonicalSourceFormat(o['sourceFormat']),
'sourceCompression': CartographerState.canonicalSourceCompression(o['sourceCompression']),
};
const sharedBody = {
'scanSeq': CartographerState.num(b['scanSeq']),
'latitude': CartographerState.num(b['latitude']),
'longitude': CartographerState.num(b['longitude']),
'ipAddress': CartographerState.str(b['ipAddress']),
'localeTag': CartographerState.str(b['localeTag']),
'countryCode': CartographerState.str(b['countryCode']),
'legFromLat': CartographerState.num(b['legFromLat']),
'legFromLng': CartographerState.num(b['legFromLng']),
'originLat': CartographerState.num(b['originLat']),
'originLng': CartographerState.num(b['originLng']),
'destLat': CartographerState.num(b['destLat']),
'destLng': CartographerState.num(b['destLng']),
'carrier': CartographerState.str(b['carrier']),
'status': CartographerState.str(b['status']),
'rawTimestamp': CartographerState.str(b['rawTimestamp']),
'address': CartographerState.str(b['address']),
'phone': CartographerState.str(b['phone']),
};
return CartographerState.resolveFromJsonHandler(eventType)(envelope, sharedBody, b, o);
}
private static sourceToJson(s: SourcePayload): JsonObjectType {
return {
'sourceId': s.sourceId,
'format': s.format,
'compression': s.compression,
'mappingKey': s.mappingKey,
'eventType': s.eventType,
'payload': s.payload,
};
}
private static sourceFromJson(o: Record<string, unknown>): SourcePayload {
return {
'sourceId': CartographerState.str(o['sourceId']),
'format': CartographerState.sourceFormat(o['format']),
'compression': (o['compression'] === 'none' || o['compression'] === 'gzip') ? o['compression'] : 'none',
'mappingKey': CartographerState.str(o['mappingKey'], 'json-position'),
'eventType': CartographerState.canonicalEventType(o['eventType']),
'payload': CartographerState.str(o['payload']),
};
}
private static geoStatus(value: unknown): GeoContext['status'] {
return value === 'land' || value === 'water' || value === 'coastal' || value === 'unmapped'
? value
: 'unmapped';
}
private static consentStatus(value: unknown): GdprResult['consentStatus'] {
return value === 'valid' || value === 'missing' || value === 'expired' ? value : 'missing';
}
private static lawfulBasis(value: unknown): GdprResult['lawfulBasis'] {
return value === 'contract' || value === 'consent' || value === 'legitimate-interest' || value === 'none'
? value
: 'contract';
}
private static jurisdiction(value: unknown): GeoContext['jurisdiction'] {
return value === 'GDPR' || value === 'UK-GDPR' || value === 'CCPA'
|| value === 'LGPD' || value === 'APPI' || value === 'baseline'
|| value === 'international-waters'
? value
: 'baseline';
}
private static lifecycleStatus(value: unknown): NormalizedShipment['status'] {
return value === 'SCAN' || value === 'DEPARTURE' || value === 'ARRIVAL'
|| value === 'OUT_FOR_DELIVERY' || value === 'DELIVERED' || value === 'EXCEPTION'
? value
: 'SCAN';
}
private static serviceTier(value: unknown): NormalizedShipment['serviceTier'] {
return value === 'express' || value === 'standard' || value === 'economy' ? value : 'standard';
}
private static sizeTier(value: unknown): NormalizedShipment['sizeTier'] {
return value === 'envelope' || value === 'small' || value === 'medium'
|| value === 'large' || value === 'freight'
? value
: 'small';
}
private static weightUnit(value: unknown): RawShipmentEvent['weightUnit'] {
return value === 'lb' || value === 'kg' || value === 'g' || value === 'oz' ? value : 'kg';
}
private static specialCategory(value: unknown): RawShipmentEvent['specialCategory'] {
return value === 'none' || value === 'health' ? value : 'none';
}
private static lineItemsFromJson(value: unknown): Array<{ 'productId': string; 'quantity': number }> {
if (!Array.isArray(value)) return [{ 'productId': '', 'quantity': 1 }];
const items = value
.map((li) => CartographerState.asObject(li))
.filter((li): li is JsonObjectType => li !== null)
.map((li) => ({
'productId': CartographerState.str(li['productId']),
'quantity': CartographerState.num(li['quantity'], 1),
}));
return items.length > 0 ? items : [{ 'productId': '', 'quantity': 1 }];
}
// ── Entity ↔ JSON reconstruction (field-by-field) ──────────────────────────
private static enrichedToJson(e: EnrichedShipment): JsonObjectType {
return {
'shipmentId': e.shipmentId, 'scanSeq': e.scanSeq, 'epochMs': e.epochMs,
'localIso': e.localIso, 'utcOffset': e.utcOffset, 'timezone': e.timezone, 'jurisdiction': e.jurisdiction,
'continent': e.continent, 'region': e.region, 'country': e.country, 'hub': e.hub, 'geoStatus': e.geoStatus,
'lat': e.lat, 'lng': e.lng, 'coordsCoarsened': e.coordsCoarsened, 'legKm': e.legKm,
'status': e.status, 'serviceTier': e.serviceTier, 'sizeTier': e.sizeTier,
'onTime': e.onTime, 'exception': e.exception, 'consentStatus': e.consentStatus,
'disruptionReason': e.disruptionReason,
'subtotalUsdMinor': e.subtotalUsdMinor, 'currency': e.currency,
'shippingUsdMinor': e.shippingUsdMinor, 'distanceKm': e.distanceKm,
'transitHours': e.transitHours, 'delayHours': e.delayHours,
'redactionApplied': e.redactionApplied,
'redactedSample': {
'recipientName': e.redactedSample.recipientName,
'recipientEmail': e.redactedSample.recipientEmail,
'recipientPhone': e.redactedSample.recipientPhone,
},
'routing': CartographerState.routingToJson(e.routing),
};
}
/** The all-false default routing record (single source of truth). */
static defaultRouting(): EnrichedShipment['routing'] {
return {
'path': 'order',
'geoLookupRun': false,
'geoLookupSkipped': false,
'ipGeolocateRun': false,
'ipGeolocateSkipped': false,
'geoConfidence': 0,
'geoModalities': [],
'geoSourceModel': '',
'geoSecondaryLookupUsed': false,
'redactionRun': false,
'redactionSkipped': false,
'pricingRun': false,
'pricingSkipped': false,
'etaRun': false,
'etaSkipped': false,
'coldChainRun': false,
'customsDwellRun': false,
};
}
/** An unresolved GeoCandidate for the given modality (the default). */
static unresolvedCandidate(modality: 'gps' | 'ip'): GeoCandidate {
return {
'modality': modality, 'resolved': false, 'country': '', 'countryName': '',
'continent': '', 'region': '', 'locality': '', 'lat': 0, 'lng': 0, 'water': false,
};
}
private static routingToJson(r: EnrichedShipment['routing']): JsonObjectType {
return {
'path': r.path,
'geoLookupRun': r.geoLookupRun,
'geoLookupSkipped': r.geoLookupSkipped,
'ipGeolocateRun': r.ipGeolocateRun,
'ipGeolocateSkipped': r.ipGeolocateSkipped,
'geoConfidence': r.geoConfidence,
'geoModalities': [...r.geoModalities],
'geoSourceModel': r.geoSourceModel,
'geoSecondaryLookupUsed': r.geoSecondaryLookupUsed,
'redactionRun': r.redactionRun,
'redactionSkipped': r.redactionSkipped,
'pricingRun': r.pricingRun,
'pricingSkipped': r.pricingSkipped,
'etaRun': r.etaRun,
'etaSkipped': r.etaSkipped,
'coldChainRun': r.coldChainRun,
'customsDwellRun': r.customsDwellRun,
};
}
private static routingPath(value: unknown): EnrichedShipment['routing']['path'] {
return value === 'geo-only' || value === 'sensor' || value === 'order' || value === 'customs'
? value
: 'order';
}
private static routingFromJson(value: unknown): EnrichedShipment['routing'] {
const o = CartographerState.asObject(value) ?? {};
return {
'path': CartographerState.routingPath(o['path']),
'geoLookupRun': CartographerState.bool(o['geoLookupRun']),
'geoLookupSkipped': CartographerState.bool(o['geoLookupSkipped']),
'ipGeolocateRun': CartographerState.bool(o['ipGeolocateRun']),
'ipGeolocateSkipped': CartographerState.bool(o['ipGeolocateSkipped']),
'geoConfidence': CartographerState.num(o['geoConfidence']),
'geoModalities': CartographerState.strArr(o['geoModalities']),
'geoSourceModel': CartographerState.str(o['geoSourceModel']),
'geoSecondaryLookupUsed': CartographerState.bool(o['geoSecondaryLookupUsed']),
'redactionRun': CartographerState.bool(o['redactionRun']),
'redactionSkipped': CartographerState.bool(o['redactionSkipped']),
'pricingRun': CartographerState.bool(o['pricingRun']),
'pricingSkipped': CartographerState.bool(o['pricingSkipped']),
'etaRun': CartographerState.bool(o['etaRun']),
'etaSkipped': CartographerState.bool(o['etaSkipped']),
'coldChainRun': CartographerState.bool(o['coldChainRun']),
'customsDwellRun': CartographerState.bool(o['customsDwellRun']),
};
}
private static enrichedFromJson(o: Record<string, unknown>): EnrichedShipment {
const sample = CartographerState.asObject(o['redactedSample']) ?? {};
return {
'shipmentId': CartographerState.str(o['shipmentId']),
'scanSeq': CartographerState.num(o['scanSeq']),
'epochMs': CartographerState.num(o['epochMs']),
'localIso': CartographerState.str(o['localIso']),
'utcOffset': CartographerState.str(o['utcOffset']),
'timezone': CartographerState.str(o['timezone'], 'UTC'),
'jurisdiction': CartographerState.jurisdiction(o['jurisdiction']),
'continent': CartographerState.str(o['continent'], 'Unmapped'),
'region': CartographerState.str(o['region']),
'country': CartographerState.str(o['country']),
'hub': CartographerState.str(o['hub']),
'geoStatus': CartographerState.geoStatus(o['geoStatus']),
'lat': CartographerState.num(o['lat']),
'lng': CartographerState.num(o['lng']),
'coordsCoarsened': CartographerState.bool(o['coordsCoarsened']),
'legKm': CartographerState.num(o['legKm']),
'status': CartographerState.lifecycleStatus(o['status']),
'serviceTier': CartographerState.serviceTier(o['serviceTier']),
'sizeTier': CartographerState.sizeTier(o['sizeTier']),
'onTime': CartographerState.bool(o['onTime']),
'exception': CartographerState.bool(o['exception']),
'consentStatus': CartographerState.consentStatus(o['consentStatus']),
'disruptionReason': CartographerState.str(o['disruptionReason']),
'subtotalUsdMinor': CartographerState.num(o['subtotalUsdMinor']),
'currency': CartographerState.str(o['currency'], 'USD'),
'shippingUsdMinor': CartographerState.num(o['shippingUsdMinor']),
'distanceKm': CartographerState.num(o['distanceKm']),
'transitHours': CartographerState.num(o['transitHours']),
'delayHours': CartographerState.num(o['delayHours']),
'redactionApplied': CartographerState.bool(o['redactionApplied']),
'redactedSample': {
'recipientName': CartographerState.str(sample['recipientName']),
'recipientEmail': CartographerState.str(sample['recipientEmail']),
'recipientPhone': CartographerState.str(sample['recipientPhone']),
},
'routing': CartographerState.routingFromJson(o['routing']),
};
}
}CartographerServices
The dependency record passed into node constructors. Services carry two transport adapters: ipGeolocator (live freeipapi.com or committed fixture replay) and addressGeocoder (live OpenStreetMap Nominatim or deterministic no-answer in the smoke). Coords, locale, code, and phone resolution are fully offline — GeohashTzMap, CoordTimezone, and CallingCode need no injected transport.
ts
export interface CartographerServices {
/** IP-modality transport (geolocate gateway IP → place). */
readonly ipGeolocator: IpGeolocator;
/** Address-modality transport (forward-geocode postal address → place). */
readonly addressGeocoder: AddressGeocoder;
}GeoResolvers
Factory that assembles the CartographerServices record for the chosen backend.
ts
export class GeoResolvers {
/** Live services: live freeipapi IP geolocation and Nominatim address geocoding. */
static live(): CartographerServices {
return {
'ipGeolocator': new LiveIpGeolocator(),
'addressGeocoder': new LiveAddressGeocoder(),
};
}
/** Recorded services: fixture-replay IP geolocation and deterministic address geocoding.
* Deterministic and offline — used by the smoke and `--recorded` CLI flag. */
static recorded(): CartographerServices {
return {
'ipGeolocator': new RecordedIpGeolocator(),
'addressGeocoder': new RecordedAddressGeocoder(),
};
}
}Key nodes
sourceIntake — multi-entry gather intake
Each data-type entrypoint targets intake-gather directly. The scheduler seeds one gather record per entrypoint label; the source-intake gather opens those per-type streams and merges them into state.sources. There is no seed pre-phase, no pre-node, and no scatter before the gather.
ts
/**
* sourceIntake: stream construction helpers for Cartographer's open intake gather.
*
* The intake gather is entrypoint-driven: each data-type entrypoint is a
* canonical entrypoint IRI, and the gather/source helpers open the matching
* typed source stream directly.
*/
import type { CartographerState } from '../CartographerState.ts';
import type { SourcePayload } from '../entities/SourcePayload.ts';
import type { CanonicalEventVariant } from '../entities/index.ts';
import { EventStreamSource } from '../services/EventStreamSource.ts';
import { CARTOGRAPHER_IRIS } from '../cartographerIds.ts';
import type { GatherRecordType } from '@studnicky/dagonizer/contracts';
import type { NodeStateInterface } from '@studnicky/dagonizer/types';
type CartographerEventType = CanonicalEventVariant['eventType'];
type CartographerIntakeState = NodeStateInterface & Pick<CartographerState, 'eventConfig' | 'streamCount'>;
class SourcePayloadStream {
private constructor() { /* static-only */ }
static async *roundRobin(streams: readonly AsyncIterable<SourcePayload>[]): AsyncIterable<SourcePayload> {
const active = streams.map((stream) => stream[Symbol.asyncIterator]());
while (active.length > 0) {
for (let i = 0; i < active.length;) {
const step = await active[i]?.next();
if (step === undefined || step.done === true) {
active.splice(i, 1);
continue;
}
yield step.value;
i++;
}
}
}
static async *skip(stream: AsyncIterable<SourcePayload>, count: number): AsyncIterable<SourcePayload> {
let skipped = 0;
for await (const item of stream) {
if (skipped < count) {
skipped++;
continue;
}
yield item;
}
}
static async *empty(): AsyncIterable<SourcePayload> {
return;
}
}
export class CartographerSourceIntake {
private constructor() { /* static-only */ }
static isState(state: NodeStateInterface): state is CartographerIntakeState {
const eventConfig = Reflect.get(state, 'eventConfig');
return eventConfig !== null
&& typeof eventConfig === 'object'
&& typeof Reflect.get(state, 'streamCount') === 'number';
}
static streamFor(state: CartographerIntakeState, eventType: CartographerEventType): AsyncIterable<SourcePayload> {
const totalCount = state.streamCount > 0 ? state.streamCount : undefined;
return CartographerSourceIntake.filterType(
EventStreamSource.streamTyped(state.eventConfig, totalCount),
eventType,
);
}
static mergedFor(state: CartographerState, resumeAfter: number = 0): AsyncIterable<SourcePayload> {
const streams = CARTOGRAPHER_IRIS.intakeEventTypes.map((eventType) =>
CartographerSourceIntake.streamFor(state, eventType),
);
const merged = SourcePayloadStream.roundRobin(streams);
return resumeAfter > 0 ? SourcePayloadStream.skip(merged, resumeAfter) : merged;
}
static mergeRecords(
records: readonly GatherRecordType[],
state: NodeStateInterface,
): AsyncIterable<SourcePayload> {
if (!CartographerSourceIntake.isState(state)) return SourcePayloadStream.empty();
const streams: AsyncIterable<SourcePayload>[] = [];
for (const source of CARTOGRAPHER_IRIS.intakeEventTypes) {
const record = records.find((candidate) => CartographerSourceIntake.sourceType(candidate.source) === source);
if (record === undefined) continue;
streams.push(CartographerSourceIntake.streamFor(state, source));
}
return SourcePayloadStream.roundRobin(streams);
}
private static sourceType(source: string): CartographerEventType | null {
const marker = '/entrypoint/';
const index = source.indexOf(marker);
if (index < 0) return null;
const label = decodeURIComponent(source.slice(index + marker.length));
return CARTOGRAPHER_IRIS.intakeEventTypes.includes(label as CartographerEventType)
? (label as CartographerEventType)
: null;
}
private static async *filterType(
stream: AsyncIterable<SourcePayload>,
eventType: CartographerEventType,
): AsyncIterable<SourcePayload> {
for await (const item of stream) {
if (item.eventType === eventType) yield item;
}
}
}ts
export class SourceIntakeGather extends GatherStrategy {
readonly name = 'source-intake';
readonly '@id' = 'urn:noocodec:node:source-intake';
override initial(
_config: GatherConfigType,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
accessor.set(state, 'sources', []);
}
override reduce(
_config: GatherConfigType,
batch: Parameters<GatherStrategy['reduce']>[1],
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
const records: GatherRecordType[] = [];
for (const item of batch) records.push(item.state);
accessor.set(state, 'sources', CartographerSourceIntake.mergeRecords(records, state));
}
}
GatherStrategies.register(new SourceIntakeGather());canonicalizeCore — timestamp and location normalization
After geo-enrichment sets state.geoContext.timezone, canonicalizeCore converts the raw timestamp to a UTC epoch, then derives the local time at the scan's IANA timezone using Intl.DateTimeFormat. Cross-zone journeys show different local times and UTC offsets per scan.
ts
export class CanonicalizeCoreNode extends MonadicNode<CartographerState, 'normalized' | 'rejected'> {
readonly '@id' = 'urn:noocodec:node:canonicalize-core';
readonly 'name' = 'canonicalize-core';
readonly 'outputs' = ['normalized', 'rejected'] as const;
override get outputSchema(): Record<'normalized' | 'rejected', SchemaObjectType> {
return {
'normalized': { 'type': 'object' },
'rejected': { 'type': 'object' },
};
}
override async execute(
batch: Batch<CartographerState>,
_context: NodeContextType,
): Promise<RoutedBatchType<'normalized' | 'rejected', CartographerState>> {
const acc = new Map<'normalized' | 'rejected', ItemType<CartographerState>[]>();
for (const item of batch) {
const result = this.routeItem(item.state);
for (const error of result.errors) {
item.state.collectError(error);
}
const bucket = acc.get(result.output);
if (bucket === undefined) {
acc.set(result.output, [item]);
} else {
bucket.push(item);
}
}
const routed = new Map<'normalized' | 'rejected', Batch<CartographerState>>();
for (const [output, items] of acc) {
routed.set(output, Batch.from(items));
}
return routed;
}
private routeItem(state: CartographerState): NodeOutputType<'normalized' | 'rejected'> {
const raw = state.raw;
const epochMs = TimeNormalizer.toEpochMs(raw.rawTimestamp);
if (!isFinite(epochMs) || epochMs <= 0) {
return NodeOutput.create('rejected');
}
const dispatchEpochMs = TimeNormalizer.toEpochMs(raw.rawDispatchAt);
const validDispatch = isFinite(dispatchEpochMs) && dispatchEpochMs > 0 ? dispatchEpochMs : epochMs;
const promisedEpochMs = TimeNormalizer.toEpochMs(raw.rawPromisedDeliveryAt);
const validPromised = isFinite(promisedEpochMs) && promisedEpochMs > 0 ? promisedEpochMs : validDispatch + 7 * 86_400_000;
const { localIso, utcOffset } = TimeZoneResolver.localParts(epochMs, state.geoContext.timezone);
const { carrierId, carrierName } = CarrierRegistry.canonical(raw.carrier);
const countryIso3 = CountryCodes.toIso3(raw.recipientCountry);
const disruptionHours = Disruptions.hoursFor(raw.disruptionReason);
// Derive classification at weightGrams=0. canonicalizeFacility will override
// weightGrams/serviceTier/sizeTier with the real weight for facility-scan events.
const status = EventClassifier.eventType(raw.rawStatus);
const serviceTier = EventClassifier.serviceTier(carrierId, 0);
const sizeTier = EventClassifier.sizeTier(0);
state.normalized = {
'shipmentId': raw.shipmentId,
'scanSeq': raw.scanSeq,
'epochMs': epochMs,
'dispatchEpochMs': validDispatch,
'isoTimestamp': TimeNormalizer.toIso(epochMs),
'localIso': localIso,
'utcOffset': utcOffset,
'carrierId': carrierId,
'carrierName': carrierName,
'countryIso3': countryIso3,
'weightGrams': 0,
'status': status,
'serviceTier': serviceTier,
'sizeTier': sizeTier,
'lineItems': [{ 'productId': '', 'quantity': 1 }],
'facilityId': '',
'latitude': raw.latitude,
'longitude': raw.longitude,
'legFromLat': raw.legFromLat,
'legFromLng': raw.legFromLng,
'originLat': raw.originLat,
'originLng': raw.originLng,
'destLat': raw.destLat,
'destLng': raw.destLng,
'recipientName': '',
'recipientEmail': '',
'recipientPhone': '',
'recipientAddress': '',
'recipientCountry': raw.recipientCountry,
'marketingConsent': false,
'promisedEpochMs': validPromised,
'disruptionHours': disruptionHours,
'disruptionReason': raw.disruptionReason,
};
return NodeOutput.create('normalized');
}
}
export const canonicalizeCore = new CanonicalizeCoreNode();aggregateEvent — writes the enriched record
Pulls every enrichment result out of the clone's state and assembles the compact EnrichedShipment record. The routing decisions, redacted PII sample, and pricing/ shipping/ETA figures all land here.
ts
export class AggregateEventNode extends MonadicNode<CartographerState, 'done'> {
readonly '@id' = 'urn:noocodec:node:aggregate-event';
readonly 'name' = 'aggregate-event';
readonly 'outputs' = ['done'] as const;
private static readonly DEFAULT_GEO_LABEL = 'Unmapped';
override get outputSchema(): Record<'done', SchemaObjectType> {
return {
'done': { 'type': 'object' },
};
}
override async execute(
batch: Batch<CartographerState>,
_context: NodeContextType,
): Promise<RoutedBatchType<'done', CartographerState>> {
for (const item of batch) {
const state = item.state;
const norm = state.normalized;
const geo = state.geoContext;
const gdpr = state.gdprResult;
const po = state.pricedOrder;
const sq = state.shippingQuote;
const de = state.deliveryEstimate;
const ev = state.currentEvent;
const isException = norm.status === 'EXCEPTION';
state.enriched = {
'shipmentId': norm.shipmentId,
'scanSeq': norm.scanSeq,
'epochMs': norm.epochMs,
'localIso': norm.localIso,
'utcOffset': norm.utcOffset,
'timezone': geo.timezone,
'jurisdiction': geo.jurisdiction,
// Macro continent for the per-region insights rollup (from the real API).
'continent': geo.continent || AggregateEventNode.DEFAULT_GEO_LABEL,
'region': geo.region || AggregateEventNode.DEFAULT_GEO_LABEL,
'country': geo.country || AggregateEventNode.DEFAULT_GEO_LABEL,
'hub': geo.hub || AggregateEventNode.DEFAULT_GEO_LABEL,
'geoStatus': geo.status,
// Stored coords come from currentEvent, which GDPR coarsened in-place
// when the jurisdiction is strict or consent is not valid.
'lat': ev.latitude,
'lng': ev.longitude,
'coordsCoarsened': gdpr.coordsCoarsened,
'legKm': state.legKm,
'status': norm.status,
'serviceTier': norm.serviceTier,
'sizeTier': norm.sizeTier,
'onTime': de.onTime,
'exception': isException,
'consentStatus': gdpr.consentStatus,
'disruptionReason': norm.disruptionReason,
'subtotalUsdMinor': po.subtotalUsdMinor,
'currency': po.currency,
'shippingUsdMinor': sq.costUsdMinor,
'distanceKm': sq.distanceKm,
'transitHours': de.transitHours,
'delayHours': de.delayHours,
'redactionApplied': gdpr.redactionApplied,
'redactedSample': {
'recipientName': ev.recipientName,
'recipientEmail': ev.recipientEmail,
'recipientPhone': ev.recipientPhone,
},
// This scan's conditional-routing decisions (RAN vs SKIPPED per branch),
// recorded by the route-* nodes on this clone. The parent's summarize
// totals them into the savings view.
'routing': { ...state.routing },
};
}
return RoutedBatch.create('done', batch);
}
}summarizeInsights — finalize insight views
In the streaming path (the browser demo and any caller using insights-fold) the insights-fold gather accumulates state.insights, state.journeys, and state.sampleRecords incrementally as each clone completes, so summarizeInsights is a pure pass-through — it detects the pre-populated maps and routes success immediately. The records-based fold (iterating state.records) is retained as a default path for callers that use the array path without the insights-fold gather. Either way the final state exposes:
- Per-continent rollup (
state.insights): counts, on-time rate, revenue (USD), distance. - Per-journey rollup (
state.journeys): grouped byshipmentId, ordered by epoch; path distance, elapsed time, timezones crossed, jurisdictions traversed.
ts
export class SummarizeInsightsNode extends MonadicNode<CartographerState, 'success'> {
readonly '@id' = 'urn:noocodec:node:summarize';
private static readonly sizeTierDispatch: Readonly<Record<SizeTierKey, (entry: RegionInsights) => void>> = {
'envelope': (entry) => { entry.sizeTierEnvelope++; },
'small': (entry) => { entry.sizeTierSmall++; },
'medium': (entry) => { entry.sizeTierMedium++; },
'large': (entry) => { entry.sizeTierLarge++; },
'freight': (entry) => { entry.sizeTierFreight++; },
};
readonly 'name' = 'summarize';
readonly 'outputs' = ['success'] as const;
override get outputSchema(): Record<'success', SchemaObjectType> {
return {
'success': { 'type': 'object' },
};
}
override async execute(
batch: Batch<CartographerState>,
_context: NodeContextType,
): Promise<RoutedBatchType<'success', CartographerState>> {
for (const item of batch) {
this.summarizeItem(item.state);
}
return RoutedBatch.create('success', batch);
}
private summarizeItem(state: CartographerState): void {
// Streaming path: insights-fold gather already produced state.insights,
// state.journeys, and state.sampleRecords with bounded memory. Nothing to do.
if (state.insights.size > 0 || state.journeys.size > 0) {
return;
}
// Array-path default: fold state.records into insights and journeys for
// callers that did not use the insights-fold gather strategy.
state.insights = new Map<string, RegionInsights>();
state.journeys = new Map<string, JourneyInsights>();
// ── (b) Group scans by shipmentId for per-journey reconstruction ──────────
const scansByShipment = new Map<string, JourneyScan[]>();
for (const record of state.records) {
if (!record.shipmentId) continue;
// ── (a) per-region accumulation (rolled up to CONTINENT) ────────────────
// Bucket by the macro continent the real geo API resolved (not the fine
// subdivision/country), so the table reads ~6–8 rows. Maritime pings (open
// water → no continent) collapse into one 'International Waters / Maritime'
// bucket. The continent is always present (default 'Unmapped' upstream),
// so the key is consistent — never a bare country code or subdivision.
const key = record.geoStatus === 'water'
? 'International Waters / Maritime'
: record.continent;
let entry = state.insights.get(key);
if (entry === undefined) {
entry = {
'region': key,
'country': key,
'hub': key,
'deliveries': 0,
'exceptions': 0,
'onTimeCount': 0,
'lateCount': 0,
'totalSubtotalUsdMinor': 0,
'totalShippingUsdMinor': 0,
'totalDistanceKm': 0,
'totalDelayHours': 0,
'consentValid': 0,
'consentMissing': 0,
'consentExpired': 0,
'sizeTierEnvelope': 0,
'sizeTierSmall': 0,
'sizeTierMedium': 0,
'sizeTierLarge': 0,
'sizeTierFreight': 0,
'shipmentCount': 0,
};
state.insights.set(key, entry);
}
entry.shipmentCount++;
entry.totalSubtotalUsdMinor += record.subtotalUsdMinor;
entry.totalShippingUsdMinor += record.shippingUsdMinor;
entry.totalDistanceKm += record.distanceKm;
// On-time is only meaningful for order-lane events that ran the ETA node;
// position/sensor/customs events skip pricing/eta (the branching saves it),
// so they are NOT counted toward on-time% (which would otherwise read 0%).
if (record.routing.etaRun) {
if (record.onTime) entry.onTimeCount++;
else {
entry.lateCount++;
entry.totalDelayHours += record.delayHours;
}
}
if (record.status === 'DELIVERED') entry.deliveries++;
if (record.exception) entry.exceptions++;
if (record.consentStatus === 'valid') entry.consentValid++;
if (record.consentStatus === 'missing') entry.consentMissing++;
if (record.consentStatus === 'expired') entry.consentExpired++;
SummarizeInsightsNode.sizeTierDispatch[record.sizeTier]?.(entry);
// ── (b) collect the scan for this journey ───────────────────────────────
let scans = scansByShipment.get(record.shipmentId);
if (scans === undefined) {
scans = [];
scansByShipment.set(record.shipmentId, scans);
}
scans.push({
'scanSeq': record.scanSeq,
'epochMs': record.epochMs,
'localIso': record.localIso,
'utcOffset': record.utcOffset,
'timezone': record.timezone,
'jurisdiction': record.jurisdiction,
'status': record.status,
'hub': record.hub,
'region': record.region,
'country': record.country,
'lat': record.lat,
'lng': record.lng,
'legKm': record.legKm,
'disruptionReason': record.disruptionReason,
});
}
// ── (b) build the per-journey aggregates ──────────────────────────────────
for (const [shipmentId, rawScans] of scansByShipment) {
// Order by scanSeq (the authoritative journey order); epochMs is a display
// value that can collapse/reorder under lossy raw timestamp formats, so it
// is only a tiebreak.
const scans = [...rawScans].sort((a, b) => a.scanSeq - b.scanSeq || a.epochMs - b.epochMs);
const first = scans[0];
const last = scans[scans.length - 1];
if (first === undefined || last === undefined) continue;
let pathKm = 0;
let minEpoch = first.epochMs;
let maxEpoch = first.epochMs;
const offsets: string[] = [];
const timezones: string[] = [];
const jurisdictions: string[] = [];
const statusProgression: string[] = [];
let delivered = false;
for (const s of scans) {
pathKm += s.legKm;
if (s.epochMs < minEpoch) minEpoch = s.epochMs;
if (s.epochMs > maxEpoch) maxEpoch = s.epochMs;
if (!offsets.includes(s.utcOffset)) offsets.push(s.utcOffset);
if (!timezones.includes(s.timezone)) timezones.push(s.timezone);
if (!jurisdictions.includes(s.jurisdiction)) jurisdictions.push(s.jurisdiction);
statusProgression.push(s.status);
if (s.status === 'DELIVERED') delivered = true;
}
// Shipment-level facts (on-time, delay, pricing) come from an ORDER-lane
// record of the journey — one that actually ran pricing/eta. Position/
// sensor/customs scans skip that work, so prefer an eta-bearing record;
// use any record only if the journey has no order-lane scan.
const orderRecord = state.records.find(
(r) => r.shipmentId === shipmentId && r.shipmentId.length > 0 && r.routing.etaRun,
);
const deliveryRecord = orderRecord
?? state.records.find((r) => r.shipmentId === shipmentId && r.shipmentId.length > 0);
const onTime = deliveryRecord?.onTime ?? false;
const delayHours = deliveryRecord?.delayHours ?? 0;
const subtotalUsdMinor = deliveryRecord?.subtotalUsdMinor ?? 0;
const shippingUsdMinor = deliveryRecord?.shippingUsdMinor ?? 0;
const journey: JourneyInsights = {
'shipmentId': shipmentId,
'scans': scans,
'scanCount': scans.length,
'pathKm': pathKm,
'firstEpochMs': minEpoch,
'lastEpochMs': maxEpoch,
'elapsedHours': (maxEpoch - minEpoch) / 3_600_000,
'timezones': timezones,
'offsets': offsets,
'jurisdictions': jurisdictions,
'statusProgression': statusProgression,
'lastStatus': last.status,
'lastHub': last.hub,
'delivered': delivered,
'onTime': onTime,
'delayHours': delayHours,
'subtotalUsdMinor': subtotalUsdMinor,
'shippingUsdMinor': shippingUsdMinor,
};
state.journeys.set(shipmentId, journey);
}
}
}
export const summarizeInsights = new SummarizeInsightsNode();Entities
EnrichedShipment — the per-scan enriched record
ts
import type { FromSchema } from 'json-schema-to-ts';
import { Validator } from '@studnicky/dagonizer/validation';
export const EnrichedShipmentSchema = {
'$id': 'https://noocodec.dev/schemas/cartographer/EnrichedShipment',
'$schema': 'https://json-schema.org/draft/2020-12/schema',
'type': 'object',
'required': [
'shipmentId', 'scanSeq', 'epochMs', 'localIso', 'utcOffset', 'timezone', 'jurisdiction',
'continent', 'region', 'country', 'hub', 'geoStatus',
'lat', 'lng', 'coordsCoarsened', 'legKm',
'status', 'serviceTier', 'sizeTier',
'onTime', 'exception', 'consentStatus', 'disruptionReason',
'subtotalUsdMinor', 'currency',
'shippingUsdMinor', 'distanceKm',
'transitHours', 'delayHours',
'redactionApplied', 'redactedSample', 'routing',
],
'properties': {
'shipmentId': { 'type': 'string', 'minLength': 1 },
'scanSeq': { 'type': 'number', 'minimum': 0 },
'epochMs': { 'type': 'number' },
'localIso': { 'type': 'string' },
'utcOffset': { 'type': 'string' },
'timezone': { 'type': 'string' },
'jurisdiction': { 'type': 'string', 'enum': ['GDPR', 'UK-GDPR', 'CCPA', 'LGPD', 'APPI', 'baseline', 'international-waters'] },
// Macro continent (from a real API) — the per-region insights table buckets by this.
'continent': { 'type': 'string', 'minLength': 1 },
'region': { 'type': 'string', 'minLength': 1 },
'country': { 'type': 'string', 'minLength': 1 },
'hub': { 'type': 'string', 'minLength': 1 },
'geoStatus': { 'type': 'string', 'enum': ['land', 'water', 'coastal', 'unmapped'] },
'lat': { 'type': 'number' },
'lng': { 'type': 'number' },
'coordsCoarsened': { 'type': 'boolean' },
'legKm': { 'type': 'number', 'minimum': 0 },
'status': { 'type': 'string', 'enum': ['SCAN', 'DEPARTURE', 'ARRIVAL', 'OUT_FOR_DELIVERY', 'DELIVERED', 'EXCEPTION'] },
'serviceTier': { 'type': 'string', 'enum': ['express', 'standard', 'economy'] },
'sizeTier': { 'type': 'string', 'enum': ['envelope', 'small', 'medium', 'large', 'freight'] },
'onTime': { 'type': 'boolean' },
'exception': { 'type': 'boolean' },
'consentStatus': { 'type': 'string', 'enum': ['valid', 'missing', 'expired'] },
'disruptionReason': { 'type': 'string' },
'subtotalUsdMinor': { 'type': 'number', 'minimum': 0 },
'currency': { 'type': 'string', 'minLength': 3, 'maxLength': 3 },
'shippingUsdMinor': { 'type': 'number', 'minimum': 0 },
'distanceKm': { 'type': 'number', 'minimum': 0 },
'transitHours': { 'type': 'number', 'minimum': 0 },
'delayHours': { 'type': 'number', 'minimum': 0 },
'redactionApplied': { 'type': 'boolean' },
'redactedSample': {
'type': 'object',
'required': ['recipientName', 'recipientEmail', 'recipientPhone'],
'properties': {
'recipientName': { 'type': 'string' },
'recipientEmail': { 'type': 'string' },
'recipientPhone': { 'type': 'string' },
},
'additionalProperties': false,
},
// This scan's conditional-routing decisions (RAN vs SKIPPED per branch),
// including REAL geo-API call accounting (reverse-geocode + ip-geolocate).
'routing': {
'type': 'object',
'required': [
'path',
'geoLookupRun', 'geoLookupSkipped',
'ipGeolocateRun', 'ipGeolocateSkipped',
'geoConfidence', 'geoModalities',
'geoSourceModel', 'geoSecondaryLookupUsed',
'redactionRun', 'redactionSkipped',
'pricingRun', 'pricingSkipped',
'etaRun', 'etaSkipped',
'coldChainRun', 'customsDwellRun',
],
'properties': {
// The per-event-type enrichment lane this event took.
'path': { 'type': 'string', 'enum': ['geo-only', 'sensor', 'order', 'customs'] },
// Whether the whole geo-resolve sub-DAG (real API calls) ran or was skipped.
'geoLookupRun': { 'type': 'boolean' },
'geoLookupSkipped': { 'type': 'boolean' },
// Real API-call accounting inside geo-resolve.
'ipGeolocateRun': { 'type': 'boolean' },
'ipGeolocateSkipped': { 'type': 'boolean' },
// Multi-modal fusion outcome carried for the report.
'geoConfidence': { 'type': 'number', 'minimum': 0, 'maximum': 1 },
'geoModalities': { 'type': 'array', 'items': { 'type': 'string' } },
// Source-model classification: which geo signal classify-geo-source selected.
'geoSourceModel': { 'type': 'string' },
// Whether CoordTimezone secondary lookup fired.
'geoSecondaryLookupUsed': { 'type': 'boolean' },
'redactionRun': { 'type': 'boolean' },
'redactionSkipped': { 'type': 'boolean' },
'pricingRun': { 'type': 'boolean' },
'pricingSkipped': { 'type': 'boolean' },
'etaRun': { 'type': 'boolean' },
'etaSkipped': { 'type': 'boolean' },
'coldChainRun': { 'type': 'boolean' },
'customsDwellRun': { 'type': 'boolean' },
},
'additionalProperties': false,
},
},
'additionalProperties': false,
} as const;
export type EnrichedShipment = FromSchema<typeof EnrichedShipmentSchema>;
const enrichedShipmentValidator = Validator.compile<EnrichedShipment>(EnrichedShipmentSchema);
export class EnrichedShipmentGuard {
/**
* Type-guard for EnrichedShipment. Narrows `unknown` to the schema-derived type
* by verifying required fields that consumers rely on after narrowing.
*/
static is(value: unknown): value is EnrichedShipment {
return enrichedShipmentValidator.is(value);
}
}CanonicalEventVariant — the per-type event model
The canonical model is a discriminated union on eventType. Each member carries only the fields its event type owns. Five types are generated:
position-ping— a moving asset's satellite position fix with GPS coordinatesfacility-scan— a parcel scanned at a depot or facility; carries PII and order fieldssensor-reading— cold-chain telemetry (temperature, humidity, shock); triggers the cold-chain checkcustoms-event— a customs clearance or hold event; carriescustomsStatusdelivery-confirmation— proof-of-delivery (the terminal event); carries PII anddelivered: true
Format is an independent axis: each event type specifies a format mix (csv/json/ndjson/yaml with per-format weights and compression), so position-pings might arrive as gzip JSON while facility-scans come as CSV. The same event type can appear in multiple formats in one feed.
ts
import type { SourcePayload } from './SourcePayload.ts';
import {
CustomsEventSchema,
type CustomsEvent,
} from './events/CustomsEvent.ts';
import {
DeliveryConfirmationEventSchema,
type DeliveryConfirmationEvent,
} from './events/DeliveryConfirmationEvent.ts';
import {
FacilityScanEventSchema,
type FacilityScanEvent,
} from './events/FacilityScanEvent.ts';
import {
PositionPingEventSchema,
type PositionPingEvent,
} from './events/PositionPingEvent.ts';
import {
SensorReadingEventSchema,
type SensorReadingEvent,
} from './events/SensorReadingEvent.ts';
import { Validator } from '@studnicky/dagonizer/validation';
export const CanonicalEventVariantSchema = {
'$id': 'https://noocodec.dev/schemas/cartographer/CanonicalEventVariant',
'$schema': 'https://json-schema.org/draft/2020-12/schema',
'oneOf': [
PositionPingEventSchema,
FacilityScanEventSchema,
SensorReadingEventSchema,
CustomsEventSchema,
DeliveryConfirmationEventSchema,
],
} as const;
export type CanonicalEventVariant =
| PositionPingEvent
| FacilityScanEvent
| SensorReadingEvent
| CustomsEvent
| DeliveryConfirmationEvent;
const canonicalEventVariantValidator = Validator.compile<CanonicalEventVariant>(CanonicalEventVariantSchema);
// Complete 'position-ping' default. The producer overrides only what it knows;
// every required envelope/body field is present so the consumer never sees a hole.
const POSITION_PING_DEFAULT: PositionPingEvent = {
'shipmentId': 'unknown',
'eventId': 'unknown',
'epochMs': 0,
'eventType': 'position-ping',
'sourceId': 'unknown',
'sourceFormat': 'json',
'sourceCompression': 'none',
'body': {
'scanSeq': 0,
'latitude': 0,
'longitude': 0,
'ipAddress': '',
'localeTag': '',
'countryCode': '',
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'carrier': '',
'status': '',
'rawTimestamp': '',
'address': '',
'phone': '',
},
};
// Per-type body defaults for fromSourcePayload. Each constant covers ONLY the
// fields owned by that type. Declaration order matches the schema's `required`
// array for V8 shape stability.
const POSITION_PING_BODY_DEFAULT: PositionPingEvent['body'] = {
'scanSeq': 0,
'latitude': 0,
'longitude': 0,
'ipAddress': '',
'localeTag': '',
'countryCode': '',
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'carrier': '',
'status': '',
'rawTimestamp': '',
'address': '',
'phone': '',
};
const FACILITY_SCAN_BODY_DEFAULT: FacilityScanEvent['body'] = {
'scanSeq': 0,
'latitude': 0,
'longitude': 0,
'ipAddress': '',
'localeTag': '',
'countryCode': '',
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'carrier': '',
'status': '',
'rawTimestamp': '',
'facilityId': '',
'weight': 0,
'weightUnit': 'kg',
'lineItems': [],
'rawDispatchAt': '',
'rawPromisedDeliveryAt': '',
'disruptionReason': '',
'recipientName': '',
'recipientEmail': '',
'recipientPhone': '',
'recipientAddress': '',
'recipientCountry': '',
'marketingConsent': false,
'lawfulBasis': 'contract',
'specialCategory': 'none',
'address': '',
'phone': '',
};
const SENSOR_READING_BODY_DEFAULT: SensorReadingEvent['body'] = {
'scanSeq': 0,
'latitude': 0,
'longitude': 0,
'ipAddress': '',
'localeTag': '',
'countryCode': '',
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'carrier': '',
'status': '',
'rawTimestamp': '',
'tempC': 0,
'humidityPct': 0,
'shockG': 0,
'address': '',
'phone': '',
};
const CUSTOMS_EVENT_BODY_DEFAULT: CustomsEvent['body'] = {
'scanSeq': 0,
'latitude': 0,
'longitude': 0,
'ipAddress': '',
'localeTag': '',
'countryCode': '',
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'carrier': '',
'status': '',
'rawTimestamp': '',
'customsStatus': '',
'address': '',
'phone': '',
};
const DELIVERY_BODY_DEFAULT: DeliveryConfirmationEvent['body'] = {
'scanSeq': 0,
'latitude': 0,
'longitude': 0,
'ipAddress': '',
'localeTag': '',
'countryCode': '',
'legFromLat': 0,
'legFromLng': 0,
'originLat': 0,
'originLng': 0,
'destLat': 0,
'destLng': 0,
'carrier': '',
'status': '',
'rawTimestamp': '',
'delivered': false,
'rawPromisedDeliveryAt': '',
'disruptionReason': '',
'recipientName': '',
'recipientEmail': '',
'recipientPhone': '',
'recipientAddress': '',
'recipientCountry': '',
'marketingConsent': false,
'lawfulBasis': 'contract',
'specialCategory': 'none',
'address': '',
'phone': '',
};
interface FromSourcePayloadContext {
readonly envelope: {
readonly shipmentId: string;
readonly eventId: string;
readonly epochMs: number;
readonly sourceId: string;
readonly sourceFormat: PositionPingEvent['sourceFormat'];
readonly sourceCompression: PositionPingEvent['sourceCompression'];
};
readonly decoded: Record<string, unknown>;
readonly sharedGeo: {
readonly scanSeq: number;
readonly latitude: number;
readonly longitude: number;
readonly ipAddress: string;
readonly localeTag: string;
readonly countryCode: string;
readonly legFromLat: number;
readonly legFromLng: number;
readonly originLat: number;
readonly originLng: number;
readonly destLat: number;
readonly destLng: number;
readonly carrier: string;
readonly status: string;
readonly rawTimestamp: string;
readonly address: string;
readonly phone: string;
};
readonly preResolvedGeo: { readonly country: string; readonly continent: string; readonly region: string } | undefined;
}
export class CanonicalEventVariantBuilder {
private static readonly fromSourcePayloadDispatch: Readonly<Record<string, (ctx: FromSourcePayloadContext) => CanonicalEventVariant>> = {
'position-ping': ({ envelope, sharedGeo, preResolvedGeo }) => {
const body: PositionPingEvent['body'] = {
'scanSeq': sharedGeo.scanSeq,
'latitude': sharedGeo.latitude,
'longitude': sharedGeo.longitude,
'ipAddress': sharedGeo.ipAddress,
'localeTag': sharedGeo.localeTag,
'countryCode': sharedGeo.countryCode,
'legFromLat': sharedGeo.legFromLat,
'legFromLng': sharedGeo.legFromLng,
'originLat': sharedGeo.originLat,
'originLng': sharedGeo.originLng,
'destLat': sharedGeo.destLat,
'destLng': sharedGeo.destLng,
'carrier': sharedGeo.carrier,
'status': sharedGeo.status,
'rawTimestamp': sharedGeo.rawTimestamp.length > 0 ? sharedGeo.rawTimestamp : POSITION_PING_BODY_DEFAULT.rawTimestamp,
'address': sharedGeo.address,
'phone': sharedGeo.phone,
};
return { ...envelope, 'eventType': 'position-ping', 'body': body, ...(preResolvedGeo !== undefined && { 'geo': preResolvedGeo }) };
},
'facility-scan': ({ envelope, decoded, sharedGeo, preResolvedGeo }) => {
const body: FacilityScanEvent['body'] = {
'scanSeq': sharedGeo.scanSeq,
'latitude': sharedGeo.latitude,
'longitude': sharedGeo.longitude,
'ipAddress': sharedGeo.ipAddress,
'localeTag': sharedGeo.localeTag,
'countryCode': sharedGeo.countryCode,
'legFromLat': sharedGeo.legFromLat,
'legFromLng': sharedGeo.legFromLng,
'originLat': sharedGeo.originLat,
'originLng': sharedGeo.originLng,
'destLat': sharedGeo.destLat,
'destLng': sharedGeo.destLng,
'carrier': sharedGeo.carrier,
'status': sharedGeo.status,
'rawTimestamp': sharedGeo.rawTimestamp.length > 0 ? sharedGeo.rawTimestamp : FACILITY_SCAN_BODY_DEFAULT.rawTimestamp,
'facilityId': CanonicalEventVariantBuilder.str(decoded['facilityId']),
'weight': CanonicalEventVariantBuilder.num(decoded['weight']),
'weightUnit': CanonicalEventVariantBuilder.weightUnit(decoded['weightUnit']),
'lineItems': CanonicalEventVariantBuilder.lineItems(decoded['lineItems']),
'rawDispatchAt': CanonicalEventVariantBuilder.str(decoded['dispatchRaw']),
'rawPromisedDeliveryAt': CanonicalEventVariantBuilder.str(decoded['promisedRaw']),
'disruptionReason': CanonicalEventVariantBuilder.str(decoded['disruptionReason']),
'recipientName': CanonicalEventVariantBuilder.str(decoded['recipientName']),
'recipientEmail': CanonicalEventVariantBuilder.str(decoded['recipientEmail']),
'recipientPhone': CanonicalEventVariantBuilder.str(decoded['recipientPhone']),
'recipientAddress': CanonicalEventVariantBuilder.str(decoded['recipientAddress']),
'recipientCountry': CanonicalEventVariantBuilder.str(decoded['recipientCountry']),
'marketingConsent': CanonicalEventVariantBuilder.bool(decoded['marketingConsent']),
'lawfulBasis': CanonicalEventVariantBuilder.lawfulBasis(decoded['lawfulBasis']),
'specialCategory': CanonicalEventVariantBuilder.specialCategory(decoded['specialCategory']),
'address': sharedGeo.address,
'phone': sharedGeo.phone,
};
return { ...envelope, 'eventType': 'facility-scan', 'body': body, ...(preResolvedGeo !== undefined && { 'geo': preResolvedGeo }) };
},
'sensor-reading': ({ envelope, decoded, sharedGeo, preResolvedGeo }) => {
const body: SensorReadingEvent['body'] = {
'scanSeq': sharedGeo.scanSeq,
'latitude': sharedGeo.latitude,
'longitude': sharedGeo.longitude,
'ipAddress': sharedGeo.ipAddress,
'localeTag': sharedGeo.localeTag,
'countryCode': sharedGeo.countryCode,
'legFromLat': sharedGeo.legFromLat,
'legFromLng': sharedGeo.legFromLng,
'originLat': sharedGeo.originLat,
'originLng': sharedGeo.originLng,
'destLat': sharedGeo.destLat,
'destLng': sharedGeo.destLng,
'carrier': sharedGeo.carrier,
'status': sharedGeo.status,
'rawTimestamp': sharedGeo.rawTimestamp.length > 0 ? sharedGeo.rawTimestamp : SENSOR_READING_BODY_DEFAULT.rawTimestamp,
'tempC': CanonicalEventVariantBuilder.num(decoded['tempC']),
'humidityPct': CanonicalEventVariantBuilder.num(decoded['humidityPct']),
'shockG': CanonicalEventVariantBuilder.num(decoded['shockG']),
'address': sharedGeo.address,
'phone': sharedGeo.phone,
};
return { ...envelope, 'eventType': 'sensor-reading', 'body': body, ...(preResolvedGeo !== undefined && { 'geo': preResolvedGeo }) };
},
'customs-event': ({ envelope, decoded, sharedGeo, preResolvedGeo }) => {
const body: CustomsEvent['body'] = {
'scanSeq': sharedGeo.scanSeq,
'latitude': sharedGeo.latitude,
'longitude': sharedGeo.longitude,
'ipAddress': sharedGeo.ipAddress,
'localeTag': sharedGeo.localeTag,
'countryCode': sharedGeo.countryCode,
'legFromLat': sharedGeo.legFromLat,
'legFromLng': sharedGeo.legFromLng,
'originLat': sharedGeo.originLat,
'originLng': sharedGeo.originLng,
'destLat': sharedGeo.destLat,
'destLng': sharedGeo.destLng,
'carrier': sharedGeo.carrier,
'status': sharedGeo.status,
'rawTimestamp': sharedGeo.rawTimestamp.length > 0 ? sharedGeo.rawTimestamp : CUSTOMS_EVENT_BODY_DEFAULT.rawTimestamp,
'customsStatus': CanonicalEventVariantBuilder.str(decoded['customsStatus']),
'address': sharedGeo.address,
'phone': sharedGeo.phone,
};
return { ...envelope, 'eventType': 'customs-event', 'body': body, ...(preResolvedGeo !== undefined && { 'geo': preResolvedGeo }) };
},
'delivery-confirmation': ({ envelope, decoded, sharedGeo, preResolvedGeo }) => {
const body: DeliveryConfirmationEvent['body'] = {
'scanSeq': sharedGeo.scanSeq,
'latitude': sharedGeo.latitude,
'longitude': sharedGeo.longitude,
'ipAddress': sharedGeo.ipAddress,
'localeTag': sharedGeo.localeTag,
'countryCode': sharedGeo.countryCode,
'legFromLat': sharedGeo.legFromLat,
'legFromLng': sharedGeo.legFromLng,
'originLat': sharedGeo.originLat,
'originLng': sharedGeo.originLng,
'destLat': sharedGeo.destLat,
'destLng': sharedGeo.destLng,
'carrier': sharedGeo.carrier,
'status': sharedGeo.status,
'rawTimestamp': sharedGeo.rawTimestamp.length > 0 ? sharedGeo.rawTimestamp : DELIVERY_BODY_DEFAULT.rawTimestamp,
'delivered': CanonicalEventVariantBuilder.bool(decoded['delivered']),
'rawPromisedDeliveryAt': CanonicalEventVariantBuilder.str(decoded['promisedRaw']),
'disruptionReason': CanonicalEventVariantBuilder.str(decoded['disruptionReason']),
'recipientName': CanonicalEventVariantBuilder.str(decoded['recipientName']),
'recipientEmail': CanonicalEventVariantBuilder.str(decoded['recipientEmail']),
'recipientPhone': CanonicalEventVariantBuilder.str(decoded['recipientPhone']),
'recipientAddress': CanonicalEventVariantBuilder.str(decoded['recipientAddress']),
'recipientCountry': CanonicalEventVariantBuilder.str(decoded['recipientCountry']),
'marketingConsent': CanonicalEventVariantBuilder.bool(decoded['marketingConsent']),
'lawfulBasis': CanonicalEventVariantBuilder.lawfulBasis(decoded['lawfulBasis']),
'specialCategory': CanonicalEventVariantBuilder.specialCategory(decoded['specialCategory']),
'address': sharedGeo.address,
'phone': sharedGeo.phone,
};
return { ...envelope, 'eventType': 'delivery-confirmation', 'body': body, ...(preResolvedGeo !== undefined && { 'geo': preResolvedGeo }) };
},
};
/**
* Type-guard for CanonicalEventVariant. Narrows `unknown` to the discriminated
* union by verifying the object shape and eventType discriminant.
*/
static is(value: unknown): value is CanonicalEventVariant {
return canonicalEventVariantValidator.is(value);
}
static from(partial: Partial<PositionPingEvent> = {}): CanonicalEventVariant {
return {
...POSITION_PING_DEFAULT,
...partial,
'eventType': 'position-ping',
'body': { ...POSITION_PING_DEFAULT.body, ...partial.body },
};
}
// Private coercion helpers — mirror coerce-types node semantics without
// importing the node. Defined locally so defaults live in one place.
private static str(v: unknown): string {
return typeof v === 'string' ? v : '';
}
private static num(v: unknown): number {
if (typeof v === 'number' && isFinite(v)) return v;
if (typeof v === 'string') { const n = Number(v); return isFinite(n) ? n : 0; }
return 0;
}
private static bool(v: unknown): boolean {
if (typeof v === 'boolean') return v;
if (typeof v === 'string') return v === 'true' || v === '1';
return false;
}
private static lawfulBasis(v: unknown): DeliveryConfirmationEvent['body']['lawfulBasis'] {
return v === 'contract' || v === 'consent' || v === 'legitimate-interest' || v === 'none'
? v : 'contract';
}
private static specialCategory(v: unknown): DeliveryConfirmationEvent['body']['specialCategory'] {
return v === 'health' ? 'health' : 'none';
}
private static weightUnit(v: unknown): FacilityScanEvent['body']['weightUnit'] {
return v === 'lb' || v === 'kg' || v === 'g' || v === 'oz' ? v : 'kg';
}
private static lineItems(v: unknown): Array<{ 'productId': string; 'quantity': number }> {
if (!Array.isArray(v)) return [];
const out: Array<{ 'productId': string; 'quantity': number }> = [];
for (const li of v) {
if (li !== null && typeof li === 'object' && !Array.isArray(li)) {
const o: Record<string, unknown> = { ...li };
out.push({
'productId': CanonicalEventVariantBuilder.str(o['productId']),
'quantity': CanonicalEventVariantBuilder.num(o['quantity']) || 1,
});
}
}
return out;
}
/**
* Build a CanonicalEventVariant from a SourcePayload and its decoded canonical
* record. Switches on payload.eventType (the authoritative type) and constructs
* ONLY the variant member for that type, populating each owned body field from
* the decoded record with a default for any genuinely absent field.
*
* Envelope fields (shipmentId, eventId, epochMs) come from the decoded record.
* Provenance (sourceId, sourceFormat, sourceCompression) comes from the payload.
* The rawTimestamp body field is decoded.epochRaw (the raw timestamp string).
*/
static fromSourcePayload(
payload: SourcePayload,
decoded: Record<string, unknown>,
): CanonicalEventVariant {
const shipmentId = CanonicalEventVariantBuilder.str(decoded['shipmentId']);
const eventId = CanonicalEventVariantBuilder.str(decoded['eventId']);
const epochMs = CanonicalEventVariantBuilder.num(decoded['epochMs']);
const envelope = {
'shipmentId': shipmentId.length > 0 ? shipmentId : 'unknown',
'eventId': eventId.length > 0 ? eventId : 'unknown',
'epochMs': epochMs,
'sourceId': payload.sourceId,
'sourceFormat': payload.format,
'sourceCompression': payload.compression,
} as const;
// Shared geometry fields present on every variant.
const scanSeq = CanonicalEventVariantBuilder.num(decoded['scanSeq']);
const latitude = CanonicalEventVariantBuilder.num(decoded['latitude']);
const longitude = CanonicalEventVariantBuilder.num(decoded['longitude']);
const ipAddress = CanonicalEventVariantBuilder.str(decoded['ipAddress']);
const localeTag = CanonicalEventVariantBuilder.str(decoded['localeTag']);
const countryCode = CanonicalEventVariantBuilder.str(decoded['countryCode']);
const legFromLat = CanonicalEventVariantBuilder.num(decoded['legFromLat']);
const legFromLng = CanonicalEventVariantBuilder.num(decoded['legFromLng']);
const originLat = CanonicalEventVariantBuilder.num(decoded['originLat']);
const originLng = CanonicalEventVariantBuilder.num(decoded['originLng']);
const destLat = CanonicalEventVariantBuilder.num(decoded['destLat']);
const destLng = CanonicalEventVariantBuilder.num(decoded['destLng']);
const carrier = CanonicalEventVariantBuilder.str(decoded['carrier']);
const status = CanonicalEventVariantBuilder.str(decoded['status']);
const rawTimestamp = CanonicalEventVariantBuilder.str(decoded['epochRaw']);
const address = CanonicalEventVariantBuilder.str(decoded['address']);
const phone = CanonicalEventVariantBuilder.str(decoded['phone']);
// Pre-resolved geo from RICH sources (JSON/YAML API with offline country-coder).
// Present when the encoder set geoCountry / geoContinent / geoRegion on the record.
// All three must be non-empty for the geo block to be valid; absent otherwise.
const geoCountry = CanonicalEventVariantBuilder.str(decoded['geoCountry']);
const geoContinent = CanonicalEventVariantBuilder.str(decoded['geoContinent']);
const geoRegion = CanonicalEventVariantBuilder.str(decoded['geoRegion']);
const preResolvedGeo = geoCountry.length > 0 && geoContinent.length > 0 && geoRegion.length > 0
? { 'country': geoCountry, 'continent': geoContinent, 'region': geoRegion }
: undefined;
return CanonicalEventVariantBuilder.resolveFromSourcePayloadHandler(payload.eventType)({
envelope,
decoded,
sharedGeo: {
scanSeq, latitude, longitude, ipAddress, localeTag, countryCode,
legFromLat, legFromLng, originLat, originLng,
destLat, destLng, carrier, status, rawTimestamp, address, phone,
},
preResolvedGeo,
});
}
private static resolveFromSourcePayloadHandler(eventType: string): (ctx: FromSourcePayloadContext) => CanonicalEventVariant {
const positionPingHandler = CanonicalEventVariantBuilder.fromSourcePayloadDispatch['position-ping'];
const positionPingDefault = (ctx: FromSourcePayloadContext): CanonicalEventVariant => {
const body: PositionPingEvent['body'] = {
'scanSeq': ctx.sharedGeo.scanSeq,
'latitude': ctx.sharedGeo.latitude,
'longitude': ctx.sharedGeo.longitude,
'ipAddress': ctx.sharedGeo.ipAddress,
'localeTag': ctx.sharedGeo.localeTag,
'countryCode': ctx.sharedGeo.countryCode,
'legFromLat': ctx.sharedGeo.legFromLat,
'legFromLng': ctx.sharedGeo.legFromLng,
'originLat': ctx.sharedGeo.originLat,
'originLng': ctx.sharedGeo.originLng,
'destLat': ctx.sharedGeo.destLat,
'destLng': ctx.sharedGeo.destLng,
'carrier': ctx.sharedGeo.carrier,
'status': ctx.sharedGeo.status,
'rawTimestamp': ctx.sharedGeo.rawTimestamp.length > 0 ? ctx.sharedGeo.rawTimestamp : POSITION_PING_BODY_DEFAULT.rawTimestamp,
'address': ctx.sharedGeo.address,
'phone': ctx.sharedGeo.phone,
};
return { ...ctx.envelope, 'eventType': 'position-ping', 'body': body, ...(ctx.preResolvedGeo !== undefined && { 'geo': ctx.preResolvedGeo }) };
};
return CanonicalEventVariantBuilder.fromSourcePayloadDispatch[eventType] ?? positionPingHandler ?? positionPingDefault;
}
}GeoContext — geo-enrichment result
ts
import type { FromSchema } from 'json-schema-to-ts';
export const GeoContextSchema = {
'$id': 'https://noocodec.dev/schemas/cartographer/GeoContext',
'$schema': 'https://json-schema.org/draft/2020-12/schema',
'type': 'object',
'required': ['gridZone', 'country', 'continent', 'countries', 'region', 'hub', 'status', 'waterBodies', 'timezone', 'jurisdiction'],
'properties': {
'gridZone': { 'type': 'string', 'minLength': 1 },
'country': { 'type': 'string', 'minLength': 1 },
// Macro continent (from a real API) — the insights table buckets by this.
'continent': { 'type': 'string', 'minLength': 1 },
'countries': { 'type': 'array', 'items': { 'type': 'string' } },
'region': { 'type': 'string', 'minLength': 1 },
'hub': { 'type': 'string', 'minLength': 1 },
'status': { 'type': 'string', 'enum': ['land', 'water', 'coastal', 'unmapped'] },
'waterBodies': { 'type': 'array', 'items': { 'type': 'string' } },
'timezone': { 'type': 'string', 'minLength': 1 },
'jurisdiction': { 'type': 'string', 'enum': ['GDPR', 'UK-GDPR', 'CCPA', 'LGPD', 'APPI', 'baseline', 'international-waters'] },
},
'additionalProperties': false,
} as const;
export type GeoContext = FromSchema<typeof GeoContextSchema>;CLI
bash
# Run with 200 journeys (live IP geolocation when network reachable):
npx tsx examples/the-cartographer/runCartographer.ts
# Force offline / recorded mode:
npx tsx examples/the-cartographer/runCartographer.ts --recorded
# Custom event count:
npx tsx examples/the-cartographer/runCartographer.ts --events 50ts
import { CartographerState } from './CartographerState.ts';
import type { JourneyInsights, RegionInsights } from './CartographerState.ts';
import type { CartographerServices } from './CartographerServices.ts';
import { cartographerBundle, cartographerResumeBundle, cartographerWorkersBundle } from './dag.ts';
import { gdprComplianceBundle } from './embedded-dags/GdprComplianceDAG.ts';
import { GeoSourceResolveDAG } from './embedded-dags/GeoSourceResolveDAG.ts';
import { ingestSourceBundle } from './embedded-dags/IngestSourceDAG.ts';
import { orderEnrichmentBundle } from './embedded-dags/OrderEnrichmentDAG.ts';
import type { EnrichedShipment } from './entities/EnrichedShipment.ts';
import type { ConsoleLogger } from './logger/ConsoleLogger.ts';
import { ObservedCartographer } from './ObservedCartographer.ts';
import { normalizeSourcesPlugin } from './plugins/NormalizeSourcesPlugin.ts';
import { CartographerSourceIntake } from './nodes/sourceIntake.ts';
import type { DagonizerOptionsType } from '@studnicky/dagonizer';
import { GeoResolvers } from './services/GeoResolvers.ts';
import { ErrorRollup, type ErrorRollupType } from './errors/ErrorRollup.ts';
import { DAGError } from '@studnicky/dagonizer/errors';
import { StreamCursor } from '@studnicky/dagonizer/channels';
import { Signal } from '@studnicky/signal';
// ── Parse CLI args ────────────────────────────────────────────────────────────
let eventCount = 200;
let forceRecorded = false;
let useWorkers = process.env['CARTO_WORKERS'] === '1';
let useStreaming = process.env['CARTO_STREAM'] === '1';
let streamCount = 0;
const args = process.argv.slice(2);
for (let i = 0; i < args.length; i++) {
if (args[i] === '--events' && args[i + 1] !== undefined) {
const parsed = parseInt(args[i + 1] ?? '200', 10);
if (!isNaN(parsed) && parsed > 0) eventCount = parsed;
} else if (args[i] === '--recorded') {
forceRecorded = true;
} else if (args[i] === '--workers') {
useWorkers = true;
} else if (args[i] === '--stream') {
useStreaming = true;
} else if (args[i] === '--stream-count' && args[i + 1] !== undefined) {
const parsed = parseInt(args[i + 1] ?? '0', 10);
if (!isNaN(parsed) && parsed > 0) streamCount = parsed;
i++;
} else if (/^\d+$/.test(args[i] ?? '')) {
const parsed = parseInt(args[i] ?? '200', 10);
if (!isNaN(parsed) && parsed > 0) eventCount = parsed;
}
}
// ── CLI utilities ─────────────────────────────────────────────────────────────
/**
* AbortingCartographer: ObservedCartographer subclass for the resume scenario.
*
* Fires an AbortController after N scatter item completions (detected by
* watching for `aggregate-event` nodes inside the `process-stream` scatter body).
* Injection via constructor keeps the abort logic out of the main dispatcher.
*/
class AbortingCartographer extends ObservedCartographer {
readonly #controller: AbortController;
readonly #threshold: number;
#count: number;
constructor(options: DagonizerOptionsType, controller: AbortController, threshold: number) {
super(options);
this.#controller = controller;
this.#threshold = threshold;
this.#count = 0;
}
protected override onNodeEnd(
nodeName: string,
output: string | null,
state: CartographerState,
placementPath: readonly string[],
): void {
super.onNodeEnd(nodeName, output, state, placementPath);
// Count completions of aggregate-event inside the process-stream scatter.
// aggregate-event is the last enrichment node before the scatter body terminal.
if (nodeName === 'aggregate-event' && placementPath.includes('process-stream')) {
if (++this.#count >= this.#threshold) {
this.#controller.abort();
}
}
}
get signal(): AbortSignal {
return this.#controller.signal;
}
}
/** Fixed event count for the interrupted+resume run pair. */
const RESUME_EVENT_COUNT = 40;
/**
* Number of scatter item completions (aggregate-event inside process-stream)
* after which the interrupted run aborts. cartographerResumeDAG has no reservoir,
* so items are dispatched one-at-a-time (ScatterWorkerPool path) and the abort
* signal fires between pulls — giving a non-zero StreamCursor value.
*/
const ABORT_AFTER_ITEMS = 8;
/**
* InsightsFingerprint: deterministic canonical digest of a regional insights Map.
*
* Sorts entries by region → country → hub and emits all numeric fields of each
* RegionInsights plus the string keys. JSON.stringify over the sorted plain array
* gives a stable string suitable for equality comparison.
*/
class InsightsFingerprint {
private constructor() { /* static-only */ }
static of(insights: Map<string, RegionInsights>): string {
const rows = [...insights.values()].sort((a, b) => {
const byRegion = a.region.localeCompare(b.region);
if (byRegion !== 0) return byRegion;
const byCountry = a.country.localeCompare(b.country);
if (byCountry !== 0) return byCountry;
return a.hub.localeCompare(b.hub);
});
const normalized = rows.map((r) => ({
'region': r.region,
'country': r.country,
'hub': r.hub,
'deliveries': r.deliveries,
'exceptions': r.exceptions,
'onTimeCount': r.onTimeCount,
'lateCount': r.lateCount,
'totalSubtotalUsdMinor': r.totalSubtotalUsdMinor,
'totalShippingUsdMinor': r.totalShippingUsdMinor,
'totalDistanceKm': r.totalDistanceKm,
'totalDelayHours': r.totalDelayHours,
'consentValid': r.consentValid,
'consentMissing': r.consentMissing,
'consentExpired': r.consentExpired,
'sizeTierEnvelope': r.sizeTierEnvelope,
'sizeTierSmall': r.sizeTierSmall,
'sizeTierMedium': r.sizeTierMedium,
'sizeTierLarge': r.sizeTierLarge,
'sizeTierFreight': r.sizeTierFreight,
'shipmentCount': r.shipmentCount,
}));
return JSON.stringify(normalized);
}
}
/**
* CartographerResumableScenario: self-contained abort→cursor→resume verification.
*
* Uses `cartographerResumeDAG` (no reservoir) so abort fires mid-scatter, leaving
* acked items in the checkpoint and un-pulled items un-acked.
*
* Baseline — Full streaming pass over all RESUME_EVENT_COUNT items (no abort).
* Produces the reference InsightsFingerprint.
* Step A — Interrupted run: abort after ABORT_AFTER_ITEMS aggregate-event
* completions; read durable cursor from checkpoint.
* Step B — Resume: restore from firstState.snapshot() (carries accumulator +
* checkpoint) and supply the remainder through CartographerSourceIntake.
* Assert cursor > 0 and resumeResult.cursor === null (completed).
* Proof — Compare InsightsFingerprint of resumed state to baseline fingerprint.
* Equal → exactly-once; unequal → throw with full diff.
*/
class CartographerResumableScenario {
private constructor() { /* static-only */ }
/** Register cartographerResumeBundle bundles onto a fresh ObservedCartographer. */
static #buildResumeDispatcher(services: CartographerServices): ObservedCartographer {
const d = new ObservedCartographer({});
d.registerBundle(GeoSourceResolveDAG.build(services.ipGeolocator, services.addressGeocoder));
d.registerBundle(orderEnrichmentBundle);
d.registerBundle(gdprComplianceBundle);
d.registerPlugin(normalizeSourcesPlugin);
d.registerBundle(ingestSourceBundle);
d.registerBundle(cartographerResumeBundle);
return d;
}
static async run(
_dispatcher: ObservedCartographer,
services: CartographerServices,
logger: ConsoleLogger,
_eventCount: number,
): Promise<void> {
logger.info('CartographerResumableScenario', 'run', `Starting streamed-resume verification (${RESUME_EVENT_COUNT} events, abort after ${ABORT_AFTER_ITEMS})`);
// ── Baseline: full streaming pass (no abort) ─────────────────────────────
// Runs the same producer + same event count through the same DAG without
// interruption. Produces the reference accumulator for the exactly-once proof.
const baselineDispatcher = CartographerResumableScenario.#buildResumeDispatcher(services);
const baselineState = new CartographerState();
baselineState.useStreamingSource = true;
baselineState.eventCount = RESUME_EVENT_COUNT;
baselineState.streamCount = RESUME_EVENT_COUNT;
await baselineDispatcher.execute('urn:noocodec:dag:cartographer-resume', baselineState);
const baselineFingerprint = InsightsFingerprint.of(baselineState.insights);
logger.info('CartographerResumableScenario', 'baseline', `Baseline streamed run folded ${baselineState.insights.size} region(s).`);
// ── Step A: Interrupted run ──────────────────────────────────────────────
// AbortingCartographer fires abort after ABORT_AFTER_ITEMS aggregate-event
// completions inside process-stream. cartographerResumeDAG has no reservoir,
// so the ScatterWorkerPool checks abort between item pulls — giving cursor > 0.
const interruptAc = new AbortController();
const abortingDispatcher = new AbortingCartographer({}, interruptAc, ABORT_AFTER_ITEMS);
abortingDispatcher.registerBundle(GeoSourceResolveDAG.build(services.ipGeolocator, services.addressGeocoder));
abortingDispatcher.registerBundle(orderEnrichmentBundle);
abortingDispatcher.registerBundle(gdprComplianceBundle);
abortingDispatcher.registerPlugin(normalizeSourcesPlugin);
abortingDispatcher.registerBundle(ingestSourceBundle);
abortingDispatcher.registerBundle(cartographerResumeBundle);
const firstState = new CartographerState();
firstState.useStreamingSource = true;
firstState.eventCount = RESUME_EVENT_COUNT;
firstState.streamCount = RESUME_EVENT_COUNT;
let interruptedCursor: string | null = null;
try {
const interruptedResult = await abortingDispatcher.execute(
'cartographer-resume', firstState, { 'signal': interruptAc.signal },
);
interruptedCursor = interruptedResult.cursor;
} catch (err) {
if (!(err instanceof DAGError && err.code === 'EXECUTION_ERROR')) throw err;
}
// Read the durable stream cursor from the interrupted checkpoint.
const cursor = StreamCursor.resumeAfter(firstState, 'process-stream');
logger.info(
'CartographerResumableScenario', 'interrupted',
`Interrupted after ${ABORT_AFTER_ITEMS} items. execution cursor='${String(interruptedCursor)}' stream cursor=${cursor}`,
);
process.stdout.write(`ASSERT cursor > 0: ${cursor > 0 ? 'PASS' : 'FAIL'} (cursor=${cursor})\n`);
if (cursor === 0) {
throw new Error('CartographerResumableScenario: cursor is 0 — checkpoint not preserved after abort');
}
// ── Step B: Resume ───────────────────────────────────────────────────────
// Restore from the interrupted snapshot — this is the faithful cross-process
// restart path: the partial insights accumulator AND the SCATTER_PROGRESS_KEY
// checkpoint are both carried by CartographerState.restore(firstState.snapshot()).
// Acked items (below the watermark) already contributed to state.insights and
// are NOT replayed by the engine; the accumulator carry ensures their folds
// survive. Un-acked items in the durable inbox are replayed by the engine.
const resumeDispatcher = CartographerResumableScenario.#buildResumeDispatcher(services);
const resumeState = CartographerState.restore(firstState.snapshot());
resumeState.useStreamingSource = true;
resumeState.eventCount = RESUME_EVENT_COUNT;
resumeState.streamCount = RESUME_EVENT_COUNT;
// Supply the remainder: the top-level intake gather is already complete in
// the restored checkpoint, so resume enters process-stream directly.
resumeState.sources = CartographerSourceIntake.mergedFor(resumeState, cursor);
const resumeResult = await resumeDispatcher.resume('urn:noocodec:dag:cartographer-resume', resumeState, 'process-stream');
logger.info(
'CartographerResumableScenario', 'resume',
`Resume complete. cursor=${String(resumeResult.cursor)} (expected null)`,
);
process.stdout.write(`ASSERT resume completed: ${resumeResult.cursor === null ? 'PASS' : 'FAIL'} (cursor=${String(resumeResult.cursor)})\n`);
if (resumeResult.cursor !== null) {
throw new Error(`CartographerResumableScenario: resume did not complete (cursor='${String(resumeResult.cursor)}')`);
}
// ── Exactly-once proof: compare resumed fingerprint to baseline ───────────
// The fingerprint encodes ALL numeric fields for every region, sorted
// deterministically. Equal → every acked fold was carried (not lost, not
// double-counted); unequal → the accumulator carry is broken.
const resumeFingerprint = InsightsFingerprint.of(resumeState.insights);
const exactlyOnce = resumeFingerprint === baselineFingerprint;
process.stdout.write(`ASSERT exactly-once (resume insights == baseline insights): ${exactlyOnce ? 'PASS' : 'FAIL'}\n`);
if (!exactlyOnce) {
throw new Error(
`CartographerResumableScenario: exactly-once violated — resumed insights differ from baseline.\n` +
` baseline: ${baselineFingerprint}\n` +
` resumed: ${resumeFingerprint}`,
);
}
// ── Shipment-count cross-check ────────────────────────────────────────────
// Grand total of shipmentCount across all regions must be identical between
// the resumed run and the baseline (gross undercount / double-count guard).
let baselineTotal = 0;
for (const r of baselineState.insights.values()) baselineTotal += r.shipmentCount;
let resumeTotal = 0;
for (const r of resumeState.insights.values()) resumeTotal += r.shipmentCount;
const countMatch = resumeTotal === baselineTotal;
process.stdout.write(`ASSERT shipment-count (resume=${resumeTotal} == baseline=${baselineTotal}): ${countMatch ? 'PASS' : 'FAIL'}\n`);
if (!countMatch) {
throw new Error(
`CartographerResumableScenario: shipment-count mismatch — resumed total (${resumeTotal}) != baseline (${baselineTotal})`,
);
}
process.stdout.write('Streamed resume: COMPLETE. Exactly-once verified.\n');
}
}
class CartographerCli {
static async networkReachable(): Promise<boolean> {
try {
const signal = Signal.timeout(4000);
const res = await fetch('https://freeipapi.com/api/json/8.8.8.8', {
signal,
'headers': { 'accept': 'application/json' },
});
return res.ok;
} catch {
return false;
}
}
static fmtCoord(lat: number, lng: number): string {
const ns = `${Math.abs(lat).toFixed(2)}${lat >= 0 ? 'N' : 'S'}`;
const ew = `${Math.abs(lng).toFixed(2)}${lng >= 0 ? 'E' : 'W'}`;
return `${ns} ${ew}`;
}
static printJourney(logger: ConsoleLogger, j: JourneyInsights): void {
const km = Math.round(j.pathKm).toLocaleString('en-US');
const elapsedH = Math.floor(j.elapsedHours);
const elapsedM = Math.round((j.elapsedHours - elapsedH) * 60);
logger.result(`${j.shipmentId} (${j.scanCount} scans, ${j.timezones.length} timezone(s))`);
for (const s of j.scans) {
const time = s.localIso.slice(11, 16);
const cum = `+${Math.round(s.legKm).toLocaleString('en-US')} km`;
logger.result(
` ${time} ${s.utcOffset.padEnd(7)} ${s.status.padEnd(16)} ` +
`${s.hub.slice(0, 18).padEnd(19)} ${CartographerCli.fmtCoord(s.lat, s.lng).padEnd(20)} ${cum}`,
);
}
const tzCrossings = Math.max(0, j.offsets.length - 1);
const jurisLabel = j.jurisdictions.length > 1 ? `${j.jurisdictions.join('→')}` : j.jurisdictions[0] ?? 'baseline';
const otLabel = j.delivered ? (j.onTime ? 'on-time' : `late ${j.delayHours}h`) : `in transit (${j.lastStatus})`;
logger.result(` journey: ${km} km · ${elapsedH}h${String(elapsedM).padStart(2, '0')}m elapsed · ${tzCrossings} tz crossing(s) · jurisdiction ${jurisLabel} · ${otLabel}`);
}
/**
* Error-analysis section: total captured-exception count and a table of
* `source · variant · count · sample-message`, ordered by descending count so the
* dominant error source is first. This is the DAG-flow error collection made
* visible — every captured exception folded by the gather is reported here.
*/
static printErrorAnalysis(logger: ConsoleLogger, rollup: ErrorRollupType): void {
logger.result('=== (e) Error Analysis — captured exceptions folded through the DAG ===\n');
if (rollup.total === 0) {
logger.result(' No exceptions captured this run. (A clean run — zero swallowed faults.)\n');
return;
}
logger.result(` Total captured exceptions: ${rollup.total.toLocaleString('en-US')}\n`);
const COL_SOURCE = 18;
const COL_VARIANT = 14;
const COL_COUNT = 8;
const hdr =
'Source'.padEnd(COL_SOURCE) +
'Variant'.padEnd(COL_VARIANT) +
'Count'.padStart(COL_COUNT) +
' Sample message';
logger.result(` ${hdr}`);
logger.result(` ${'-'.repeat(hdr.length + 24)}`);
for (const group of ErrorRollup.ranked(rollup)) {
const sample = group.samples[0] ?? '';
logger.result(
` ${group.source.slice(0, COL_SOURCE - 1).padEnd(COL_SOURCE)}` +
`${group.variant.slice(0, COL_VARIANT - 1).padEnd(COL_VARIANT)}` +
`${String(group.count).padStart(COL_COUNT)} ${sample}`,
);
}
logger.result('');
}
static printRedaction(logger: ConsoleLogger, label: string, rec: EnrichedShipment): void {
logger.result(` [${label}] ${rec.shipmentId} jurisdiction=${rec.jurisdiction} consent=${rec.consentStatus}`);
logger.result(` Name: ${rec.redactedSample.recipientName}`);
logger.result(` Email: ${rec.redactedSample.recipientEmail}`);
logger.result(` Phone: ${rec.redactedSample.recipientPhone}`);
logger.result(` Coords: ${CartographerCli.fmtCoord(rec.lat, rec.lng)} ${rec.coordsCoarsened ? '(COARSENED to grid centroid)' : '(precise)'}`);
}
}
// ── Geo backend selection: LIVE IP if a network is reachable, else RECORDED ────
// GPS reverse-geocode is ALWAYS offline (the `@rapideditor/country-coder` boundary
// dataset — deterministic, no network) — only the IP modality is a live API call.
// `useLive` selects the live freeipapi.com IP geolocator when reachable; otherwise
// (and with `--recorded`) the recorded IP fixture replays for a deterministic,
// offline run. The probe targets freeipapi (the only live modality).
const useLive = !forceRecorded && (await CartographerCli.networkReachable());
const services: CartographerServices = useLive ? GeoResolvers.live() : GeoResolvers.recorded();
// ── Worker container (only when --workers / CARTO_WORKERS=1) ─────────────────
// WorkerThreadContainer is only imported when the worker path is active. In
// tsx mode (in-process default) this branch is never reached.
//
// The registry module URL is resolved relative to this compiled file's location
// (examples/the-cartographer/dist/runCartographer.js) so the path resolves to
// examples/the-cartographer/dist/workers/eventPipelineRegistry.js — the compiled
// output of workers/eventPipelineRegistry.ts.
//
// Workers receive servicesConfig.useRecordedIp so they select the same IP backend
// as the parent (recorded when offline/--recorded, live when the parent is live).
let workerContainers: Array<{ destroy(): Promise<void> }> = [];
// ── Dispatcher ────────────────────────────────────────────────────────────────
// ObservedCartographer subclasses Dagonizer and wires every lifecycle hook to
// its internal ConsoleLogger — the sanctioned class-extension observability
// demonstration. Progress / status / diagnostic lines flow through the hooks;
// the final tabular report is routed through `dispatcher.logger.result(...)`.
let dispatcher: ObservedCartographer;
if (useWorkers) {
// Dynamic import keeps WorkerThreadContainer out of the tsx bundle; workers
// are only instantiated when the compiled path is active.
const { WorkerThreadContainer } = await import('@studnicky/dagonizer-executor-node');
const registryUrl = new URL('./workers/eventPipelineRegistry.js', import.meta.url).href;
const cpuContainer = new WorkerThreadContainer({
'registryModule': registryUrl,
'registryVersion': '1.0.0',
'servicesConfig': { 'useRecordedIp': !useLive },
'poolSize': 4,
});
const ioContainer = new WorkerThreadContainer({
'registryModule': registryUrl,
'registryVersion': '1.0.0',
'servicesConfig': { 'useRecordedIp': !useLive },
'poolSize': 1,
});
workerContainers = [cpuContainer, ioContainer];
// Workers mode: cartographerWorkersDAG delegates process-stream to `cpu` and
// summarize-insights to `io`. The parent dispatcher still registers the full
// bundle so DAG validation resolves every embedded reference.
dispatcher = new ObservedCartographer({
'containers': {
'cpu': cpuContainer,
'io': ioContainer,
},
});
// Sub-DAG bundles (needed for DAG validator; execution stays in the workers).
dispatcher.registerBundle(GeoSourceResolveDAG.build(services.ipGeolocator, services.addressGeocoder));
dispatcher.registerBundle(orderEnrichmentBundle);
dispatcher.registerBundle(gdprComplianceBundle);
dispatcher.registerPlugin(normalizeSourcesPlugin);
// ingestSourceBundle owns all unique ingest nodes + all format sub-DAGs.
dispatcher.registerBundle(ingestSourceBundle);
// Top-level DAG (cartographerWorkersDAG binds process-stream to cpu and
// summarize-insights to io).
dispatcher.registerBundle(cartographerWorkersBundle);
} else {
dispatcher = new ObservedCartographer({});
dispatcher.registerBundle(GeoSourceResolveDAG.build(services.ipGeolocator, services.addressGeocoder));
dispatcher.registerBundle(orderEnrichmentBundle);
dispatcher.registerBundle(gdprComplianceBundle);
dispatcher.registerPlugin(normalizeSourcesPlugin);
// ingestSourceBundle owns all unique ingest nodes + all format sub-DAGs.
dispatcher.registerBundle(ingestSourceBundle);
dispatcher.registerBundle(cartographerBundle);
}
// The example's own logger, owned by the subclass. Display (the tabular report)
// goes through `logger.result(...)`; diagnostics flow from the hook overrides.
const logger = dispatcher.logger;
const state = new CartographerState();
state.eventCount = eventCount;
state.useStreamingSource = useStreaming;
state.streamCount = streamCount;
const executionMode = useWorkers
? 'WORKER THREADS (containers: cpu pool=4, io pool=1)'
: useStreaming
? `IN-PROCESS + STREAMING SOURCE${streamCount > 0 ? ` (count=${streamCount})` : ''}`
: 'IN-PROCESS (no container)';
// Run-configuration banner: status diagnostics → leveled info on the logger.
logger.info('runCartographer', 'banner', `${String(eventCount)} journeys -> multi-format sources -> fan-in -> streaming enrichment (concurrency=16)`);
logger.info('runCartographer', 'banner', `execution mode: ${executionMode}`);
logger.info('runCartographer', 'banner', `geo backend: offline country-coder reverse-geocode + ${useLive ? 'LIVE freeipapi.com IP geolocation' : 'RECORDED IP fixture replay (offline)'}`);
// ── Execute ───────────────────────────────────────────────────────────────────
const ac = new AbortController();
process.once('SIGINT', () => {
logger.warn('runCartographer', 'onSigint', 'aborting pipeline');
ac.abort();
});
let stageCount = 0;
let peakHeap = process.memoryUsage().heapUsed;
try {
const execution = dispatcher.execute('urn:noocodec:dag:cartographer', state, { 'signal': ac.signal });
for await (const stage of execution) {
const cur = process.memoryUsage().heapUsed;
if (cur > peakHeap) peakHeap = cur;
if (!stage.skipped) {
stageCount++;
// Periodic progress heartbeat — leveled diagnostic, not raw stdout.
// Per-node detail flows from the subclass hooks (onNodeStart/onNodeEnd).
if (stageCount % 80 === 0) {
logger.trace('runCartographer', 'progress', `${String(stageCount)} stages executed`);
}
}
}
await execution;
} catch (err) {
if (err instanceof DAGError && err.code === 'EXECUTION_ERROR') {
logger.fatal('runCartographer', 'execute', `execution failed: ${err.message}`);
process.exit(1);
}
throw err;
}
logger.debug('runCartographer', 'execute', `pipeline drained: ${String(stageCount)} stages executed`);
// Bounded sample of enriched scans (cap 200). state.records is always empty
// in the streaming path (and in the insights-fold non-streaming path). All
// per-scan display sections iterate this sample — honest and memory-bounded.
const sampleProcessed = state.sampleRecords.filter((r) => r.shipmentId.length > 0);
// Total scans folded: exact sum from the insights accumulator (all scans,
// not sampled). The insights fold counts every event regardless of scatter
// concurrency, so this is authoritative.
let totalScans = 0;
for (const r of state.insights.values()) totalScans += r.shipmentCount;
// ── (0) Streaming source summary ──────────────────────────────────────────────
// The streaming topology decodes mixed formats inline per scan; there is no
// separate ingestion fan-in stage. Report what the accumulators know.
logger.result('=== (0) Streaming source — mixed formats decoded inline per scan ===\n');
// Per-event-type lane distribution derived from the bounded sample.
const byEventType = new Map<string, number>();
for (const r of sampleProcessed) {
// The routing path encodes the event-type lane (e.g. "facility-scan/…").
// Extract the first segment as a human-readable lane label.
const lane = r.routing.path.split('/')[0] ?? r.routing.path;
byEventType.set(lane, (byEventType.get(lane) ?? 0) + 1);
}
// Per-format distribution from the bounded sample routing path (the path
// encodes the lane name, not the wire format; use sampleRecords directly
// as a representative distribution indicator).
const distinctFormats = new Set<string>();
if (Array.isArray(state.sources)) {
for (const item of state.sources) {
distinctFormats.add(item.format);
}
}
// Fall back to eventConfig format mix labels when sources is exhausted.
if (distinctFormats.size === 0) {
for (const cfg of state.eventConfig) {
for (const mix of cfg.formatMix) distinctFormats.add(mix.format);
}
}
logger.result(` Total scans folded (exact, from insights accumulator): ${totalScans.toLocaleString()}`);
logger.result(` Continents resolved: ${state.insights.size}`);
logger.result(` Wire formats in feed: ${distinctFormats.size > 0 ? [...distinctFormats].sort().join(', ') : 'mixed (json, csv, ndjson, yaml)'}`);
logger.result(`\n Event-type lane distribution (from a representative sample of ${sampleProcessed.length} scans):`);
for (const lane of [...byEventType.keys()].sort()) {
logger.result(` ${lane.padEnd(28)} ${String(byEventType.get(lane) ?? 0).padStart(5)}`);
}
logger.result('');
// ── (e) Error analysis — the DAG-flow error collection made visible ───────────
// The geo transports and ingest parsers capture every caught exception into a
// GeoErrorRecord on state.errors; the insights-fold gather folds them into
// state.errorRollup (bounded, grouped by source+variant). Print the distribution.
CartographerCli.printErrorAnalysis(logger, state.errorRollup);
// ── (a) Normalization sample — a multi-zone, multi-scan journey ───────────────
// Prefer a journey that crosses >=2 timezones to show differing local offsets.
const multiZoneJourney =
[...state.journeys.values()].find((j) => j.scanCount >= 3 && j.offsets.length >= 2)
?? [...state.journeys.values()].find((j) => j.scanCount >= 2 && j.offsets.length >= 2)
?? [...state.journeys.values()].find((j) => j.scanCount >= 2);
logger.result('=== (a) Normalization Sample — one journey, per-scan LOCAL time ===\n');
if (multiZoneJourney !== undefined) {
logger.result(`${multiZoneJourney.shipmentId} (${multiZoneJourney.scanCount} scans, ${multiZoneJourney.timezones.length} timezone(s), offsets: ${multiZoneJourney.offsets.join(', ')})`);
for (const s of multiZoneJourney.scans) {
const time = s.localIso.slice(11, 16);
logger.result(
` seq ${s.scanSeq} ${time} ${s.utcOffset.padEnd(7)} ${s.status.padEnd(16)} ` +
`${s.hub.slice(0, 18).padEnd(19)} ${CartographerCli.fmtCoord(s.lat, s.lng).padEnd(20)} [${s.jurisdiction}]`,
);
}
}
// ── (b) Per-continent insights table ─────────────────────────────────────────
// Rolled up to the macro continent the real geo API resolved (~6–8 rows), plus a
// single maritime bucket — the precise locality/country stays on each journey scan.
logger.result('\n=== (b) Per-Continent Insights ===\n');
const COL_REGION = 34;
const COL_COUNT = 7;
const COL_EXC = 6;
const COL_ONTIME = 8;
const COL_REV = 12;
const COL_SHIP = 11;
const COL_DIST = 10;
const hdr =
'Continent'.padEnd(COL_REGION) +
'Scans'.padStart(COL_COUNT) +
'Exc'.padStart(COL_EXC) +
'OnTime%'.padStart(COL_ONTIME) +
'Rev $USD'.padStart(COL_REV) +
'Ship $USD'.padStart(COL_SHIP) +
'Dist km'.padStart(COL_DIST);
logger.result(hdr);
logger.result('-'.repeat(hdr.length));
const sortedRegions = [...state.insights.values()].sort((a, b) => a.region.localeCompare(b.region));
for (const r of sortedRegions) {
const total = r.onTimeCount + r.lateCount;
const onTimePct = total > 0 ? Math.round((r.onTimeCount / total) * 100) : 0;
const revUsd = (r.totalSubtotalUsdMinor / 100).toFixed(0);
const shipUsd = (r.totalShippingUsdMinor / 100).toFixed(0);
const distKm = r.totalDistanceKm > 0 ? Math.round(r.totalDistanceKm / r.shipmentCount).toString() : '0';
logger.result(
r.region.slice(0, COL_REGION - 1).padEnd(COL_REGION) +
String(r.shipmentCount).padStart(COL_COUNT) +
String(r.exceptions).padStart(COL_EXC) +
`${onTimePct}%`.padStart(COL_ONTIME) +
`$${revUsd}`.padStart(COL_REV) +
`$${shipUsd}`.padStart(COL_SHIP) +
`${distKm}`.padStart(COL_DIST),
);
}
logger.result(`\nTotal scans folded: ${totalScans.toLocaleString()} · Journeys sampled: ${state.journeys.size}`);
// ── (b2) SOURCE-MODEL ROUTING VIEW (the thesis made tangible — §B0.7c) ──────────
// Each clone records its own RAN/SKIPPED decisions on the enriched record; the
// gather appends each to sampleRecords (bounded FIFO, cap 200). The totals here
// are computed over that representative sample — honest because the sample is a
// cross-section of all event types and routing paths.
//
// Node cost model (the nodes a branch runs/skips per event):
// geo-lookup chain : validate-coords + geo-grid + geo-context = 3 nodes
// (skip path runs apply-geo = 1 node → 2 avoided per skip)
// order enrichment : enrich-pricing + enrich-shipping + enrich-eta = 3 nodes
// redaction sub-DAG: consent-gate + classify-pii + redact-pii = 3 nodes
// (skip path bypasses all 3 directly → 3 avoided)
const GEO_CHAIN_NODES = 3; // validate-coords, geo-grid, geo-context
const GEO_SKIP_ADAPTER = 1; // apply-geo
const ORDER_ENRICH_NODES = 3; // pricing, shipping, eta
const REDACTION_NODES = 3; // consent-gate, classify-pii, redact-pii
const REDACTION_SKIP_ADAPTER = 0; // no intermediate node on skip path
let geoRun = 0, geoSkip = 0, redRun = 0, redSkip = 0, priceSkip = 0, etaSkip = 0;
let coldRun = 0, customsRun = 0;
// Source-model tally
let modelCoords = 0, modelLocale = 0, modelCode = 0, modelIp = 0, modelNone = 0;
let coordsPlusIp = 0, secondaryLookupFired = 0;
let ipgeoRun = 0, ipgeoSkip = 0;
let actualNodes = 0, naiveNodes = 0;
const pathCounts = new Map<string, number>();
for (const r of sampleProcessed) {
const rt = r.routing;
if (rt.geoLookupRun) geoRun++;
if (rt.geoLookupSkipped) geoSkip++;
if (rt.ipGeolocateRun) ipgeoRun++;
if (rt.ipGeolocateSkipped) ipgeoSkip++;
if (rt.redactionRun) redRun++;
if (rt.redactionSkipped) redSkip++;
if (rt.pricingSkipped) priceSkip++;
if (rt.etaSkipped) etaSkip++;
if (rt.coldChainRun) coldRun++;
if (rt.customsDwellRun) customsRun++;
// Source-model tally
if (rt.geoSourceModel === 'coords') modelCoords++;
else if (rt.geoSourceModel === 'locale') modelLocale++;
else if (rt.geoSourceModel === 'code') modelCode++;
else if (rt.geoSourceModel === 'ip') modelIp++;
else modelNone++;
if (rt.geoModalities.includes('ip') && (rt.geoModalities.includes('coords') || rt.geoModalities.includes('geohash'))) coordsPlusIp++;
if (rt.geoSecondaryLookupUsed) secondaryLookupFired++;
pathCounts.set(rt.path, (pathCounts.get(rt.path) ?? 0) + 1);
naiveNodes += GEO_CHAIN_NODES + ORDER_ENRICH_NODES + REDACTION_NODES;
actualNodes += rt.geoLookupRun ? GEO_CHAIN_NODES : GEO_SKIP_ADAPTER;
if (rt.pricingRun) actualNodes += ORDER_ENRICH_NODES;
actualNodes += rt.redactionRun ? REDACTION_NODES : REDACTION_SKIP_ADAPTER;
}
const sampleTotal = sampleProcessed.length;
class Percent {
private constructor() { /* static-only */ }
static of(n: number, base: number): string { return base > 0 ? `${Math.round((n / base) * 100)}%` : '0%'; }
}
const skippedNodes = naiveNodes - actualNodes;
const redactionPassesAvoided = redSkip;
const pricingEtaAvoided = priceSkip * ORDER_ENRICH_NODES;
logger.result('\n=== (b2) Source-Model Routing — from a bounded sample of recent scans ===\n');
logger.result(` Sample size: ${sampleTotal} scans (representative bounded FIFO, cap 200)\n`);
logger.result(` HEADLINE: deterministic routing skipped ${skippedNodes.toLocaleString('en-US')} node-executions in sample ` +
`(~${Percent.of(skippedNodes, naiveNodes)} of the ${naiveNodes.toLocaleString('en-US')} always-run maximum).\n`);
logger.result(' Geo source-model distribution (from classify-geo-source):');
logger.result(` • coords (lat/lng present): ${String(modelCoords).padStart(5)}`);
logger.result(` • code (ISO-2 country code): ${String(modelCode).padStart(5)}`);
logger.result(` • locale (BCP-47 tag): ${String(modelLocale).padStart(5)}`);
logger.result(` • ip (gateway IP only): ${String(modelIp).padStart(5)}`);
logger.result(` • none (no signal): ${String(modelNone).padStart(5)}`);
logger.result('');
logger.result(` coords+IP enriched (dual modality): ${coordsPlusIp}`);
logger.result(` CoordTimezone secondary lookup fired: ${secondaryLookupFired}`);
logger.result('');
logger.result(` geo-lookup: RAN ${geoRun} · SKIPPED ${geoSkip} (${Percent.of(geoSkip, sampleTotal)} — source already resolved → geo sub-DAG avoided)`);
logger.result(` ip-geolocate (freeipapi.com): RAN ${ipgeoRun} · SKIPPED ${ipgeoSkip}`);
logger.result(` redaction: RAN ${redRun} · SKIPPED ${redSkip} (${Percent.of(redSkip, sampleTotal)} — no PII / not required → redaction sub-DAG bypassed)`);
logger.result(` pricing+eta: RAN ${sampleTotal - priceSkip} · SKIPPED ${priceSkip} (${Percent.of(priceSkip, sampleTotal)} — non-order event types carry no basket/delivery)`);
logger.result(` per-event-type lanes: ${[...pathCounts.entries()].sort().map(([p, n]) => `${p}=${n}`).join(' ')}`);
logger.result(` cold-chain-check RAN ${coldRun} (sensor lane only) · customs-dwell RAN ${customsRun} (customs lane only)`);
logger.result('\n Compute avoided in sample:');
logger.result(` • ${redactionPassesAvoided.toLocaleString('en-US')} redaction passes avoided`);
logger.result(` • ${pricingEtaAvoided.toLocaleString('en-US')} pricing/shipping/ETA node-executions avoided — don't price a position ping.`);
// ── (c) Per-journey summaries (a few) ─────────────────────────────────────────
logger.result('\n=== (c) Per-Journey Summaries ===\n');
const journeysSorted = [...state.journeys.values()].sort((a, b) => b.scanCount - a.scanCount);
// Show a few: one multi-tz, one multi-jurisdiction, one delivered.
const shown = new Set<string>();
const picks: JourneyInsights[] = [];
const multiTz = journeysSorted.find((j) => j.offsets.length >= 2);
if (multiTz !== undefined) { picks.push(multiTz); shown.add(multiTz.shipmentId); }
const multiJuris = journeysSorted.find((j) => j.jurisdictions.length >= 2 && !shown.has(j.shipmentId));
if (multiJuris !== undefined) { picks.push(multiJuris); shown.add(multiJuris.shipmentId); }
const deliveredJourney = journeysSorted.find((j) => j.delivered && !shown.has(j.shipmentId));
if (deliveredJourney !== undefined) { picks.push(deliveredJourney); shown.add(deliveredJourney.shipmentId); }
for (const j of picks) {
CartographerCli.printJourney(logger, j);
logger.result('');
}
const tzCrossingJourneys = [...state.journeys.values()].filter((j) => j.offsets.length >= 2).length;
const jurisChangeJourneys = [...state.journeys.values()].filter((j) => j.jurisdictions.length >= 2).length;
logger.result(`Journeys crossing >=2 timezones: ${tzCrossingJourneys}`);
logger.result(`Journeys changing jurisdiction mid-path: ${jurisChangeJourneys}`);
// ── (d) Location-driven redaction comparison ──────────────────────────────────
// Drawn from the bounded sample (cap 200) — sufficient to find representative
// strict-jurisdiction and baseline records.
const strictRecord = sampleProcessed.find(
(r) => r.coordsCoarsened && (r.jurisdiction === 'GDPR' || r.jurisdiction === 'UK-GDPR' || r.jurisdiction === 'LGPD'),
) ?? sampleProcessed.find((r) => r.coordsCoarsened);
const baselineRecord = sampleProcessed.find(
(r) => !r.coordsCoarsened && r.jurisdiction === 'baseline' && r.consentStatus === 'valid',
) ?? sampleProcessed.find((r) => !r.coordsCoarsened);
logger.result('\n=== (d) Location-Driven Redaction (strict vs baseline) ===\n');
if (strictRecord !== undefined) CartographerCli.printRedaction(logger, 'strict', strictRecord);
if (baselineRecord !== undefined) {
logger.result('');
CartographerCli.printRedaction(logger, 'baseline', baselineRecord);
}
logger.result(`\nDone. ${state.insights.size} continent(s), ${state.journeys.size} journey(s). No Date.now. No Math.random.`);
logger.result(`Peak heap: ${Math.round(peakHeap / 1048576)} MB · scans folded: ${totalScans.toLocaleString()} · journeys sampled: ${state.journeys.size} · sampleRecords: ${state.sampleRecords.length}`);
logger.result(`Execution mode: ${executionMode}\n`);
// ── Streamed resume verification (--stream only) ──────────────────────────────
// Runs the three-phase abort→cursor→resume scenario to verify exactly-once
// delivery across a genuine interrupt. Only runs when --stream is passed so it
// does not add latency to the default in-process array path.
if (useStreaming) {
await CartographerResumableScenario.run(dispatcher, services, logger, eventCount);
}
// Release worker pools so the process exits cleanly.
for (const container of workerContainers) {
await container.destroy();
}Details for Nerds
The thesis
Data orchestration = the same engine. LLM-agent workflows and deterministic ETL pipelines are both DAGs of typed nodes with state. The engine does not know or care whether a node calls an LLM, decodes CSV, or runs a haversine formula.
The Cartographer makes the value of the DAG concrete: deterministic conditional routing skips unnecessary work. A position-ping that already carries resolved geo never touches the geo-resolution sub-DAG. An event with no PII never touches the GDPR redaction sub-DAG. The savings are visible in the routing table.
Offline geo resolution
Coords resolution uses two offline primitives — no HTTP, no key, deterministic, identical in Node 18+ and the browser:
GeohashTzMap— a base64-embedded binary geohash→timezone lookup table. The primary fast path: a single table scan resolves lat/lng to an IANA timezone with no network call.CoordTimezone—tz-lookup+@rapideditor/country-coder. The browser-safe default path for border regions and gaps where the geohash table is ambiguous.CoordTimezoneguards theRangeErrorthat out-of-range coords would otherwise raise: when a coord pair falls outside all known boundaries, resolution degrades to an empty timezone/country rather than throwing, and the event continues through the pipeline at baseline.
Locale and code resolution are also fully offline (BCP-47 → IANA via LocaleTimezone; ISO-2 → timezone via CountryLocale). The only live network call in the geo path is IP geolocation (freeipapi.com, CORS-enabled, no key), or committed fixture replay in the smoke tests.
ts
/**
* CoordTimezone: browser-safe WGS-84 coordinate → IANA timezone + ISO country.
*
* Uses tz-lookup (nearest-neighbour IANA zone, browser-safe) for the timezone
* and @rapideditor/country-coder for the ISO 3166-1 alpha-2 country code.
*
* Browser-safe: no node:fs, no node:path, no geo-tz, no Buffer.
*
* @module
*/
import tzlookup from 'tz-lookup';
import { iso1A2Code } from '@rapideditor/country-coder';
export class CoordTimezone {
public static resolve(latitude: number, longitude: number): { timezone: string; country: string } {
let timezone: string;
try {
timezone = tzlookup(latitude, longitude);
} catch {
timezone = '';
}
const countryCode = iso1A2Code([longitude, latitude], { level: 'country' });
const country = countryCode ?? '';
return { timezone: timezone, country: country };
}
}Related Concepts
Read these next when you want to connect Cartographer behavior to scatter, embedded DAGs, workers, streaming, and plugin-defined reusable flows.
- The Archivist - LLM agent orchestration on the same engine
- Concepts - Dagonizer vocabulary the Cartographer exercises
- Example 04: Scatter Scout - streaming scatter + bounded concurrency
- Example 05: Embedded DAGs - nested sub-DAG composition
- Visualization - render a DAG with CytoscapeGraph
Cartographer Feature Map
These numbered examples are the small-form counterparts to Cartographer behavior:
| Example | Principle in the runnable Cartographer |
|---|---|
| Example 04C: Container-Bound Scatter | process-stream is a scatter placement with a container role; the example page isolates the worker-bound body shape. |
| Example 12: Worker Containers | The stream-event body DAG runs through the same DagContainerInterface seam when container roles are bound. |
| Example 13: Multi-Backend Roles | process-stream binds to cpu and summarize-insights binds to io in the browser Cartographer runner while the parent DAG stays JSON-LD. |
| Example 14: Gather Strategies | Cartographer’s InsightsFoldGather and first-class geo-weighted-fusion gather show scatter-local folds and embedded-producer fan-in. |
| Example 15: Incremental Gather | The insights panel updates through incremental fold semantics rather than waiting for a final batch merge. |
| Example 16: Scatter Resume | The durable-inbox model is the checkpoint substrate for long-running stream scatters. |
| Example 17: Async Scatter Source | seed can provide sources as an async stream; bounded scatter pulls only as capacity opens. |
| Example 27: Runtime DAG Dispatch | Dynamic DagReference dispatch belongs here if hierarchical route expansion enters the demo. |
| Example 33: Plugin-Defined DAGs | Plugin packaging belongs here for normalization pipelines (NormalizeCsvDAG, NormalizeJsonDAG, etc.) so plugins and embedded DAGs stay one interface. |
| Examples 34-36: Streaming Substrate | Intake stream assembly, resumable cursors, and DagStreamProducer are the substrate beneath Cartographer’s event stream. |