Use subscription for config flows in progress (#24985)

src/data/config_flow.ts

@@ -1,7 +1,5 @@
 import type { Connection } from "home-assistant-js-websocket";
-import { getCollection } from "home-assistant-js-websocket";
 import type { LocalizeFunc } from "../common/translations/localize";
-import { debounce } from "../common/util/debounce";
 import type { HomeAssistant } from "../types";
 import type {
   DataEntryFlowProgress,
@@ -93,31 +91,20 @@ export const fetchConfigFlowInProgress = (
     type: "config_entries/flow/progress",
   });
 
-const subscribeConfigFlowInProgressUpdates = (conn: Connection, store) =>
-  conn.subscribeEvents(
-    debounce(
-      () =>
-        fetchConfigFlowInProgress(conn).then((flows: DataEntryFlowProgress[]) =>
-          store.setState(flows, true)
-        ),
-      500,
-      true
-    ),
-    "config_entry_discovered"
-  );
-
-export const getConfigFlowInProgressCollection = (conn: Connection) =>
-  getCollection<DataEntryFlowProgress[]>(
-    conn,
-    "_configFlowProgress",
-    fetchConfigFlowInProgress,
-    subscribeConfigFlowInProgressUpdates
-  );
+export interface ConfigFlowInProgressMessage {
+  type: null | "added" | "removed";
+  flow_id: string;
+  flow: DataEntryFlowProgress;
+}
 
 export const subscribeConfigFlowInProgress = (
   hass: HomeAssistant,
-  onChange: (flows: DataEntryFlowProgress[]) => void
-) => getConfigFlowInProgressCollection(hass.connection).subscribe(onChange);
+  onChange: (update: ConfigFlowInProgressMessage[]) => void
+) =>
+  hass.connection.subscribeMessage<ConfigFlowInProgressMessage[]>(
+    (message) => onChange(message),
+    { type: "config_entries/flow/subscribe" }
+  );
 
 export const localizeConfigFlowTitle = (
   localize: LocalizeFunc,

src/onboarding/onboarding-integrations.ts

@@ -35,16 +35,20 @@ class OnboardingIntegrations extends SubscribeMixin(LitElement) {
 
   @state() private _entries: ConfigEntry[] = [];
 
-  @state() private _discoveredDomains?: Set<string>;
+  @state() private _discoveredDomains = new Set<string>();
 
   public hassSubscribe(): (UnsubscribeFunc | Promise<UnsubscribeFunc>)[] {
     return [
-      subscribeConfigFlowInProgress(this.hass, (flows) => {
-        this._discoveredDomains = new Set(
-          flows
-            .filter((flow) => !HIDDEN_DOMAINS.has(flow.handler))
-            .map((flow) => flow.handler)
-        );
+      subscribeConfigFlowInProgress(this.hass, (messages) => {
+        messages.forEach((message) => {
+          if (
+            message.type === "removed" ||
+            HIDDEN_DOMAINS.has(message.flow.handler)
+          ) {
+            return;
+          }
+          this._discoveredDomains.add(message.flow.handler);
+        });
         this.hass.loadBackendTranslation(
           "title",
           Array.from(this._discoveredDomains)
@@ -93,7 +97,7 @@ class OnboardingIntegrations extends SubscribeMixin(LitElement) {
   }
 
   protected render() {
-    if (!this._discoveredDomains) {
+    if (!this._discoveredDomains.size) {
       return nothing;
     }
     // Render discovered and existing entries together sorted by localized title.

src/panels/config/integrations/ha-config-integrations-dashboard.ts

@@ -27,7 +27,6 @@ import "../../../components/search-input";
 import "../../../components/search-input-outlined";
 import type { ConfigEntry } from "../../../data/config_entries";
 import { getConfigEntries } from "../../../data/config_entries";
-import { getConfigFlowInProgressCollection } from "../../../data/config_flow";
 import { fetchDiagnosticHandlers } from "../../../data/diagnostics";
 import type { EntityRegistryEntry } from "../../../data/entity_registry";
 import { subscribeEntityRegistry } from "../../../data/entity_registry";
@@ -754,7 +753,6 @@ class HaConfigIntegrationsDashboard extends KeyboardShortcutMixin(
   }
 
   private _handleFlowUpdated() {
-    getConfigFlowInProgressCollection(this.hass.connection).refresh();
     this._reScanImprovDevices();
     this._fetchManifests();
   }

src/panels/config/integrations/ha-config-integrations.ts

@@ -77,6 +77,12 @@ class HaConfigIntegrations extends SubscribeMixin(HassRouterPage) {
       subscribeConfigEntries(
         this.hass,
         async (messages) => {
+          if (messages.length === 0) {
+            this._configEntries = [];
+            return;
+          }
+          let fullUpdate = this._configEntries === undefined;
+          const newEntries: ConfigEntryExtended[] = [];
           await this._loadTranslationsPromise?.then(
             () =>
               // allow hass to update
@@ -84,8 +90,6 @@ class HaConfigIntegrations extends SubscribeMixin(HassRouterPage) {
                 window.setTimeout(resolve, 0);
               })
           );
-          let fullUpdate = this._configEntries === undefined;
-          const newEntries: ConfigEntryExtended[] = [];
           messages.forEach((message) => {
             if (message.type === null || message.type === "added") {
               newEntries.push({
@@ -114,30 +118,60 @@ class HaConfigIntegrations extends SubscribeMixin(HassRouterPage) {
               );
             }
           });
-          if (!newEntries.length && !fullUpdate) {
-            return;
-          }
           const existingEntries = fullUpdate ? [] : this._configEntries;
           this._configEntries = [...existingEntries!, ...newEntries];
         },
         { type: ["device", "hub", "service", "hardware"] }
       ),
-      subscribeConfigFlowInProgress(this.hass, async (flowsInProgress) => {
-        const integrations = new Set<string>();
-        flowsInProgress.forEach((flow) => {
-          // To render title placeholders
-          if (flow.context.title_placeholders) {
-            integrations.add(flow.handler);
+      subscribeConfigFlowInProgress(this.hass, async (messages) => {
+        if (messages.length === 0) {
+          this._configEntriesInProgress = [];
+          return;
+        }
+
+        let fullUpdate = this._configEntriesInProgress === undefined;
+        const newEntries: DataEntryFlowProgressExtended[] = [];
+
+        messages.forEach((message) => {
+          if (message.type === "removed") {
+            if (!this._configEntriesInProgress) {
+              return;
+            }
+            this._configEntriesInProgress =
+              this._configEntriesInProgress.filter(
+                (flow) => flow.flow_id !== message.flow_id
+              );
+            return;
+          }
+
+          if (message.type === null || message.type === "added") {
+            if (message.type === null) {
+              fullUpdate = true;
+            }
+
+            newEntries.push(message.flow);
           }
         });
-        const localize = await this.hass.loadBackendTranslation(
-          "config",
-          Array.from(integrations)
-        );
-        this._configEntriesInProgress = flowsInProgress.map((flow) => ({
-          ...flow,
-          localized_title: localizeConfigFlowTitle(localize, flow),
-        }));
+
+        if (!newEntries.length && !fullUpdate) {
+          return;
+        }
+        const existingEntries = fullUpdate ? [] : this._configEntriesInProgress;
+
+        const titleIntegrations = newEntries
+          .filter((flow) => flow.context.title_placeholders)
+          .map((flow) => flow.handler);
+        const localize = titleIntegrations.length
+          ? await this.hass.loadBackendTranslation("config", titleIntegrations)
+          : this.hass.localize;
+
+        this._configEntriesInProgress = [
+          ...existingEntries!,
+          ...newEntries.map((flow) => ({
+            ...flow,
+            localized_title: localizeConfigFlowTitle(localize, flow),
+          })),
+        ];
       }),
     ];
   }