Build a real-time chat app end-to-end

A complete walkthrough of the pattern most people think needs a dedicated realtime backend: a channelled chat app with live SSE message streaming, a typing indicator on a broadcast topic, and an AI moderation workflow that flags abusive messages. One prompt, real endpoints, no websocket infra.

0. Prep

• Create a project (Getting started) using the blank template.
• Attach an AI provider under BYOK or use the platform pool — the moderation step calls ctx.integrations.invoke against it. BYOK.

1. The initial prompt

Build a real-time chat app.

Data:
- channels(id uuid pk, slug text unique, name text,
           created_by uuid, created_at timestamptz default now())
- channel_members(id uuid pk, channel_id uuid fk, user_id uuid,
                  unique(channel_id, user_id))
- messages(id uuid pk, channel_id uuid fk, author_id uuid,
           body text, flagged boolean default false, flag_reason text,
           created_at timestamptz default now())
- RLS: users can only read channels/messages for channels they are a member of.

UI:
- /login, /signup.
- /channels — list channels I'm a member of + "New channel" form.
- /c/[slug] — message list (oldest → newest), input box, typing indicator
  ("Anna is typing…").
- Live: subscribe to topic "chat.<channelId>.message" for new messages,
  and "chat.<channelId>.typing" for keystrokes.
- Send: runtime.api.create("messages", ...).

Use runtime.api and named queries. No raw SQL in components.

2. Data model & RLS

Open .appbricx/backend/migrations/001_chat.sql:

