Update nest config flow to dramatically simplify end user setup with automated pub/sub subscription creation (#59260)

* Configure nest pubsub subscriber automatically

Update the config flow to configure the nest pubsub subscriber automatically.
After completing the authentication step, the user is now asked for the google
cloud console ID, which is needed to create a subscription.

Home Assistant manages the lifecycle of a subscription only when it is created
by the ConfigFlow. Otherwise (if specified in configuration.yaml) it treats
it similarly as before.

These are the considerations or failure modes taken into account:
- Subscription is created with reasonable default values as previously recommended (e.g. retion only keeps 5-15 minutes of backlog messages)
- Subscriptions are created with a naming scheme that makes it clear they came from home assistant, and with a random
  string
- Subscriptions are cleaned up when the ConfigEntry is removed. If removal fails, a subscription that is orphaned will
  be deleted after 30 days
- If the subscription gets into a bad state or deleted, the user can go through the re-auth flow to re-create it.
- Users can still specifcy a CONF_SUBSCRIBER_ID in the configuration.yaml, and
skip automatic subscriber creation

* Remove unnecessary nest config flow diffs and merge in upstream changes

* Incorporate review feedback into nest subscription config flow

* Update text wording in nest config flow

homeassistant/components/nest/api.py

@@ -1,6 +1,9 @@
 """API for Google Nest Device Access bound to Home Assistant OAuth."""
 
+from __future__ import annotations
+
 import datetime
+import logging
 from typing import cast
 
 from aiohttp import ClientSession
@@ -23,7 +26,7 @@ from .const import (
     SDM_SCOPES,
 )
 
-# See https://developers.google.com/nest/device-access/registration
+_LOGGER = logging.getLogger(__name__)
 
 
 class AsyncConfigEntryAuth(AbstractAuth):
@@ -71,14 +74,31 @@ class AsyncConfigEntryAuth(AbstractAuth):
 
 async def new_subscriber(
     hass: HomeAssistant, entry: ConfigEntry
-) -> GoogleNestSubscriber:
+) -> GoogleNestSubscriber | None:
     """Create a GoogleNestSubscriber."""
     implementation = (
         await config_entry_oauth2_flow.async_get_config_entry_implementation(
             hass, entry
         )
     )
+    config = hass.data[DOMAIN][DATA_NEST_CONFIG]
+    if not (
+        subscriber_id := entry.data.get(
+            CONF_SUBSCRIBER_ID, config.get(CONF_SUBSCRIBER_ID)
+        )
+    ):
+        _LOGGER.error("Configuration option 'subscriber_id' required")
+        return None
+    return await new_subscriber_with_impl(hass, entry, subscriber_id, implementation)
 
+
+async def new_subscriber_with_impl(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    subscriber_id: str,
+    implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
+) -> GoogleNestSubscriber:
+    """Create a GoogleNestSubscriber, used during ConfigFlow."""
     config = hass.data[DOMAIN][DATA_NEST_CONFIG]
     session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
     auth = AsyncConfigEntryAuth(
@@ -87,6 +107,4 @@ async def new_subscriber(
         config[CONF_CLIENT_ID],
         config[CONF_CLIENT_SECRET],
     )
-    return GoogleNestSubscriber(
-        auth, config[CONF_PROJECT_ID], config[CONF_SUBSCRIBER_ID]
-    )
+    return GoogleNestSubscriber(auth, config[CONF_PROJECT_ID], subscriber_id)