FlowGlow Lighting

Architektur in vier Schichten

Die Plattform ist um vier klar getrennte Schichten gebaut, und jede Schicht spricht über ein einziges, dokumentiertes Protokoll mit den Nachbarn:

1. Firmware (ESP32 / C++): baut nach dem Boot WLAN auf, holt Credentials aus dem NVS und verbindet sich per TLS mit dem Mosquitto-Broker. Sie veröffentlicht Statusänderungen auf d/<id>/state, OTA-Events auf d/<id>/ota_events, Crash-Summaries auf d/<id>/coredump und akzeptiert eingehende Kommandos auf d/<id>/cmd.
2. MQTT-Consumer (Node.js): abonniert alle Geräte-Topics, validiert die Payloads, dedupliziert, und schreibt das Ergebnis transaktional in Postgres. Veröffentlicht außerdem Kommandos in die andere Richtung, indem er aus einer lamp_commands-Tabelle pollt.
3. Supabase / Postgres: ist die Single Source of Truth. RLS-Policies regeln Sichtbarkeit pro Kunde, ein BEFORE INSERT-Trigger auf device_events snappt jeden Zustandsübergang, und ein AFTER UPDATE-Trigger auf device_state broadcastet die Änderung an genau den richtigen Nutzer-Kanal.
4. Clients: eine Flutter-App für Endkunden (iOS, Android), ein Angular-Dashboard für Support und Service. Beide reden ausschließlich GraphQL für Lese-Zugriffe und Realtime-Broadcast für Live-Updates — kein direkter MQTT-Pfad ins Frontend.

Die strikte Trennung bedeutet, dass jede Schicht einzeln deploybar ist: Firmware-Updates rollen über die Geräte, der Consumer ist ein zustandsloser Worker hinter Coolify, das Dashboard ist eine SPA, die Flutter-App ein App-Store-Build. Kein „Big-Deploy”-Moment im Kalender, kein Lock-Step zwischen Frontend und Backend.

Vorgehen

Drei Designentscheidungen ziehen sich durch das gesamte Backend und sind der Grund, weshalb das System auch mit Hunderten Geräten gleichzeitig stabil läuft.

Erstens: nichts blockiert. Der Consumer arbeitet rein async, jede Nachricht ist eine eigene Transaktion. Wenn Postgres einmal lahmt, stauen sich Nachrichten im Broker, nicht im Code. Zweitens: alles ist idempotent. Jeder Handler kann eine Nachricht beliebig oft empfangen, ohne dass dadurch falsche Daten entstehen. Drittens: Diagnose-Daten kommen automatisch. Stürzt ein Gerät ab, lädt es eigenständig einen ELF-Coredump hoch — niemand muss einen Kunden anrufen und „bitte schicken Sie mir das Log” sagen.

// Staged firmware rollout: 1% → 10% → 100% mit Health-Gate dazwischen.
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 });
  }
}

Tiefe Einblicke

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 zuerst — Trigger snappt previous_state aus 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, nur Felder die im Payload stehen.
  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; // unvollständige Summary verwerfen
  }

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

  // Auto-trigger raw upload nur beim ersten Auftreten dieser Crash-Signatur.
  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 per GraphQL — die App startet immer mit aktuellem Stand.
  await _fetchInitialStates(userId);

  // 2. Per-User-Broadcast-Channel für 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,  // gleicher Slot → idempotent
      seconds,
      inner_steps: row.inner_steps,
    });
  }
}

Durchsatz, gerechnet in 118 Byte WebAssembly

Die ganze Architektur-Story oben — Per-User-Broadcast, Dedup im Consumer, idempotenter State — existiert, weil die naive Antwort (jedes Gerät, jede Änderung, Fan-out überall) nicht skaliert. Schieben Sie die Regler, um zu sehen, warum:

Was im Betrieb wichtig wurde

Drei Dinge, die im Lasttest nicht aufgefallen sind, sich aber im Realbetrieb als ausschlaggebend herausgestellt haben:

Ack-Topologie ist nicht symmetrisch. Geräte schicken Acks auf d/<id>/ack, aber je nach Quelle (Cloud-Command vs. lokaler Timer) ist die ID-Form anders: UUID für command_id, lokal generierte String-IDs für Timer-Fires. Der Consumer routet anhand des Feldes (command_id vs. timer_id), nicht über separate Topics — eine spätere Konsolidierung, die viel Komplexität rausgenommen hat.

OTA-Roll-outs scheitern selten am Update selbst, häufiger am Re-Verify. Nach einem erfolgreichen Flash bootet das Gerät, verifiziert die neue Partition und sendet ota_success. Hängt es zwischen Boot und Verify, betrachten wir das nach 10 Minuten als ota_failed und rollen zurück. Ohne diesen Watchdog hätte schon das eine oder andere Build die halbe Flotte ge-bricked.

Operations-Skripte greifen über dieselbe RLS-Ebene zu. Auch interne Skripte (Massen-Updates, Diagnose-Exports) gehen über die Supabase-Service-Role und werden in den device_events mit source = 'admin' protokolliert. Das hat sich beim Audit als Gold wert herausgestellt: jede Änderung an Kundendaten hat eine nachvollziehbare Quelle, auch wenn sie manuell ausgelöst war.