CREATE TABLE IF NOT EXISTS channels (
  id         uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  slug       text UNIQUE NOT NULL,
  name       text NOT NULL,
  created_by uuid NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS channel_members (
  id         uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  channel_id uuid NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
  user_id    uuid NOT NULL,
  UNIQUE (channel_id, user_id)
);

CREATE TABLE IF NOT EXISTS messages (
  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  channel_id  uuid NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
  author_id   uuid NOT NULL,
  body        text NOT NULL,
  flagged     boolean NOT NULL DEFAULT false,
  flag_reason text,
  created_at  timestamptz NOT NULL DEFAULT now()
);

ALTER TABLE channels        ENABLE ROW LEVEL SECURITY;
ALTER TABLE channel_members ENABLE ROW LEVEL SECURITY;
ALTER TABLE messages        ENABLE ROW LEVEL SECURITY;

CREATE POLICY ch_by_membership ON channels
  USING (EXISTS (
    SELECT 1 FROM channel_members m
    WHERE m.channel_id = channels.id
      AND m.user_id::text = current_setting('app.user_id', true)
  ));

CREATE POLICY chm_by_membership ON channel_members
  USING (EXISTS (
    SELECT 1 FROM channel_members m2
    WHERE m2.channel_id = channel_members.channel_id
      AND m2.user_id::text = current_setting('app.user_id', true)
  ));

CREATE POLICY msg_by_membership ON messages
  USING (EXISTS (
    SELECT 1 FROM channel_members m
    WHERE m.channel_id = messages.channel_id
      AND m.user_id::text = current_setting('app.user_id', true)
  ))
  WITH CHECK (author_id::text = current_setting('app.user_id', true));

Background: Auth & RLS. TheWITH CHECK on messages means people can only insert messages authored as themselves, into channels they belong to.

3. ACL & auto CRUD

// .appbricx/backend/api/tables.json
{
  "allow": ["channels", "channel_members", "messages"],
  "tables": {
    "channels":        { "expose": true, "methods": ["GET", "POST"] },
    "channel_members": { "expose": true, "methods": ["GET", "POST", "DELETE"] },
    "messages":        { "expose": true, "methods": ["GET", "POST"] }
  }
}

Messages are insert-only from the client. Full reference: Auto CRUD REST API.

4. Send a message

Send is a plain runtime.api.create. The mutation emits CDC, which drives both the live stream (step 5) and the moderation workflow (step 7).

async function send() {
  if (!body.trim()) return;
  const user = await db.auth.getUser();
  await runtime.api.create("messages", {
    channel_id: channel.id,
    author_id:  user!.id,
    body,
  });
  setBody("");
}

5. Live stream via CDC → topic

Bind the messagestable's insert op to a topic named by channel id. Write .appbricx/backend/cdc/bindings.json:

{
  "bindings": [
    {
      "id": "messages-insert-to-topic",
      "table": "messages",
      "ops": ["insert"],
      "topic": "chat.messages.inserted",
      "workflow": "fanout-message"
    }
  ]
}

A tiny fan-out workflow republishes the row on a channel-specific topic so subscribers can filter cheaply. .appbricx/backend/workflows/fanout-message.workflow.js:

/** @param {import("@appbricx/runtime").WorkflowContext} ctx */
export async function run(ctx) {
  const row = ctx.trigger.payload?.row;
  if (!row) return { ok: true, skipped: true };
  await ctx.topics.publish(`chat.${row.channel_id}.message`, row);
  return { ok: true };
}

On the client, subscribe by channel id — the SSE stream stays pinned to the project worker while a subscriber is connected.

useEffect(() => {
  runtime.queries.run("list_messages", { channel_id: channel.id })
    .then((r) => r.ok && setRows(r.rows ?? []));

  const unsub = runtime.topics.subscribe(
    `chat.${channel.id}.message`,
    (ev) => setRows((prev) => [...prev, ev]),
  );
  return unsub;
}, [channel.id]);

SSE mechanics + how CDC gets wired: Topics & CDC.

6. Typing indicator — pure broadcast topic

Typing does not need to persist. Publish a topic straight from the client and forget it. Debounce so you do not spam the bus.

// client — throttled to once per 2s
const notifyTyping = useMemo(
  () => throttle(async () => {
    const user = await db.auth.getUser();
    await runtime.topics.publish(
      `chat.${channel.id}.typing`,
      { user_id: user!.id, name: user!.name, at: Date.now() },
    );
  }, 2000),
  [channel.id],
);

// subscribe elsewhere on the screen
useEffect(() => {
  const unsub = runtime.topics.subscribe(
    `chat.${channel.id}.typing`,
    (ev) => setTyping((prev) => upsertTyping(prev, ev)),
  );
  return unsub;
}, [channel.id]);

Or via HTTP if you want to test from curl:

curl -sS -X POST "$ORIGIN/__appbricx/topics/chat.$CHANNEL_ID.typing/publish" \
  -H "Authorization: Bearer $DATA_TOKEN" \
  -H "x-appbricx-data-api: 1" \
  -H "content-type: application/json" \
  -d '{"payload":{"user_id":"'$USER_ID'","name":"Anna","at":'$(date +%s000)'}}'

7. Moderation workflow — flag abusive messages

The same CDC binding that fans out the message also triggeredfanout-message. Extend the bindings so a moderation workflow runs in parallel:

{
  "bindings": [
    { "id": "messages-fanout", "table": "messages", "ops": ["insert"],
      "topic": null, "workflow": "fanout-message" },
    { "id": "messages-moderate", "table": "messages", "ops": ["insert"],
      "topic": null, "workflow": "moderate-message" }
  ]
}

.appbricx/backend/workflows/moderate-message.workflow.js calls the connected AI integration and updates the row via auto CRUD:

/** @param {import("@appbricx/runtime").WorkflowContext} ctx */
export async function run(ctx) {
  const row = ctx.trigger.payload?.row;
  if (!row?.body) return { ok: true, skipped: true };

  const verdict = await ctx.integrations.invoke("ai", "classify", {
    input: row.body,
    labels: ["safe", "harassment", "hate", "sexual"],
    instructions: "Return the single most fitting label plus a 1-sentence reason.",
  });

  if (verdict.label !== "safe") {
    await ctx.api.update("messages", row.id, {
      flagged:     true,
      flag_reason: `${verdict.label}: ${verdict.reason ?? ""}`,
    });
    await ctx.topics.publish(`chat.${row.channel_id}.flagged`, {
      id: row.id, label: verdict.label,
    });
    ctx.log.warn(`flagged message ${row.id} as ${verdict.label}`);
  }

  return { ok: true, label: verdict.label };
}

On the client, hide flagged rows or replace their body with a placeholder — the update fires a second CDC event, so a small extra subscription can react without re-fetching.

8. Manually invoke a workflow for testing

# via HTTP
curl -sS -X POST "$ORIGIN/__appbricx/runtime/workflows/moderate-message/run" \
  -H "Authorization: Bearer $DATA_TOKEN" \
  -H "x-appbricx-data-api: 1" \
  -H "content-type: application/json" \
  -d '{"payload":{"row":{"id":"'$MSG_ID'","channel_id":"'$CH_ID'","body":"go away"}}}'

# inspect the run
curl -sS "$ORIGIN/__appbricx/runtime/runs/$RUN_ID" \
  -H "Authorization: Bearer $DATA_TOKEN" \
  -H "x-appbricx-data-api: 1"

Full surface: Workflows.

9. Publish

1. Editor → Deploy / Publish.
2. Two browsers, two accounts, one channel — the second browser sees new messages within one SSE frame, sees the typing indicator, and never sees flagged rows.

Custom domain, TLS, publish flow: Publish & domains.

Internal map (what just happened)

Browser (input box)
  → /__appbricx/api/v1/messages           (RLS: author = app.user_id)
     └─ CDC emit (insert)
        ├─ fanout-message      → /__appbricx/topics/chat.<ch>.message/publish
        └─ moderate-message    → AI integration → api.update(flagged=true)

Browser (message list)
  ← /__appbricx/topics/chat.<ch>.message/subscribe          (SSE)
  ← /__appbricx/topics/chat.<ch>.typing/subscribe           (SSE, ephemeral)
  ← /__appbricx/topics/chat.<ch>.flagged/subscribe          (SSE, replace body)