Skip to main content
26 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

Choosing where the id comes from​

Nothing in the SDK decides this. updates returns whatever id you give it and entities resolves that; which shape fits depends on how your source structures its channels and its payloads.

The stream says which entity byTake the id fromEntries per message
the topic — one device per channelevent.channel.split("/")[2]one
a field in the payloadpayload.serialone
the payload's own keysthe key, or a name you map it toone per key
the topic and the keysboth, combinedone per key

The first two are the common case — one message, one entity, the id read straight off the event:

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

When the identity is in the key, you also choose whether the id is that key or a name you map it to. id: tag keeps one table and makes a warning name the exact controller address (No entity mapping for id "Q0.3"); mapping it to LAMP_0_3 costs an indirection and puts the scene's vocabulary in your logs and your per_mapping statistics instead. Prefer the second when the two vocabularies genuinely differ.

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] } } },
    ];
},

One entry per entity, always. The length of the array is the number of entities the message touches. Two motors reporting in one frame is two entries — they cannot be collapsed, because an entry is what carries an id, and an id is what finds an entity.

That matters as soon as the frame stops being hand-written. A PLC publishes its whole data block at once, and which entity each value belongs to is written in the key:

{ "DB1.DBW0_SPEED_RPM": 90, "Q0.1": true, "Q0.2": false, "Q0.3": true }

Walk the payload and emit one entry per key you recognise: a motion for the rate, a plain directive for each boolean. One array holds both.

// An rpm is a rate, so the shaft has to keep turning between messages.
function spin(rpm: number): AnyContinuousUpdate {
    return continuous<{ angle_deg: number }>(
        ({ delta_seconds, state }) => {
            state.angle_deg = (state.angle_deg + rpm * 6 * delta_seconds) % 360;
            return { local_transform: { eulerOrientation: [state.angle_deg, 0, 0] } };
        },
        { initial_state: { angle_deg: 0 } },
    );
}

updates: event => {
    const payload = event.payload as Record<string, unknown>;
    const entityUpdates: Array<EntityUpdate> = [];

    // A frame carries more than this mapping cares about, so tags fall through untouched.
    for (const [tag, value] of Object.entries(payload)) {
        if (tag === "DB1.DBW0_SPEED_RPM" && typeof value === "number") {
            entityUpdates.push({ id: tag, update: spin(value) });
        } else if (/^Q\d+\.\d+$/.test(tag)) {
            entityUpdates.push({ id: tag, update: value ? "show" : "hide" });
        }
    }

    return entityUpdates;
},

A partial frame — the normal case for a controller reporting only what changed — needs nothing extra: Object.entries yields only the keys that are present, so the entities behind the absent ones are left untouched, and a motion already running for an absent key keeps running. Silence about an entity is not an instruction to stop it, the same rule as returning null from updates.

Factoring the update, not the entries​

The entry count is fixed by how many entities the message touches, but the code need not grow with it. Ten lamps share one builder and three motors share another; keep them in a table with one row per tag, and adding a tag becomes adding a row:

type TagBinding = {
    entity_uuid: UUID;
    build: (value: unknown) => EntityUpdate["update"] | null;
};

// One builder per kind of value, reusing `spin` from above. Returning null drops a tag
// whose value is not the type it should carry.
function rpm(value: unknown): EntityUpdate["update"] | null {
    if (typeof value !== "number") {
        return null;
    }
    return spin(value);
}

function onOff(value: unknown): EntityUpdate["update"] | null {
    if (typeof value !== "boolean") {
        return null;
    }
    return value ? "show" : "hide";
}

// A Map rather than an object literal: these keys come straight from the payload, and a tag
// named `constructor` or `toString` would find a member of `Object.prototype` in a literal.
const TAGS = new Map<string, TagBinding>([
    ["DB1.DBW0_SPEED_RPM", { entity_uuid: MOTOR_0, build: rpm }],
    ["DB1.DBW2_SPEED_RPM", { entity_uuid: MOTOR_1, build: rpm }],
    ["Q0.1", { entity_uuid: LAMP_0_1, build: onOff }],
    ["Q0.2", { entity_uuid: LAMP_0_2, build: onOff }],
]);

// Built from the same table, so the two cannot drift apart.
const ENTITY_BY_TAG: Record<string, UUID> = {};
for (const [tag, binding] of TAGS) {
    ENTITY_BY_TAG[tag] = binding.entity_uuid;
}

const mapping: EventMapping = {
    channel: "sixd/mockplc/plc01/telemetry",
    entities: { byUuid: ENTITY_BY_TAG },

    updates: event => {
        const payload = event.payload as Record<string, unknown>;
        const entityUpdates: Array<EntityUpdate> = [];

        for (const [tag, value] of Object.entries(payload)) {
            const binding = TAGS.get(tag);
            if (binding === undefined) {
                continue;
            }

            const update = binding.build(value);
            if (update !== null) {
                entityUpdates.push({ id: tag, update });
            }
        }

        return entityUpdates;
    },
};

A builder returning null for a value of the wrong type drops that tag rather than writing a NaN into a component — ignoring an event, one tag at a time. updates then stays this size whether the table has four rows or four hundred, and supporting a new tag means adding a row rather than editing the code that dispatches.

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.

Updates that keep going​

Most events carry a value — a position, a temperature — and writing it is the end of the story. Some carry a rate: "the shaft is turning at 90 rpm". A rate still means something after the message that delivered it, so writing a finished patch would leave the shaft frozen until the next message — and a machine that reports only when a value changes may not send one for minutes.

Wrap the update in continuous() and it keeps producing values on its own, until a later event for the same id replaces it:

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

