Skip to main content
13 min read

Mapping Events to Entities

The Event Mapping is the only part you typically write. It defines how incoming events update your scene, while transports, agents, and sessions are simply configuration.

Every mapping answers three questions:

  1. Which events should be processed? — selecting events
  2. Which entities should they affect? — resolving entities
  3. What should be updated? — updating entities

Anatomy of a mapping​

import type { EventMapping } from "@3dverse/livelink-agent";

const mapping: EventMapping = {
    // ── Which events should be processed ──────────────────────────
    channel: "devices/+/telemetry", // optional: a wildcard channel pattern
    when: event => event.payload !== null, // optional: a payload predicate
    schema: {
        /* optional JSON Schema */
    },

    // ── Which entities they affect ────────────────────────────────
    entities: { byName: "device-{id}" },

    // ── What should be updated ────────────────────────────────────
    updates: event => ({
        id: event.channel.split("/")[1],
        update: { local_transform: { position: (event.payload as { pos: [number, number, number] }).pos } },
    }),
};

Only entities and updates are required.

A source can contain multiple event types, each represented by its own mapping. Every incoming event is evaluated against every mapping, and only matching mappings are executed.

What an event looks like​

Whatever the transport, a mapping sees the same envelope:

type IngestEvent = {
    channel: string; // where it arrived — meaning is transport-defined
    payload: unknown; // the JSON-decoded message
    received_at?: Date; // when this process saw it
    source_timestamp?: Date; // when the source says it was produced
    metadata?: Record<string, unknown>; // qos and retain flags, stream offsets...
};

Prefer source_timestamp over received_at for staleness checks: the arrival clock says when this process saw the event, not when the machine produced it.

Selecting events​

A mapping can filter incoming events using:

  • channel — a wildcard pattern matched against the event's channel
  • when — a predicate evaluated on the event payload
  • schema — optional JSON Schema validation

All three are optional, and channel and when combine: a mapping declaring neither handles every event.

By channel​

channel is matched segment by segment against the event's channel. The syntax follows MQTT topic filters, but it applies to every transport:

PatternMatches
+ or *exactly one segment
# (trailing)one or more remaining segments — plant/# matches plant/a, not plant
channel: "plant/+/+/motor"; // plant/line-a/3/motor      ✓
channel: "plant/#"; // plant/line-a/3/motor      ✓

Omit it to handle every channel.

By payload​

Some sources publish every kind of event on a single channel — one Event Hub, one webhook endpoint, an OPC UA server without aliases. when looks inside the message body and returns true for the events this mapping handles:

when: event => (event.payload as { type?: string }).type === "telemetry";

Resolving entities​

Once an event has been selected, the mapping needs to determine which scene entity it should update. Choose exactly one strategy.

StrategyHow the id finds its entityTypical use
byNamelooks up an entity already in the scene by name, from a pattern or a functionthe scene is already named after the stream's ids
byUuida fixed id-to-UUID tablea small, known population — the named parts of one machine
resolvean arbitrary function returning a UUID, or nullan external lookup service, or a naming convention with exceptions
spawnno pre-existing entity: one is created per new id, from a templatethe stream defines the population — a device joining a fleet

They are mutually exclusive: set exactly one.

The lightest option, and the one to reach for first. The pattern embeds an {id} placeholder:

entities: {
    byName: "cell-{id}";
}

The id line-a-3 then drives the entity named cell-line-a-3. Only the literal token {id} is substituted — there is no {0}, no {payload.serial}, no expression syntax. When the name is not a simple substitution, use the function form:

entities: {
    byName: ({ id }) => `cell-${id.toUpperCase()}`;
}

Whenever possible, name your scene entities using the same identifiers as your live data. This avoids maintaining UUID lookup tables and keeps mappings simple — see preparing a scene for live data.

byUuid​

A fixed table, for a closed population you know up front:

entities: {
    byUuid: {
        "servo-01": "1f3c9a4e-...",
        "servo-02": "8b21d0f7-...",
    },
}

resolve​

Anything the two above cannot express — an external lookup, a normalization step, an id that needs decoding. It may be asynchronous, and returns null for an id that addresses nothing:

entities: {
    resolve: async ({ id, event }) => (await assetRegistry.lookup(id))?.entity_uuid ?? null,
}

byName, byUuid and resolve all find entities that are already in the scene, and all accept a linkage — the chain of scene references leading to them, for entities that live inside a sub-scene:

entities: { byName: "{id}", linkage: [SUB_SCENE_UUID] }

spawn​

When the stream itself defines the population, there is nothing to look up. spawn creates one entity per new id, from a template, the first time that id is seen:

entities: {
    spawn: {
        name: "AGV-{id}",
        components: {
            mesh_ref: { value: MESH_UUID },
            local_transform: { position: [0, 0, 0] },
        },
        options: { delete_on_client_disconnection: true },
    },
}

One entity is created per distinct id, the first time that id appears; every later event carrying the same id reuses it. name substitutes {id} exactly as byName does, so a stream carrying the ids 07 and 12 yields two entities, AGV-07 and AGV-12.

name and components both accept a function, which receives the id and the triggering event — so an entity can be created with values read from the very first message rather than snapping into place on the second. The function runs once per id, on that first message only:

components: ({ id, event }) => ({
    debug_name: { value: `AGV-${id}` },
    local_transform: { position: initialPosition(event) },
}),

options are the usual entity creation options. delete_on_client_disconnection: true is worth considering: the spawned entities then vanish when the agent stops, instead of accumulating in the scene.

