Live Data

warning

Experimental. This feature may change or be removed.

A shell can receive writes on devices and entities while the page is open. Ask sql.run what a query read with watch: true, pass that list to stream.open, and the host subscribes to the platform's realtime feed with the viewer's own session. The viewer's token never enters the iframe. Everything the feed delivers for the stream reaches your page as a dashboard:event message. The snippets on this page use the request helper from the bridge example.

Inserts carry the records that were written, so a page can render them without another query. Updates and deletes carry what the write sent, which is not enough to patch a query result, so re-run the query on those. The example below re-runs the query on every data message, debounced to one run every 5 seconds, so its table stays current, and lists insert records as they arrive. Each re-run executes the query and counts toward the per-minute execute limit (Executing queries), so pick the debounce for the rate your devices write at.

sql.run with watch

Add watch: true to a sql.run payload. The result gains a watch key:

const result = await request("sql.run", { query_id: "...", watch: true });
// result.watch: {
//   devices: [{ id: "...", variables: ["temperature", "humidity"] }],
//   entities: [{ id: "..." }]
// }

Field	Meaning
devices	Devices the query read. variables lists the names the query filtered on; an entry without it read every variable.
entities	Entities the query read.

Without watch, the result has no watch key. The list describes this run: a query that reads devices by tag names the devices that matched now, and a device that joins the tag later is not on it. Run the query with watch again when you want a fresh list, then close the old stream and open a new one.

stream.open

Opens one stream for a list of devices and entities and returns a handle.

const { stream_id } = await request("stream.open", {
  devices: [{ id: "...", variables: ["temperature"] }],
  entities: [{ id: "...", fields: ["status"] }],
});

Field	Required	Meaning
devices	No	Array of { id, variables? }. Without variables, or with an empty list, every variable streams.
entities	No	Array of { id, fields? }. Without fields, or with an empty list, every field streams.

A watch list from sql.run is a valid payload when it names at least one device or entity. A list with neither, as a query that read only list tables returns, fails with bad_request. The host keeps the connection open until you close the stream, the page unloads, or nothing is left to stream (status: "closed" below). Reconnects after a network drop are the host's job.

Variable and field names narrow what a channel delivers; they are not a permission. When the viewer may stream at least one resource on the list, each one they may not stream is reported with a revoked event, not with an error on stream.open. When they may stream none of them, the server refuses the connection: no revoked arrives, the host retries with backoff, and the stream reports status: "failed" after one to two minutes.

Each embed may hold up to 5 open streams. A sixth stream.open fails with bad_request. Close a stream you no longer need before opening another. Other errors follow the bridge error table.

stream.close

Closes a stream and frees its slot. The result is {}, also for a handle that is unknown or already closed.

await request("stream.close", { stream_id });

The host closes every open stream when the dashboard unmounts or the iframe reloads, so a page that lives until navigation does not have to close its streams itself.

dashboard:event

{
  "type": "dashboard:event",
  "stream_id": "...",
  "event": "data",
  "data": {
    "channel": "device_data.5f1b2c3d4e5f6a7b8c9d0e1f",
    "payload": { "operation": "insert", "variables": [] }
  }
}

event	data	When
data	{ channel, payload }, the feed message unchanged	A write on a subscribed device or entity. See Data messages.
status	{ status }	The connection changed state: connecting, open, reconnecting, failed, or closed. See below.
revoked	{ channel }	The viewer may not stream that resource. channel is device_data.<id> or entity_data.<id>. Other resources on the stream continue; when none is left, status: "closed" follows.
resync	{ reason }	The connection may have missed events. reason is overflow or reconnect.
closed	{ reason }	The viewer's session stopped verifying. reason is unauthorized. status: "closed" follows.

Match events to streams by stream_id. Ignore events whose stream_id you did not open, events whose event you do not know, and reason values you do not know. The feed adds events and fields over time and never renames or removes one a page reads.

Events for a stream reach the page only after the stream.open response, so a handler set when the response arrives misses nothing, including the first status: "open".

Re-run your queries on status: "open" and on resync. A stream opens after the query it came from ran, and a reconnect delivers nothing that was published while the connection was down, so the query is the only way to catch up. open arrives on the first connect and after every reconnect.

status: "failed" means the host's retries ran out. The stream keeps its slot, and the host starts a fresh round of attempts when the tab becomes visible after being hidden: reconnecting, then open when one succeeds, which re-runs your queries as any open does. Close the stream yourself when you would rather not wait.

status: "closed" means nothing is left to stream: the server sent closed because the viewer's session stopped verifying, for example a deactivated Run user, or every resource on the stream was revoked. The host has already dropped the stream and freed its slot; show that the page is no longer live. Your own stream.close produces no status event.

Data messages

data.channel names the resource and data.payload is what the realtime feed publishes:

