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

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";

updates: event => {
    const { rpm } = event.payload as { rpm: number };

    return {
        id,
        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

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.

Driving many kinds from one channel

When one channel carries genuinely different kinds of thing, keep the behaviour in a table rather than in a chain of ifs. Each entry returns an update that keeps going, so the same table serves a spinning shaft and a sliding carriage:

// One row per kind of thing that moves. Adding a kind is adding a row.
const MOTION_BY_KIND: Record<string, (part: Part) => AnyContinuousUpdate> = {
    spindle: part =>
        continuous<{ angle_deg: number }>(
            ({ delta_seconds, state }) => {
                state.angle_deg = (state.angle_deg + part.rpm * 6 * delta_seconds) % 360;
                return { local_transform: { eulerOrientation: [state.angle_deg, 0, 0] } };
            },
            { initial_state: { angle_deg: 0 } },
        ),

    carriage: part =>
        continuous<{ position_m: number }>(
            ({ delta_seconds, state }) => {
                state.position_m += part.speed_ms * delta_seconds;
                return { local_transform: { position: [0, state.position_m, 0] } };
            },
            { initial_state: { position_m: 0 } },
        ),
};

updates: event =>
    partsOf(event).map(part => ({
        id: part.id,
        update: MOTION_BY_KIND[part.kind](part),
    })),

updates stays this size whether the table has three rows or three hundred, and supporting a new kind of part means adding a row rather than editing the code that dispatches.

Note that state is per entity, so each part accumulates its own angle or position and none of them needs to carry a "starting from" value in its payload — every one of a hundred spindles gets its own { angle_deg: 0 } on first sight, and keeps it across every event that changes its speed.

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​