Improve error handling

Author: Bre77

.changeset/huge-chefs-change.md

@@ -6,3 +6,14 @@ Standardize SSE event handling to use EventEmitter pattern:
 - Remove `onConnection()` helper method - use `on("connect")` and `on("disconnect")` instead
 - Add `"all"` event type to subscribe to all SSE events at once
 - Update documentation and examples to reflect new patterns
+
+Fix unhandled promise rejections in SSE streaming:
+- Add error handling to `connect()` method's internal connection loop
+- `updateFields()` now returns a promise that resolves/rejects when the debounced update completes
+- Multiple calls within the debounce window share the same promise
+- `addField()` now properly returns the promise from `updateFields()`
+
+Refactor `updateFields()` internals:
+- Consolidate batch state into single `_fieldUpdateBatch` object
+- Extract flush logic into `_flushFieldUpdate()` method
+- Use Deferred pattern for cleaner promise handling

.changeset/node-red-sse-events.md

@@ -6,4 +6,15 @@ Update SSE event handling to use standard EventEmitter pattern:
 - Replace `onData()`, `onState()`, etc. with `on("data")`, `on("state")` pattern
 - Use new `on("all")` event for subscribing to all events
 - Add VIN filtering in event callbacks
-- Emit catchable errors only on initial connection failure (not on disconnect)
+
+Improve error handling:
+- Config node tracks errors from initial API verification (auth/subscription issues)
+- All nodes now show "Error" status with red ring when API verification fails
+- Nodes wait for API verification before attempting SSE connections
+- Clearer error messages in node logs
+
+Refactor shared helpers:
+- Add `getInstance()` to get and validate config instance
+- Add `hasInstanceError()` for synchronous error checking (command nodes)
+- Add `verifyInstance()` for async verification before SSE connection
+- Add `getErrorMessage()` to extract useful messages from any error type (handles hey-api response objects)

packages/api/src/TeslemetryVehicleStream.ts

