fix(hubspot): simplify integration to only create contacts

api/app/settings/common.py

@@ -1187,17 +1187,6 @@
 # Hubspot settings
 HUBSPOT_ACCESS_TOKEN = env.str("HUBSPOT_ACCESS_TOKEN", None)
 ENABLE_HUBSPOT_LEAD_TRACKING = env.bool("ENABLE_HUBSPOT_LEAD_TRACKING", False)
-HUBSPOT_IGNORE_DOMAINS = env.list(
-    "HUBSPOT_IGNORE_DOMAINS",
-    subcast=str,
-    default=[],
-)
-HUBSPOT_IGNORE_DOMAINS_REGEX = env.str("HUBSPOT_IGNORE_DOMAINS_REGEX", "")
-HUBSPOT_IGNORE_ORGANISATION_DOMAINS = env.list(
-    "HUBSPOT_IGNORE_ORGANISATION_DOMAINS",
-    subcast=str,
-    default=[],
-)
 
 # Number of minutes to wait for a user that has signed up to
 # join or create an organisation before creating a lead in

api/integrations/lead_tracking/hubspot/client.py

@@ -5,7 +5,6 @@
 import hubspot  # type: ignore[import-untyped]
 import requests
 from django.conf import settings
-from hubspot.crm.associations.v4 import AssociationSpec  # type: ignore[import-untyped]
 from hubspot.crm.companies import (  # type: ignore[import-untyped]
     PublicObjectSearchRequest,
     SimplePublicObjectInput,
@@ -116,21 +115,6 @@ def create_lead_form(
             )
         return response.json()  # type: ignore[no-any-return]
 
-    def associate_contact_to_company(self, contact_id: str, company_id: str) -> None:
-        association_spec = [
-            AssociationSpec(
-                association_category="HUBSPOT_DEFINED", association_type_id=1
-            )
-        ]
-
-        self.client.crm.associations.v4.basic_api.create(
-            object_type="contacts",
-            object_id=contact_id,
-            to_object_type="companies",
-            to_object_id=company_id,
-            association_spec=association_spec,
-        )
-
     def get_company_by_domain(self, domain: str) -> dict[str, Any] | None:
         """
         Domain should be unique in Hubspot by design, so we should only ever have

api/integrations/lead_tracking/hubspot/lead_tracker.py

@@ -5,66 +5,19 @@
 from django.conf import settings
 
 from integrations.lead_tracking.lead_tracking import LeadTracker
-from organisations.models import (
-    HubspotOrganisation,
-    Organisation,
-    Subscription,
-)
+from organisations.models import Organisation
 from users.models import FFAdminUser, HubspotLead, HubspotTracker
 
 from .client import HubspotClient
 from .constants import HUBSPOT_FORM_ID_SAAS
 
 logger = logging.getLogger(__name__)
 
-try:
-    import re2 as re  # type: ignore[import-untyped]
-
-    logger.info("Using re2 library for regex.")
-except ImportError:
-    logger.warning("Unable to import re2. Falling back to re.")
-    import re
-
 
 class HubspotLeadTracker(LeadTracker):
     @staticmethod
     def should_track(user: FFAdminUser) -> bool:
-        if not settings.ENABLE_HUBSPOT_LEAD_TRACKING:
-            return False
-
-        domain = user.email_domain
-
-        if settings.HUBSPOT_IGNORE_DOMAINS_REGEX and re.match(
-            settings.HUBSPOT_IGNORE_DOMAINS_REGEX, domain
-        ):
-            return False
-
-        if (
-            settings.HUBSPOT_IGNORE_DOMAINS
-            and domain in settings.HUBSPOT_IGNORE_DOMAINS
-        ):
-            return False
-
-        return True
-
-    def update_company_active_subscription(
-        self, subscription: Subscription
-    ) -> dict[str, Any] | None:
-        if not subscription.plan:
-            return None
-
-        organisation = subscription.organisation
-
-        # Check if we're missing the associated hubspot id.
-        if not getattr(organisation, "hubspot_organisation", None):
-            return None
-
-        response: dict[str, Any] | None = self.client.update_company(
-            active_subscription=subscription.plan,
-            hubspot_company_id=organisation.hubspot_organisation.hubspot_id,
-        )
-
-        return response
+        return settings.ENABLE_HUBSPOT_LEAD_TRACKING
 
     def create_user_hubspot_contact(self, user: FFAdminUser) -> str | None:
         tracker = HubspotTracker.objects.filter(user=user).first()
@@ -96,17 +49,9 @@ def create_user_hubspot_contact(self, user: FFAdminUser) -> str | None:
         return hubspot_contact_id
 
     def create_lead(self, user: FFAdminUser, organisation: Organisation) -> None:
-        hubspot_contact_id = self._get_or_create_user_hubspot_id(user)
-        if not hubspot_contact_id:
-            return
-        hubspot_org_id = self._get_organisation_hubspot_id(user, organisation)
-        if not hubspot_org_id:
-            return
-
-        self.client.associate_contact_to_company(
-            contact_id=hubspot_contact_id,
-            company_id=hubspot_org_id,
-        )
+        # Only create the contact. HubSpot handles company creation and
+        # association automatically from the contact's email domain.
+        self._get_or_create_user_hubspot_id(user)
 
     def _get_new_contact_with_retry(
         self, user: FFAdminUser, max_retries: int = 3
@@ -138,58 +83,5 @@ def _get_or_create_user_hubspot_id(self, user: FFAdminUser) -> str | None:
 
         return hubspot_contact_id
 
-    def _get_organisation_hubspot_id(
-        self,
-        user: FFAdminUser,
-        organisation: Organisation,
-    ) -> str | None:
-        """
-        Return the Hubspot API's id for an organisation.
-        """
-        if getattr(organisation, "hubspot_organisation", None):
-            return organisation.hubspot_organisation.hubspot_id
-
-        if user.email_domain in settings.HUBSPOT_IGNORE_ORGANISATION_DOMAINS:
-            return None
-
-        domain = user.email_domain
-        company_kwargs = {"domain": domain}
-        company_kwargs["name"] = organisation.name
-        company_kwargs["organisation_id"] = organisation.id
-        company_kwargs["active_subscription"] = organisation.subscription.plan
-
-        # As Hubspot creates/associates companies automatically based on contact domain
-        # we need to get the hubspot id when this user creates the company for the first time
-        # and update the company name
-        company = self._get_hubspot_company_by_domain(domain)
-        if not company:
-            return None
-        org_hubspot_id: str = company["id"]
-
-        # Update the company in Hubspot with the name of the created
-        # organisation in Flagsmith, and its numeric ID.
-        self.client.update_company(
-            name=organisation.name,
-            hubspot_company_id=org_hubspot_id,
-            flagsmith_organisation_id=organisation.id,
-        )
-
-        # Store the organisation data in the database since we are
-        # unable to look them up via a unique identifier.
-        HubspotOrganisation.objects.create(
-            organisation=organisation,
-            hubspot_id=org_hubspot_id,
-        )
-
-        return org_hubspot_id
-
-    def _get_hubspot_company_by_domain(
-        self,
-        domain: str,
-    ) -> dict[str, Any]:
-        company = self.client.get_company_by_domain(domain)
-
-        return company  # type: ignore[no-any-return]
-
     def _get_client(self) -> HubspotClient:
         return HubspotClient()

api/integrations/lead_tracking/hubspot/tasks.py

@@ -51,19 +51,6 @@ def create_hubspot_contact_for_user(user_id: int) -> None:
     hubspot_lead_tracker.create_user_hubspot_contact(user)
 
 
-@register_task_handler()
-def update_hubspot_active_subscription(subscription_id: int) -> None:
-    assert settings.ENABLE_HUBSPOT_LEAD_TRACKING
-
-    from organisations.models import Subscription
-
-    from .lead_tracker import HubspotLeadTracker
-
-    subscription = Subscription.objects.get(id=subscription_id)
-    hubspot_lead_tracker = HubspotLeadTracker()
-    hubspot_lead_tracker.update_company_active_subscription(subscription)
-
-
 @register_task_handler()
 def create_self_hosted_onboarding_lead_task(
     email: str, first_name: str, last_name: str, hubspot_cookie: str = ""

api/organisations/models.py

@@ -21,7 +21,6 @@
 from features.versioning.constants import DEFAULT_VERSION_LIMIT_DAYS
 from integrations.lead_tracking.hubspot.tasks import (
     track_hubspot_lead_v2,
-    update_hubspot_active_subscription,
 )
 from organisations.chargebee import (  # type: ignore[attr-defined]
     get_customer_id_from_subscription_id,
@@ -305,13 +304,6 @@ def update_api_limit_access_block(self):  # type: ignore[no-untyped-def]
         self.organisation.block_access_to_admin = False
         self.organisation.save()
 
-    @hook(AFTER_SAVE, when="plan", has_changed=True)
-    def update_hubspot_active_subscription(self):  # type: ignore[no-untyped-def]
-        if not settings.ENABLE_HUBSPOT_LEAD_TRACKING:
-            return
-
-        update_hubspot_active_subscription.delay(args=(self.id,))
-
     def save_as_free_subscription(self):  # type: ignore[no-untyped-def]
         """
         Wipes a subscription to a normal free plan.

api/tests/unit/integrations/lead_tracking/hubspot/test_unit_hubspot_client.py

@@ -3,7 +3,6 @@
 
 import pytest
 import responses
-from hubspot.crm.associations.v4 import AssociationSpec  # type: ignore[import-untyped]
 from pytest_mock import MockerFixture
 from rest_framework import status
 
@@ -216,32 +215,6 @@ def test_create_company__without_organisation_info__creates_with_name_and_domain
     }
 
 
-def test_associate_contact_to_company__valid_ids__calls_hubspot_api(
-    hubspot_client: HubspotClient,
-) -> None:
-    # Given
-    company_id = "456"
-    contact_id = "123"
-
-    # When
-    hubspot_client.associate_contact_to_company(
-        contact_id=contact_id, company_id=company_id
-    )
-
-    # Then
-    hubspot_client.client.crm.associations.v4.basic_api.create.assert_called_once_with(
-        object_type="contacts",
-        object_id=contact_id,
-        to_object_type="companies",
-        to_object_id=company_id,
-        association_spec=[
-            AssociationSpec(
-                association_category="HUBSPOT_DEFINED", association_type_id=1
-            )
-        ],
-    )
-
-
 @pytest.mark.parametrize(
     "kwargs, expected_properties",
     [