Channel	operation	Payload
device_data.<id>	insert	variables: the written records as stored: id, variable, value, time, group, device, created_at, and unit, location, and metadata when the write sent them. Only the subscribed variables are present.
device_data.<id>	update	variables: each record's id and the changed fields, never variable.
device_data.<id>	delete	variables: the ids of the deleted records only.
entity_data.<id>	insert	items: the inserted rows, cut to the subscribed fields plus id.
entity_data.<id>	update	items: what the write sent, each row cut to the subscribed fields plus id.
entity_data.<id>	delete	items: the ids of the deleted rows.

A device update carries each record's id and the changed fields and never variable, and a delete carries ids only, so neither can be matched by variable: treat both as "this device changed" and re-run the query. An entity message always keeps id, so it is never empty. A subscription without names receives every record whole.

A device entry streams filtered only when it has at most 50 variables and each name matches ^[a-z0-9_]+$; an entity entry only when it has at most 50 fields and each name matches ^[a-z_]+$. Otherwise the host subscribes the whole resource and forwards every record it receives. Filter in your page when that matters: check record.variable against your list, or pick the fields you need from each row.

In TagoRUN

In TagoRUN the realtime feed applies the Run user's Access Management rules:

• sql.run with watch lists only the devices the Run user has Dashboard access to and the entities they have Blueprint access to. Rows are not filtered by device permission (see In TagoRUN on the bridge page), but the watch list is.
• A Run user's stream on a device or entity is allowed by Dashboard access (Blueprint access for an entity) on it, or by a widget on a Normal or Blueprint dashboard the Run user has Access to that shows it. The AI or Custom Dashboard being viewed never counts, because its layout is not made of widgets the server checks. A widget grants the whole device or entity: every variable or field streams, also ones no widget shows. Anything else arrives as revoked, or the stream fails when nothing on it passes (see stream.open).
• The two rules differ by that last grant: a device the Run user sees only through a widget on a Normal or Blueprint dashboard is absent from watch, although stream.open would stream it. A stream built from watch therefore carries only resources with a direct permission, and the next run of the query with watch reflects a rule that was removed.
• The rule is checked when the stream opens and again every 60 seconds while it is open. A rule removed later produces revoked for that resource, and status: "closed" when it was the last one; a deactivated Run user produces closed.

Example