entities: { byName: "shaft-{id}" },
updates: event => {
    const { rpm } = event.payload as { rpm: number };

    return {
        // "plant/plc01/telemetry" → the entity named shaft-plc01
        id: event.channel.split("/")[1],
        update: continuous<{ angle_deg: number }>(
            ({ delta_seconds, state }) => {
                // 360 degrees a turn, 60 seconds a minute.
                state.angle_deg = (state.angle_deg + rpm * 6 * delta_seconds) % 360;
                return { local_transform: { eulerOrientation: [state.angle_deg, 0, 0] } };
            },
            { initial_state: { angle_deg: 0 } },
        ),
    };
},

The sample function receives three things:

delta_secondsTime since the previous sample, and 0 on the installing event. Use it for a value that accumulates, as above.
since_secondsTime since the event that installed this motion. Use it for a value that is a closed-form function of age — a fade, a ramp, a countdown.
stateScratch space belonging to the entity.

state is the part worth understanding. It is not the motion's, it is the entity's: it survives the event that replaces the continuation, so a new rpm picks the shaft up at the angle it had reached instead of snapping it back to zero. initial_state is therefore applied only the first time an entity is given a state — a second event cannot reset a motion already under way. Without it you would keep the phase in a Map of your own, read at event time and written back on every tick, which is three lines and one misplaced read away from a shaft that accelerates forever.

How a motion ends​

A continuation is installed per mapping and id — not per channel, so one mapping matching plant/+/telemetry holds plc01 and plc02 independently; and not per session, so two viewers on the same scene see one shaft turning at one speed.

There are five ways it stops:

  • the sample returns null — the motion is over, and the entity keeps its last value;
  • a later event for that id installs a new one (this is the common case: a new rate takes over);
  • the sample, or an event, returns "hide" or "delete" — nothing anyone can see is moving, so the writes stop too. "delete" drops the entity's state with it; "hide" keeps it, so a hidden shaft resumes where it stopped;
  • the sample throws — it would throw again on every tick, so it is reported once and dropped;
  • you call pipeline.clearContinuations({ id }), or SceneIngestion.stop() does it for you.

What happens when the stream dies​

Nothing expires a motion on a timer, on purpose. The agent can measure how long it has been since an event replaced a motion, but that is not the same fact as "the stream is dead" — as the mixed payloads above show, a perfectly healthy topic can go a long time without refreshing any one motion. A timer built on it would stop live machines and stay quiet about dead ones.

So the honest signal is two counters read together, and it costs nothing: last_event_at says whether data is still arriving, and continuations_active says whether the scene is still moving. The two disagreeing is exactly the picture outliving the data. When you decide that has happened — a broker known bad, a gateway wedged — pipeline.clearContinuations() stops the motions; the entities keep their last value, and a later event starts them again.

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.

How many mappings?​

Mappings scale with the number of event types, not with the number of entities:

You haveWrite
Many entities, one behaviour (100 AGVs)One mapping — entities does the fan-out
Several behaviours, each on its own channelOne mapping per event type
One message carrying several partsOne mapping returning an array, one entry per part
Many kinds of thing on a single channelOne mapping, and a table keyed by kind
One channel, families needing different entitiesOne mapping per family, all on that channel

So there is never a mapping per entity: a hundred AGVs sharing one telemetry topic are one mapping, because resolving an id to its entity is exactly what entities is for.

It is worth noting that "hundreds of things, each moving differently" is almost always hundreds of instances of a handful of kinds. Each entity gets its own function, closed over its own values, so entities differ from one another without the mapping count growing.

When one channel carries genuinely different kinds of thing, keep the behaviour in a table keyed by tag rather than in a chain of ifs. That works as long as one entities strategy resolves every one of them, which is the usual case.

The last row of the table is where that stops being true. entities is declared per mapping, and exactly one strategy is allowed — so a frame whose motors are a fixed UUID table and whose lamps are found by name cannot be one mapping. It is two, on the same channel:

// Both match the same topic. Every mapping whose selectors match an event runs — there is no
// first-match-wins, so each one takes the tags it owns and ignores the rest. `spin` is the
// builder from "several objects per event" above.
const MOTOR_MAPPING: EventMapping = {
    channel: "sixd/mockplc/+/telemetry",
    entities: { byUuid: MOTOR_UUID_BY_TAG },

    updates: event => {
        const payload = event.payload as Record<string, unknown>;
        const entityUpdates: Array<EntityUpdate> = [];

        for (const [tag, value] of Object.entries(payload)) {
            if (tag.endsWith("_SPEED_RPM") && typeof value === "number") {
                entityUpdates.push({ id: tag, update: spin(value) });
            }
        }

        return entityUpdates;
    },
};

const LAMP_MAPPING: EventMapping = {
    channel: "sixd/mockplc/+/telemetry",
    entities: { byName: "lamp-{id}" },

    updates: event => {
        const payload = event.payload as Record<string, unknown>;
        const entityUpdates: Array<EntityUpdate> = [];

        for (const [tag, value] of Object.entries(payload)) {
            if (/^Q\d+\.\d+$/.test(tag)) {
                entityUpdates.push({ id: tag, update: value ? "show" : "hide" });
            }
        }

        return entityUpdates;
    },
};

new IngestionPipeline({ mappings: [MOTOR_MAPPING, LAMP_MAPPING] });

Splitting costs one extra walk of the payload and buys real separation: each mapping keeps its own resolution cache, its own deduplication memory, and its own row in per_mapping — so a bring-up problem with the lamps shows up as drops against the lamp mapping rather than against everything at once.

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​