Synapse — Sensor Ingest

Synapse is the serverless sensor/IoT ingest gateway for tissue.systems. Devices publish telemetry over MQTT; synapse authenticates each device, enforces per-account and per-device topic isolation and rate limits, and dispatches every reading into the owning Cell's sensor handler.

The name fits the tissue metaphor: a synapse is the junction that transmits a signal from one cell to the next. Synapse is that junction for physical devices — it carries each reading from the sensor into your Cell, then gets out of the way. It is stateless: synapse stores no readings. Your Cell owns storage (typically c3).

How it works

device ──MQTT(S):8883──▶ synapse (auth + ACL + rate-limit) ──HTTP POST──▶ Cell.sensor() ──▶ c3

1. A device opens an MQTT-over-TLS connection to ingest.tissue.dev:8883 and authenticates with its device_id (username) and token (password).
2. It publishes a message to tissue/<account_id>/<device_id>/<metric>.
3. Synapse checks the device's rate limit, verifies the topic is inside the device's own namespace, and looks up the owning Cell.
4. Synapse POSTs the raw payload to the Cell runtime with x-tissue-event: sensor and identity headers. The router delivers it to your Cell's sensor(event, env) export.
5. Your Cell decodes the payload and does whatever it likes with it — usually an INSERT into c3.

Dispatch is best-effort, exactly like pulse: if the Cell is unavailable or throws, the reading is not retried. Telemetry is loss-tolerant by design.

Transport

Synapse speaks MQTT 3.1.1 over TLS (MQTTS) on port 8883. The production ingest host is ingest.tissue.dev, geo-routed to the nearest edge.

Devices authenticate at MQTT CONNECT with:

Not yet supported

The following are designed but not implemented today:

• CoAP / DTLS ingest transport.
• MQTT over WebSocket.
• mTLS device authentication (client certificates). Auth is token-based today.

Downlink (cloud → device)

Data flows the other way too: a device may subscribe to topics inside its own namespace (by convention tissue/<account>/<device>/cmd/#), and your Cell publishes commands to it through the downlink endpoint — POST https://ingest.tissue.dev/publish with the device's own credentials and a topic relative to its namespace. Downlink cmd/* messages are routed to the device's live subscription and are never dispatched back into your Cell's sensor() handler, so a command can't masquerade as a reading. Full walkthrough with device-side code (LED, servo): Connect Your Own Device → Cloud to device.

Topics

Every reading is published on a canonical topic:

tissue/<account_id>/<device_id>/<metric>

• Account isolation is broker-enforced: a device can only publish under its own account prefix (tissue/<account_id>/).
• Per-device ACL is enforced per message: a device may only publish within its own device prefix (tissue/<account_id>/<device_id>/); it cannot publish as another device, even inside the same account.
• One message per metric. The <metric> segment is everything after the device id, so tissue/acct_9c/dev_a1b2c3/env/humidity carries the metric env/humidity.

Inside the Cell, derive the metric from the topic:

const metric = event.topic.split("/").slice(3).join("/");

Payload formats

Synapse forwards the raw message bytes untouched — it does not parse the payload. Your Cell decides how to decode it. Common choices:

The reference Cell (tissue-sense) tries JSON first and falls back to a bare number.

The sensor handler

Export a sensor function from your Cell module alongside fetch. Synapse calls it once per reading:

export default {
  async sensor(event, env) {
    const metric = event.topic.split("/").slice(3).join("/");
    const body = event.json();            // decode the payload
    console.log(`${event.device} ${metric} = ${body.value}${body.unit ?? ""}`);
  },

  async fetch(request, env) {
    return new Response("ok");
  },
};

The sensor export is optional. A Cell without it silently ignores any sensor events dispatched to it (synapse gets a 204 and moves on).

The env argument is identical to the one passed to fetch — all declared bindings (c3, g7, text) are fully available.

The event object

Example: store readings in c3

A complete Cell that writes each reading into a c3 table and prunes old rows hourly with pulse.

ribo.toml

[cell]
name = "tissue-sense"
js   = "./cell.js"

[[pulse]]
schedule = "0 * * * *"   # hourly retention prune

[[bindings]]
type     = "c3"
binding  = "DB"
database = "tissue-sense"

cell.js

const SCHEMA = `
  CREATE TABLE IF NOT EXISTS readings (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    account_id  TEXT NOT NULL,
    device_id   TEXT NOT NULL,
    metric      TEXT NOT NULL,
    value       REAL NOT NULL,
    unit        TEXT,
    recorded_at TEXT NOT NULL
  )
`;

export default {
  // Fired by synapse, once per reading.
  async sensor(event, env) {
    await env.DB.exec(SCHEMA);

    // topic = tissue/<account>/<device>/<metric...>
    const metric = event.topic.split("/").slice(3).join("/") || "unknown";

    // Decode JSON {value, unit}, falling back to a bare number.
    let value, unit;
    try {
      const body = event.json();
      value = Number(body.value);
      unit = body.unit ?? null;
    } catch {
      value = Number(event.text());
      unit = null;
    }
    if (Number.isNaN(value)) {
      console.log(`dropped non-numeric reading device=${event.device} metric=${metric}`);
      return;
    }

    await env.DB.prepare(
      "INSERT INTO readings (account_id, device_id, metric, value, unit, recorded_at) VALUES (?, ?, ?, ?, ?, ?)"
    ).bind(event.account, event.device, metric, value, unit, event.receivedTime.toISOString()).run();
  },

  // Hourly retention prune (> 14 days).
  async pulse(event, env) {
    await env.DB.exec(SCHEMA);
    const cutoff = new Date(Date.now() - 14 * 24 * 60 * 60 * 1000).toISOString();
    await env.DB.prepare("DELETE FROM readings WHERE recorded_at < ?").bind(cutoff).run();
  },

  // Render the latest readings as JSON.
  async fetch(request, env) {
    await env.DB.exec(SCHEMA);
    const { results } = await env.DB.prepare(
      "SELECT device_id, metric, value, unit, recorded_at FROM readings ORDER BY id DESC LIMIT 200"
    ).all();
    return Response.json(results);
  },
};

Getting started

1. Register a device under your Cell — with the CLI (ribo sensor add <cell>) or the dashboard's Synapse section. Save the token it prints; it is shown only once.
2. Configure the device's MQTT client: host ingest.tissue.dev, port 8883 (TLS), username = the device_id, password = the token, publish topic tissue/<account>/<device>/<metric>.
3. Deploy a Cell that exports sensor(event, env) with ribo deploy (see the example above).
4. Watch readings arrive — the device publishes, synapse dispatches, and your sensor handler runs. Query your c3 table (or the Cell's fetch endpoint) to confirm.

See also

• Sensor API Reference — the /v1/sensors/* device-registry endpoints
• Managing Devices — the device credential model and lifecycle
• ribo CLI Reference — ribo sensor commands
• c3 Overview — structured storage for readings