Full page that runs the first saved query with watch, opens a stream from the result, re-runs the query on open, resync, and every data message, reopens the stream when a re-run returns a different watch list, and lists the records of inserts. The request helper and table renderer match the bridge example:

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <title>Custom Dashboard live example</title>
    <style>
      :root {
        color-scheme: light dark;
      }
      body {
        font-family: system-ui, sans-serif;
        margin: 1.5rem;
      }
      table {
        border-collapse: collapse;
        width: 100%;
      }
      th,
      td {
        border: 1px solid #ccc;
        padding: 0.4rem 0.6rem;
        text-align: left;
      }
      #status {
        color: #666;
      }
      #feed li {
        font-family: monospace;
      }
    </style>
  </head>
  <body>
    <h1>Custom Dashboard live example</h1>
    <p id="status">Loading...</p>
    <div id="table"></div>
    <h2>Inserts</h2>
    <ul id="feed"></ul>
    <script>
      const statusEl = document.getElementById("status");
      const tableEl = document.getElementById("table");
      const feedEl = document.getElementById("feed");
      const pending = new Map();

      let queryId = null;
      let streamId = null;
      let streamKey = null;
      let refreshTimer = null;

      window.addEventListener("message", (event) => {
        // Accept messages only from the host window. Allowlisting the Admin and TagoRUN origins is stricter.
        if (event.source !== window.parent) return;
        const msg = event.data;
        if (!msg || typeof msg !== "object") return;

        if (msg.type === "dashboard:theme") {
          document.documentElement.dataset.theme = msg.theme;
          return;
        }

        if (msg.type === "dashboard:event") {
          if (msg.stream_id === streamId) handleStreamEvent(msg);
          return;
        }

        if (msg.type !== "dashboard:response" || !pending.has(msg.id)) return;
        const { resolve, reject } = pending.get(msg.id);
        pending.delete(msg.id);
        msg.ok ? resolve(msg.result) : reject(new Error(`${msg.error.code}: ${msg.error.message}`));
      });

      function request(op, payload) {
        const id = crypto.randomUUID();
        return new Promise((resolve, reject) => {
          pending.set(id, { resolve, reject });
          window.parent.postMessage({ type: "dashboard:request", id, op, payload }, "*");
        });
      }

      function handleStreamEvent(msg) {
        switch (msg.event) {
          case "status":
            statusEl.textContent = `Stream: ${msg.data.status}`;
            if (msg.data.status === "open") scheduleRefresh();
            // The host would retry when the tab is shown again; this page gives up instead.
            if (msg.data.status === "failed") endLive();
            // The host has already dropped the stream. No event re-runs the query from here, so
            // the page stays static; calling syncLive(watch) again would open a new stream.
            if (msg.data.status === "closed") {
              streamId = null;
              streamKey = null;
            }
            return;
          case "resync":
            scheduleRefresh();
            return;
          case "data":
            handleData(msg.data);
            return;
          case "revoked":
            statusEl.textContent = `No longer live: ${msg.data.channel}`;
            return;
          case "closed":
            statusEl.textContent = `Stream closed: ${msg.data.reason}`;
            return;
          default:
            // An event this page does not know. Ignore it.
            return;
        }
      }

      function handleData({ channel, payload }) {
        // Every write re-runs the query, debounced, so the table stays current.
        scheduleRefresh();
        // Inserts also carry the records, so the feed can show them without a query.
        if (payload.operation !== "insert") return;
        const records = channel.startsWith("device_data.") ? payload.variables : payload.items;
        for (const record of records ?? []) renderRecord(channel, record);
      }

      function renderRecord(channel, record) {
        const li = document.createElement("li");
        li.textContent = `${channel} ${JSON.stringify(record)}`;
        feedEl.prepend(li);
        while (feedEl.children.length > 20) feedEl.lastChild.remove();
      }

      // Coalesce bursts of events into one query every 5 seconds at most. Each re-run executes
      // the query and counts toward the per-minute execute limit: a device writing every second
      // costs 12 runs a minute at this debounce, not 60.
      function scheduleRefresh() {
        if (refreshTimer) return;
        refreshTimer = setTimeout(async () => {
          refreshTimer = null;
          try {
            const result = await request("sql.run", { query_id: queryId, watch: true });
            renderTable(result);
            await syncLive(result.watch ?? {});
          } catch (err) {
            statusEl.textContent = String(err.message ?? err);
          }
        }, 5000);
      }

      // One string per distinct watch list, so two runs that read the same resources compare equal.
      function watchKey(watch) {
        const devices = (watch.devices ?? []).map(({ id, variables }) => [
          id,
          variables ? [...variables].sort() : null,
        ]);
        const entities = (watch.entities ?? []).map(({ id }) => id);
        return JSON.stringify({ devices: devices.sort(), entities: entities.sort() });
      }

      function renderTable(result) {
        const columns = result.columns ?? [];
        const rows = result.rows ?? [];
        tableEl.replaceChildren();
        if (!columns.length && !rows.length) {
          tableEl.textContent = JSON.stringify(result, null, 2);
          return;
        }
        const table = document.createElement("table");
        const thead = document.createElement("thead");
        const headerRow = document.createElement("tr");
        for (const key of columns) {
          const th = document.createElement("th");
          th.textContent = key == null ? "" : String(key);
          headerRow.appendChild(th);
        }
        thead.appendChild(headerRow);
        table.appendChild(thead);
        const tbody = document.createElement("tbody");
        for (const row of rows) {
          const tr = document.createElement("tr");
          for (const key of columns) {
            const td = document.createElement("td");
            const value = row[key];
            td.textContent = value == null ? "" : String(value);
            tr.appendChild(td);
          }
          tbody.appendChild(tr);
        }
        table.appendChild(tbody);
        tableEl.appendChild(table);
      }

      let syncing = null;
      let syncAgain = false;
      let latestWatch = null;

      // One sync at a time: a call during an open sets a flag, and the newest list runs once
      // more after it settles. A rejected open leaves the page static with the status shown.
      function syncLive(watch) {
        latestWatch = watch;
        if (syncing) {
          syncAgain = true;
          return syncing;
        }
        syncing = (async () => {
          do {
            syncAgain = false;
            await syncOnce(latestWatch);
          } while (syncAgain);
        })().finally(() => {
          syncing = null;
        });
        return syncing;
      }

      // Keep the stream on what the last run read: a query that reads devices by tag names the
      // devices that match now, and a Run user's list shrinks when a rule is removed.
      async function syncOnce(watch) {
        const key = watchKey(watch);
        if (key === streamKey) return;
        endLive();
        // stream.open rejects a list with no device and no entity.
        if (!watch.devices?.length && !watch.entities?.length) {
          statusEl.textContent = "The query read nothing to watch.";
          return;
        }
        const opened = await request("stream.open", watch);
        streamId = opened.stream_id;
        streamKey = key;
      }

      function endLive() {
        const id = streamId;
        streamId = null;
        streamKey = null;
        // The host answers {} for any handle; the catch only guards an unhandled rejection.
        if (id) request("stream.close", { stream_id: id }).catch(() => {});
      }

      async function main() {
        try {
          const { queries } = await request("sql.list");
          if (!queries.length) {
            statusEl.textContent = "No saved queries on this profile.";
            return;
          }
          queryId = queries[0].id;
          const result = await request("sql.run", { query_id: queryId, watch: true });
          renderTable(result);
          await syncLive(result.watch ?? {});
        } catch (err) {
          statusEl.textContent = String(err.message ?? err);
        }
      }

      parent.postMessage({ type: "dashboard:ready" }, "*");
      main();
    </script>
  </body>
</html>

The first open re-runs the query once. That read covers writes between your first run and the moment the connection opened. Later open events follow reconnects, and the same read covers what the connection missed. Every re-run asks for watch again and compares the list to the one the stream was opened with: a changed list closes the stream and opens a new one, and an unchanged list leaves it alone, so the open that follows a reopen does not start a loop.

Related

• SQL and Endpoint Bridge
• Authoring HTML
• Access Management permissions
• Executing queries