Spawned entities are always created at the scene root, so spawn takes no linkage. An id removed by a "delete" directive is forgotten, so the next event carrying it spawns a fresh entity.

Updating entities​

The updates function receives the entire event and returns:

  • one entity update
  • several entity updates
  • null to ignore the event

One object per event​

The common case. The id can come from anywhere in the event, because the function sees all of it:

updates: event => ({
    id: (event.payload as { serial: string }).serial, // from the payload
    update: { local_transform: { position: [1, 2, 3] } },
});
updates: event => ({
    id: event.channel.split("/")[3], // from the channel
    update: { local_transform: { position: [1, 2, 3] } },
});

Several objects per event​

The norm for a machine publishing a whole-state frame — one message, every axis:

entities: { byName: "{id}" },
updates: ({ payload }) => {
    const { angle, height } = payload as { angle: number; height: number };
    return [
        { id: "blade", update: { local_transform: { eulerOrientation: [0, angle, 0] } } },
        { id: "carriage", update: { local_transform: { position: [0, height, 0] } } },
    ];
},

Ignoring an event​

Return null (or an empty array). This is how a mapping declines a message it matched but cannot use — a field missing, a value out of range:

updates: event => {
    const value = asNumber((event.payload as { value?: unknown }).value);
    if (value === null) {
        return null;
    }
    return { id: "gauge", update: { local_transform: { position: [value, 0, 0] } } };
},

Dropping deliberately is better than writing a NaN into a component.

Updating components​

The update field is a set of component patches, keyed by component name — the same shape entity.updateComponent takes, merged into whatever the component already holds:

update: {
    local_transform: { position: [1, 2, 3] },
    material_ref: { value: MATERIAL_UUID },
}

Any component works — the ones a data stream usually drives being:

Transform
Position, orientation and scale — motion of every kind
Material Reference
Recolor an entity to show a state: running, idle, in fault
Mesh Reference
Swap the geometry itself, for a part that changes shape

See the entity components reference for the full list.

Writes are deduplicated: the applier remembers the last patch per entity and component, and skips a write whose value is unchanged. A stream that repeats the same value at 40 Hz costs nothing after the first message.

Adding and removing components​

A component the entity does not have yet is created. An update naming a missing component adds it, initialized from its defaults and your patch — there is nothing to declare up front. An existing component is merged one level deep: the top-level keys you provide are replaced wholesale, and nested objects are not deep-merged.

A mapping cannot remove a component. A patch left undefined is skipped rather than treated as a removal, and the only removal a mapping can express is the whole-entity "delete" directive below. If you genuinely need entity.deleteComponent(), reach for it through the agent directly.

Whole-entity directives​

When an event does not change an entity but changes whether it is there, write a directive instead of components:

DirectiveEffect
"delete"removes the entity from the scene
"hide"makes it invisible, keeping it in the scene
"show"makes it visible again
// One boolean, two lamps.
updates: event => {
    const { value } = event.payload as { value: unknown };
    if (typeof value !== "boolean") {
        return null;
    }
    return [
        { id: "Green", update: value ? "show" : "hide" },
        { id: "Red", update: value ? "hide" : "show" },
    ];
},

A deleted entity is forgotten, so a later event carrying the same id resolves — or respawns — it. That is what makes "delete" the right way to model a vehicle leaving the fleet: it comes back on its own when it returns.

Broadcast is handled for you​

An entity driven by a data stream should not re-broadcast those writes to other clients: the agent's own update loop already sends them, once, at a controlled rate. Doing it twice costs bandwidth and makes motion stutter.

The pipeline takes care of this. On the first successful resolution it sets auto_broadcast = false on every entity it resolves or spawns — which suppresses the additional broadcast pass, not the live update every client sees. There is nothing to configure, and nothing to remember, as long as the ingestion layer is the thing touching the entity.

Pass manage_auto_broadcast: false to the pipeline to take this over yourself.

Validating the payload​

schema is a JSON Schema for the event type. The first matching event is validated against it — a cheap sanity check that the stream carries what the mapping reads:

schema: {
    type: "object",
    properties: { rpm: { type: "number" }, temp_c: { type: "number" } },
    required: ["rpm", "temp_c"],
},

Without it, a renamed field is silently ignored and the scene simply never moves. With it, you get an error the first time a message arrives.

Validation is optional, and so is its validator: run npm install ajv if any of your mappings declares a schema. It is imported lazily, only when a schema is actually used.

How resolution is cached​

An id is resolved to an entity once per scene, then served from cache — including failures, so an id that maps to nothing is looked up once rather than on every event.

The cache mostly invalidates itself from the scene's own events: an id that resolved to nothing is retried when an entity carrying the name or UUID it was looking for appears, and resolutions pointing at entities another client deleted are dropped. You should not have to think about it.

One consequence is worth knowing during bring-up: an id that resolves to nothing is counted under the unresolved_entity drop reason, and that is normal at startup in a scene where a byName mapping runs alongside a spawn mapping — retained status messages, for instance, arrive before any telemetry has spawned the entities they refer to. They resolve on their own as soon as the entities appear.

Testing a mapping​

IngestionPipeline has no dependency on an agent, a session or a broker. Bind a scene and push an event in:

const pipeline = new IngestionPipeline({ mappings: telemetryMapping });
pipeline.bind({ scene });

await pipeline.ingest({ channel: "devices/dev-01/telemetry", payload: { pos: [1, 2, 3] } });

That is the whole data layer. It makes a mapping straightforward to unit-test, to drive from a webhook or a REST handler, and to step through one frame at a time from a debug control.

Next steps​