@@ -20,6 +20,20 @@ import type {
 import { Logger } from "./logger.js";
 import { VehicleCache } from "./TeslemetryStream.js";
 
+/** Simple deferred promise pattern */
+class Deferred<T> {
+  public promise: Promise<T>;
+  public resolve!: (value: T) => void;
+  public reject!: (error: Error) => void;
+
+  constructor() {
+    this.promise = new Promise<T>((resolve, reject) => {
+      this.resolve = resolve;
+      this.reject = reject;
+    });
+  }
+}
+
 type TeslemetryStreamEventMap = {
   state: SseState;
   data: SseData;
@@ -77,8 +91,11 @@ export class TeslemetryVehicleStream extends EventEmitter {
   private root: Teslemetry;
   public vin: string;
   public fields: FieldsRequest = {}; // Allow updates from both requests, and responses
-  private _pendingFields: FieldsRequest = {}; // Used for accumulating config changes before patching
-  private _debounceTimeout: NodeJS.Timeout | null = null; // Debounce timeout for patchConfig
+  private _fieldUpdateBatch: {
+    fields: FieldsRequest;
+    deferred: Deferred<void>;
+    timeout: NodeJS.Timeout;
+  } | null = null;
   public logger: Logger;
   public data: TeslemetryVehicleStreamData;
 
@@ -128,33 +145,56 @@ export class TeslemetryVehicleStream extends EventEmitter {
     }
   }
 
-  /** Safely add field configuration to the vehicle */
-  public async updateFields(fields: FieldsRequest) {
-    this._pendingFields = { ...this._pendingFields, ...fields };
-
-    // Clear existing timeout if it exists
-    if (this._debounceTimeout) {
-      clearTimeout(this._debounceTimeout);
+  /** Safely add field configuration to the vehicle
+   * Returns a promise that resolves when the debounced update completes.
+   * Multiple calls within the debounce window share the same promise.
+   */
+  public updateFields(fields: FieldsRequest): Promise<void> {
+    if (!this._fieldUpdateBatch) {
+      this._fieldUpdateBatch = {
+        fields: {},
+        deferred: new Deferred<void>(),
+        timeout: null!,
+      };
     }
 
-    // Set new timeout to debounce patchConfig calls
-    this._debounceTimeout = setTimeout(async () => {
-      const data = await this.patchConfig(this._pendingFields);
+    this._fieldUpdateBatch.fields = {
+      ...this._fieldUpdateBatch.fields,
+      ...fields,
+    };
+
+    clearTimeout(this._fieldUpdateBatch.timeout);
+    this._fieldUpdateBatch.timeout = setTimeout(
+      () => this._flushFieldUpdate(),
+      1000,
+    );
+
+    return this._fieldUpdateBatch.deferred.promise;
+  }
+
+  /** Flush pending field updates to the API */
+  private async _flushFieldUpdate(): Promise<void> {
+    const batch = this._fieldUpdateBatch!;
+    this._fieldUpdateBatch = null;
+
+    try {
+      const data = await this.patchConfig(batch.fields);
       if (data?.updated_vehicles) {
         this.logger.info(
-          `Updated ${Object.keys(this._pendingFields).length} streaming fields for ${this.vin}`,
+          `Updated ${Object.keys(batch.fields).length} streaming fields for ${this.vin}`,
         );
-        // null means default, not delete, so its okay
-        this.fields = { ...this.fields, ...this._pendingFields };
-        this._pendingFields = {};
+        this.fields = { ...this.fields, ...batch.fields };
+        batch.deferred.resolve();
       } else {
-        this.logger.error(
+        const error = new Error(
           `Error updating streaming config for ${this.vin}`,
-          data,
         );
+        this.logger.error(error.message, data);
+        batch.deferred.reject(error);
       }
-      this._debounceTimeout = null;
-    }, 1000);
+    } catch (error: any) {
+      batch.deferred.reject(error);
+    }
   }
 
   /** Modify the field configuration of the vehicle */
@@ -181,9 +221,9 @@ export class TeslemetryVehicleStream extends EventEmitter {
    * Add a field to the vehicles streaming configuration
    * @param field Vehicle Signal
    * @param interval
-   * @returns
+   * @returns Promise that resolves when the field is added
    */
-  public async addField(field: Signals, interval?: number): Promise<void> {
+  public addField(field: Signals, interval?: number): Promise<void> {
     if (
       this.fields &&
       this.fields[field] &&
@@ -193,12 +233,12 @@ export class TeslemetryVehicleStream extends EventEmitter {
       this.logger.debug(
         `Streaming field ${field} already enabled @ ${this.fields[field]?.interval_seconds || "default"}s`,
       );
-      return;
+      return Promise.resolve();
     }
 
     const value =
       interval !== undefined ? { interval_seconds: interval } : null;
-    this.updateFields({ [field]: value } as FieldsRequest);
+    return this.updateFields({ [field]: value } as FieldsRequest);
   }
 
   /**
@@ -211,8 +251,8 @@ export class TeslemetryVehicleStream extends EventEmitter {
     field: T,
     callback: (value: Exclude<SseData["data"][T], undefined>) => void,
   ): () => void {
-    this.addField(field).catch((error) => {
-      this.logger.error(`Failed to add field ${field}:`, error);
+    this.addField(field).catch(() => {
+      this.logger.error(`Failed to add field ${field}`);
     });
     const data = this.root.sse.cache?.[this.vin]?.data?.[field];
     if (data !== undefined) {

packages/node-red-contrib-teslemetry/src/nodes/teslemetry-config.ts

@@ -1,6 +1,6 @@
 import { Node, NodeAPI, NodeDef } from "node-red";
 import { Teslemetry } from "@teslemetry/api";
-import { instances } from "../shared";
+import { instances, getErrorMessage } from "../shared";
 
 export interface TeslemetryConfigNodeDef extends NodeDef {
   token: string;
@@ -22,11 +22,24 @@ export default function (RED: NodeAPI) {
         logger: RED.log,
         stream: { cache: false },
       });
-      instances.set(this.id, {
+
+      // Create instance first so it can be referenced in catch handler
+      const instance = {
         teslemetry,
-        products: teslemetry.createProducts(),
-      });
-      console.log("Created", this.id);
+        products: Promise.resolve({ vehicles: {}, energySites: {} }),
+        error: undefined as string | undefined,
+      };
+      instances.set(this.id, instance);
+
+      // Fetch products and track errors
+      instance.products = teslemetry
+        .createProducts()
+        .catch((error: unknown) => {
+          const message = getErrorMessage(error);
+          RED.log.error(`Teslemetry error: ${message}`);
+          instance.error = message;
+          return { vehicles: {}, energySites: {} };
+        });
     }
 
     this.on("close", (done: () => void) => {
@@ -41,79 +54,91 @@ export default function (RED: NodeAPI) {
   });
 
   RED.httpAdmin.get("/teslemetry/vehicles", async (req: any, res: any) => {
-    const config = req.query.config;
+    try {
+      const config = req.query.config;
 
-    if (!config) {
-      res.status(400).send("Missing config ID");
-      return;
-    }
+      if (!config) {
+        res.status(400).send("Missing config ID");
+        return;
+      }
 
-    // Try to get the active node first
-    const instance = instances.get(req.query.config);
+      const instance = instances.get(req.query.config);
 
-    if (!instance) {
-      res.status(400).send("Missing config instance");
-      return;
-    }
+      if (!instance) {
+        res.status(400).send("Missing config instance");
+        return;
+      }
 
-    const { vehicles } = await instance.products;
+      const { vehicles } = await instance.products;
 
-    const options = Object.entries(vehicles).map(([id, { name }]) => [
-      id,
-      name,
-    ]);
+      const options = Object.entries(vehicles).map(([id, { name }]) => [
+        id,
+        name,
+      ]);
 
-    res.json(options);
+      res.json(options);
+    } catch (error: unknown) {
+      RED.log.error(`Failed to fetch vehicles: ${getErrorMessage(error)}`);
+      res.status(500).send("Failed to fetch vehicles");
+    }
   });
 
   RED.httpAdmin.get("/teslemetry/energy_sites", async (req: any, res: any) => {
-    const config = req.query.config;
+    try {
+      const config = req.query.config;
 
-    if (!config) {
-      res.status(400).send("Missing config ID");
-      return;
-    }
+      if (!config) {
+        res.status(400).send("Missing config ID");
+        return;
+      }
 
-    // Try to get the active node first
-    const instance = instances.get(req.query.config);
+      const instance = instances.get(req.query.config);
 
-    if (!instance) {
-      res.status(400).send("Missing config instance");
-      return;
-    }
+      if (!instance) {
+        res.status(400).send("Missing config instance");
+        return;
+      }
 
-    const { energySites } = await instance.products;
+      const { energySites } = await instance.products;
 
-    const options = Object.entries(energySites).map(([id, { name }]) => [
-      id,
-      name,
-    ]);
+      const options = Object.entries(energySites).map(([id, { name }]) => [
+        id,
+        name,
+      ]);
 
-    res.json(options);
+      res.json(options);
+    } catch (error: unknown) {
+      RED.log.error(`Failed to fetch energy sites: ${getErrorMessage(error)}`);
+      res.status(500).send("Failed to fetch energy sites");
+    }
   });
 
   RED.httpAdmin.get("/teslemetry/fields", async (req: any, res: any) => {
-    const { config, model } = req.query;
-
-    if (!config) {
-      res.status(400).send("Missing config ID");
-      return;
-    }
-
-    // Try to get the active node first
-    const instance = instances.get(req.query.config);
-
-    if (!instance) {
-      res.status(400).send("Missing config instance");
-      return;
+    try {
+      const { config, model } = req.query;
+
+      if (!config) {
+        res.status(400).send("Missing config ID");
+        return;
+      }
+
+      const instance = instances.get(req.query.config);
+
+      if (!instance) {
+        res.status(400).send("Missing config instance");
+        return;
+      }
+
+      const fields = await instance.teslemetry.api.getFields();
+      const options = Object.entries(fields)
+        .filter(([_, { models }]) => {
+          return model && models ? models.includes(model) : true;
+        })
+        .map(([signal]) => signal);
+      res.json(options);
+    } catch (error: unknown) {
+      RED.log.error(`Failed to fetch fields: ${getErrorMessage(error)}`);
+      res.status(500).send("Failed to fetch fields");
     }
-
-    const fields = await instance.teslemetry.api.fields();
-    const options = Object.entries(fields)
-      .filter(([_, { models }]) => {
-        return model && models ? models.includes(model) : true;
-      })
-      .map(([signal]) => signal);
-    res.json(options);
   });
 }

packages/node-red-contrib-teslemetry/src/nodes/teslemetry-energy-command.ts

@@ -1,6 +1,6 @@
 import { Node, NodeAPI, NodeDef } from "node-red";
 import { Teslemetry } from "@teslemetry/api";
-import { instances } from "../shared";
+import { getInstance, hasInstanceError } from "../shared";
 import { validateParameters } from "../validation";
 import { Msg } from "../types";
 
@@ -24,15 +24,14 @@ export default function (RED: NodeAPI) {
     RED.nodes.createNode(this, config);
     const node = this;
 
-    node.teslemetry = instances.get(config.teslemetryConfig)?.teslemetry;
+    const instance = getInstance(config.teslemetryConfig, node);
+    if (!instance) return;
+    if (hasInstanceError(instance, node)) return;
+
+    node.teslemetry = instance.teslemetry;
     node.siteId = config.siteId;
     node.command = config.command;
-
-    if (!node.teslemetry) {
-      node.status({ fill: "red", shape: "ring", text: "Config missing" });
-      node.error("No Teslemetry configuration found");
-      return;
-    } else node.status({});
+    node.status({});
 
     node.on("input", async function (msg: Msg, send, done) {
       const siteId: string = node.siteId || (msg.siteId as string) || "";