FlowGlow Lighting

Architecture in four layers

The platform is built around four cleanly separated layers, and each layer talks to its neighbours over a single, documented protocol:

1. Firmware (ESP32 / C++): brings up Wi-Fi after boot, reads credentials from NVS and connects to the Mosquitto broker over TLS. Publishes state changes on d/<id>/state, OTA events on d/<id>/ota_events, crash summaries on d/<id>/coredump, and accepts commands on d/<id>/cmd.
2. MQTT consumer (Node.js): subscribes to every device topic, validates payloads, deduplicates, and writes the result transactionally into Postgres. Also publishes commands the other way by polling a lamp_commands table.
3. Supabase / Postgres: the single source of truth. RLS policies govern visibility per customer, a BEFORE INSERT trigger on device_events snapshots every transition, and an AFTER UPDATE trigger on device_state broadcasts the change to exactly the right user channel.
4. Clients: a Flutter app for end customers (iOS, Android) and an Angular dashboard for support and operations. Both speak GraphQL for reads and Realtime Broadcast for live updates — no direct MQTT path into the frontend.

The strict separation means each layer ships independently: firmware updates roll out across devices, the consumer is a stateless worker behind Coolify, the dashboard is an SPA, the Flutter app a store build. No “big-deploy” moment on the calendar, no lock-step between frontend and backend.

Approach

Three design choices run through the entire backend and are why the system stays stable with hundreds of devices online at once.

First: nothing blocks. The consumer is purely async, every message its own transaction. If Postgres briefly lags, messages queue at the broker, not inside the code. Second: everything is idempotent. Every handler can receive the same message any number of times without corrupting state. Third: diagnostics come in automatically. When a device crashes, it uploads an ELF coredump on its own — nobody has to call a customer and say “please send me the log”.

// Staged firmware rollout: 1% → 10% → 100% with a health gate between waves.
export async function scheduleRollout(v: string) {
  const stages = [0.01, 0.10, 1.00];
  for (const share of stages) {
    await rolloutTo(v, share);
    await waitForHealthGate(v, { minHours: 24 });
  }
}

Highlights

import { z } from 'zod';

export const StateV1 = z.object({
  device_id: z.string().uuid(),
  fw_version: z.string(),
  brightness: z.number().int().min(0).max(100),
  color_kelvin: z.number().int().min(2200).max(6500),
  on: z.boolean(),
  seq: z.number().int(),          // monotonic, for dedup
  ts: z.iso.datetime(),
});
export type StateV1 = z.infer<typeof StateV1>;

CREATE OR REPLACE FUNCTION public.process_device_state_update(
  p_device_id uuid, p_event_data jsonb, p_state_data jsonb
) RETURNS void LANGUAGE plpgsql SECURITY DEFINER AS $$
BEGIN
  -- 1. Event first — trigger snapshots previous_state from device_state.
  INSERT INTO device_events (device_id, event_type, event_data, source, source_id)
  VALUES (p_device_id, 'state_change', p_event_data, 'device', p_device_id::text);

  -- 2. Upsert device_state, only fields present in payload.
  INSERT INTO device_state (device_id, power, brightness, rgb_color, online, last_seen)
  VALUES (p_device_id,
          (p_state_data->>'power')::boolean,
          (p_state_data->>'brightness')::int,
          (p_state_data->>'rgb_color')::int,
          true, NOW())
  ON CONFLICT (device_id) DO UPDATE SET
    power      = COALESCE((p_state_data->>'power')::boolean, device_state.power),
    brightness = COALESCE((p_state_data->>'brightness')::int, device_state.brightness),
    rgb_color  = COALESCE((p_state_data->>'rgb_color')::int, device_state.rgb_color),
    online     = true,
    last_seen  = NOW();
END $$;

async function handleCoredumpSummary(deviceId: string, summary: CoredumpSummary) {
  if (!summary?.elf_sha256 || !summary?.pc || summary?.cause === undefined) {
    return; // drop incomplete summary
  }

  const result = await db.upsertCrashSummary(deviceId, summary);
  if (!result) return;

  // Auto-trigger raw upload only on first occurrence of this signature.
  if (result.isNew && result.rawElfStatus === 'pending') {
    const customerId = await db.getDeviceCustomerId(deviceId);
    const commandId = await db.queueGetCoredump(deviceId, customerId);
    await db.markCrashUploadRequested(result.crashId, commandId);
    assembler.registerExpectedUpload(commandId, deviceId, customerId);
  }
}

@override
Future<void> subscribe() async {
  await unsubscribe();
  final userId = _supabase.auth.currentUser?.id;
  if (userId == null) return;

  // 1. Snapshot via GraphQL — the app starts with an accurate baseline.
  await _fetchInitialStates(userId);

  // 2. Per-user broadcast channel for deltas.
  _channel = _supabase
      .channel('user:$userId', opts: const RealtimeChannelConfig(private: true))
      .onBroadcast(event: 'device_state_change', callback: _onBroadcast)
      .subscribe();
}

private async rearmTimers(deviceId: string): Promise<void> {
  const rows = await this.db.fetchActiveTimersForDevice(deviceId);
  if (rows.length === 0) return;

  for (const row of rows) {
    const remainingMs = new Date(row.expires_at).getTime() - Date.now();
    const seconds = Math.max(1, Math.floor(remainingMs / 1000));
    await this.db.queueTimerArm(deviceId, row.automation_id, {
      execution_id: row.execution_id,  // same slot → idempotent
      seconds,
      inner_steps: row.inner_steps,
    });
  }
}

Throughput math, in 118 bytes of WebAssembly

The whole architectural story above — per-user broadcast, dedup at the consumer, idempotent state — exists because the naive answer (every device, every state change, fanned out everywhere) does not scale. Move the sliders to see why:

What mattered in production

Three things that didn’t show up in load tests but turned out to be decisive in production.

ACK topology isn’t symmetric. Devices ACK on d/<id>/ack, but the ID form differs by source: UUIDs for command_id, locally generated string ids for timer fires. The consumer routes by which field is present (command_id vs timer_id), not by separate topics — a consolidation that cut a lot of complexity later.

OTA rollouts rarely fail at the flash, more often at re-verify. After a successful flash, the device boots, verifies the new partition and sends ota_success. If it hangs between boot and verify, we treat it as ota_failed after 10 minutes and roll back. Without that watchdog, at least one bad build would have bricked half the fleet.

Ops scripts go through the same RLS layer. Internal scripts (bulk updates, diagnostic exports) authenticate via Supabase service-role and write into device_events with source = 'admin'. That turned out to be gold during audit: every change to customer data has a traceable origin, even when manually triggered.