diff --git a/.github/workflows/api-dev.yml b/.github/workflows/api-dev.yml index a06b68f2b..3a77bf8c5 100644 --- a/.github/workflows/api-dev.yml +++ b/.github/workflows/api-dev.yml @@ -26,7 +26,7 @@ jobs: OPERATIONS_OAUTH2_CLIENT_ID_1PASSWORD: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/GCP_RETOOL_OAUTH2_CREDS/username" WEB_APP_REVALIDATE_URL: "https://staging.mobilitydatabase.org/api/revalidate" WEB_APP_REVALIDATE_SECRET_1PASSWORD: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/MobilityDatabase Vercel Deployment/REVALIDATE_SECRET_QA" - BREVO_API_ANNOUNCEMENTS_LIST_ID: ${{ vars.BREVO_API_ANNOUNCEMENTS_LIST_ID }} + BREVO_API_ANNOUNCEMENTS_LIST_ID: ${{ vars.DEV_BREVO_API_ANNOUNCEMENTS_LIST_ID }} secrets: GCP_MOBILITY_FEEDS_SA_KEY: ${{ secrets.DEV_GCP_MOBILITY_FEEDS_SA_KEY }} OAUTH2_CLIENT_ID: ${{ secrets.DEV_MOBILITY_FEEDS_OAUTH2_CLIENT_ID}} diff --git a/.github/workflows/api-prod.yml b/.github/workflows/api-prod.yml index efdf0c393..a52233a0d 100644 --- a/.github/workflows/api-prod.yml +++ b/.github/workflows/api-prod.yml @@ -22,7 +22,7 @@ jobs: OPERATIONS_OAUTH2_CLIENT_ID_1PASSWORD: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/GCP_RETOOL_OAUTH2_CREDS/username" WEB_APP_REVALIDATE_URL: "https://mobilitydatabase.org/api/revalidate" WEB_APP_REVALIDATE_SECRET_1PASSWORD: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/MobilityDatabase Vercel Deployment/REVALIDATE_SECRET" - BREVO_API_ANNOUNCEMENTS_LIST_ID: ${{ vars.BREVO_API_ANNOUNCEMENTS_LIST_ID }} + BREVO_API_ANNOUNCEMENTS_LIST_ID: ${{ vars.PROD_BREVO_API_ANNOUNCEMENTS_LIST_ID }} secrets: GCP_MOBILITY_FEEDS_SA_KEY: ${{ secrets.PROD_GCP_MOBILITY_FEEDS_SA_KEY }} OAUTH2_CLIENT_ID: ${{ secrets.PROD_MOBILITY_FEEDS_OAUTH2_CLIENT_ID}} diff --git a/.github/workflows/api-qa.yml b/.github/workflows/api-qa.yml index 04a13e763..801740785 100644 --- a/.github/workflows/api-qa.yml +++ b/.github/workflows/api-qa.yml @@ -22,7 +22,7 @@ jobs: OPERATIONS_OAUTH2_CLIENT_ID_1PASSWORD: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/GCP_RETOOL_OAUTH2_CREDS/username" WEB_APP_REVALIDATE_URL: "https://staging.mobilitydatabase.org/api/revalidate" WEB_APP_REVALIDATE_SECRET_1PASSWORD: "op://rbiv7rvkkrsdlpcrz3bmv7nmcu/MobilityDatabase Vercel Deployment/REVALIDATE_SECRET_QA" - BREVO_API_ANNOUNCEMENTS_LIST_ID: ${{ vars.BREVO_API_ANNOUNCEMENTS_LIST_ID }} + BREVO_API_ANNOUNCEMENTS_LIST_ID: ${{ vars.QA_BREVO_API_ANNOUNCEMENTS_LIST_ID }} secrets: GCP_MOBILITY_FEEDS_SA_KEY: ${{ secrets.QA_GCP_MOBILITY_FEEDS_SA_KEY }} OAUTH2_CLIENT_ID: ${{ secrets.DEV_MOBILITY_FEEDS_OAUTH2_CLIENT_ID}} diff --git a/api/src/shared/common/brevo.py b/api/src/shared/common/brevo.py index 75ccf717d..50d8c6c05 100644 --- a/api/src/shared/common/brevo.py +++ b/api/src/shared/common/brevo.py @@ -73,17 +73,34 @@ def get_announcements_list_id() -> int: return int(raw) -def add_contact_to_list(email: str, list_id: int, subscription_id: str) -> None: - """Create/update a Brevo contact, add it to the list, and set MDB_SUBSCRIPTION_ID. +def add_contact_to_list( + email: str, + list_id: int, + subscription_id: str, + *, + first_name: str | None = None, + organization: str | None = None, +) -> None: + """Create/update a Brevo contact, add it to the list, and set its attributes. + + Always sets MDB_SUBSCRIPTION_ID. When provided, also sets the standard + FIRSTNAME attribute (from the user's full name) and the ORGANIZATION attribute + (from the user's legacy organisation name). Attributes whose value is None are + omitted so we never overwrite an existing Brevo value with a blank. Uses create_contact with update_enabled so it works whether or not the contact already exists. """ + attributes = {"MDB_SUBSCRIPTION_ID": subscription_id} + if first_name is not None: + attributes["FIRSTNAME"] = first_name + if organization is not None: + attributes["ORGANIZATION"] = organization api = _get_contacts_api() api.create_contact( sib_api_v3_sdk.CreateContact( email=email, - attributes={"MDB_SUBSCRIPTION_ID": subscription_id}, + attributes=attributes, list_ids=[list_id], update_enabled=True, ), diff --git a/api/src/user_service/impl/subscription_helpers.py b/api/src/user_service/impl/subscription_helpers.py index abd99b4f1..0ccac11bb 100644 --- a/api/src/user_service/impl/subscription_helpers.py +++ b/api/src/user_service/impl/subscription_helpers.py @@ -16,27 +16,131 @@ """Helpers shared between the authenticated (users) and public (subscriptions) APIs.""" import logging +from datetime import datetime, timezone from fastapi import HTTPException import sib_api_v3_sdk import urllib3 -from shared.common.brevo import add_contact_to_list, get_announcements_list_id, remove_contact_from_list +from sqlalchemy.orm import Session +from shared.common.brevo import ( + add_contact_to_list, + get_announcements_list_id, + remove_contact_from_list, +) +from shared.database.database import generate_unique_id +from shared.users_database_gen.sqlacodegen_models import ( + AppUser, + NotificationSubscription as NotificationSubscriptionOrm, +) logger = logging.getLogger(__name__) ANNOUNCEMENTS_NOTIFICATION_TYPE_ID = "api.announcements" -def sync_announcements(email: str, subscribe: bool, subscription_id: str | None = None) -> None: +def sync_announcements( + email: str, + subscribe: bool, + subscription_id: str | None = None, + *, + first_name: str | None = None, + organization: str | None = None, +) -> None: """Sync an api.announcements subscription with Brevo, mapping provider errors to 502.""" try: if subscribe: - add_contact_to_list(email, get_announcements_list_id(), subscription_id) + add_contact_to_list( + email, + get_announcements_list_id(), + subscription_id, + first_name=first_name, + organization=organization, + ) else: remove_contact_from_list(email, get_announcements_list_id()) - except (RuntimeError, sib_api_v3_sdk.rest.ApiException, urllib3.exceptions.HTTPError, OSError) as exc: + except ( + RuntimeError, + sib_api_v3_sdk.rest.ApiException, + urllib3.exceptions.HTTPError, + OSError, + ) as exc: # urllib3.exceptions.HTTPError / OSError cover connection failures and timeouts (e.g. Brevo # unreachable), so the request fails fast with a 502 instead of hanging on retries. logger.error("Brevo sync failed for %s: %s", email, exc) raise HTTPException(status_code=502, detail="Failed to sync subscription with email provider.") + + +def set_announcements_optin( + db_session: Session, + user: AppUser, + subscribe: bool, + *, + subscription: NotificationSubscriptionOrm | None = None, + release_connection_before_brevo: bool = False, +) -> NotificationSubscriptionOrm | None: + """Reconcile a user's api.announcements opt-in across all three representations + that must agree: the notification_subscription row, Brevo list membership, and + app_user.is_registered_to_receive_api_announcements. Idempotent. + + subscribe=True → create/reactivate the subscription, add the contact to the + Brevo list with its MDB_SUBSCRIPTION_ID, set the flag True. + subscribe=False → deactivate the subscription (api.announcements is never + deleted, only disabled), remove the contact from the Brevo + list, set the flag False. + Returns the subscription row (None only when unsubscribing a user that has no + subscription row). + + ``subscription``: the caller's already-loaded announcements subscription, if it + has one (endpoints that address a subscription by id). When omitted the row is + looked up by user + type (there is at most one announcements subscription per + user). + + ``release_connection_before_brevo`` commits the pending transaction before the + Brevo call so a slow/unreachable provider does not hold a pooled DB connection. + Only pass True when there are no pending writes you would want rolled back if + the Brevo call fails (e.g. the delete endpoints, which only read beforehand). + On the subscribe path the connection is always held so the whole change stays + atomic (a Brevo failure rolls the new subscription + flag back). + """ + existing = subscription + if existing is None: + existing = ( + db_session.query(NotificationSubscriptionOrm) + .filter( + NotificationSubscriptionOrm.user_id == user.id, + NotificationSubscriptionOrm.notification_type_id == ANNOUNCEMENTS_NOTIFICATION_TYPE_ID, + ) + .one_or_none() + ) + + if subscribe: + sub = existing or NotificationSubscriptionOrm( + id=generate_unique_id(), + user_id=user.id, + notification_type_id=ANNOUNCEMENTS_NOTIFICATION_TYPE_ID, + created_at=datetime.now(timezone.utc), + ) + sub.active = True + if existing is None: + db_session.add(sub) + sync_announcements( + user.email, + subscribe=True, + subscription_id=sub.id, + first_name=user.full_name, + organization=user.legacy_org_name, + ) + user.is_registered_to_receive_api_announcements = True + return sub + + # Unsubscribe. Capture the email before any commit so we never trigger a + # reload of the (possibly expired) user just to read it during the Brevo call. + email = user.email + if release_connection_before_brevo: + db_session.commit() + sync_announcements(email, subscribe=False) + if existing is not None: + existing.active = False + user.is_registered_to_receive_api_announcements = False + return existing diff --git a/api/src/user_service/impl/subscriptions_api_impl.py b/api/src/user_service/impl/subscriptions_api_impl.py index 355a0ec95..83c010767 100644 --- a/api/src/user_service/impl/subscriptions_api_impl.py +++ b/api/src/user_service/impl/subscriptions_api_impl.py @@ -24,7 +24,7 @@ ) from user_service.impl.subscription_helpers import ( ANNOUNCEMENTS_NOTIFICATION_TYPE_ID, - sync_announcements, + set_announcements_optin, ) from user_service_gen.apis.subscriptions_api_base import BaseSubscriptionsApi from user_service_gen.models.notification_subscription import NotificationSubscription @@ -55,14 +55,20 @@ def delete_subscription(self, id: str, db_session=None) -> None: if sub.notification_type_id == ANNOUNCEMENTS_NOTIFICATION_TYPE_ID: user = db_session.get(AppUser, sub.user_id) - email = user.email if user is not None else None - # Release the pooled DB connection before the (potentially slow) Brevo call so a slow - # or unreachable provider never holds a connection while we talk to it. The read above - # took no row locks, so committing here only returns the connection to the pool. - db_session.commit() - if email is not None: - sync_announcements(email, subscribe=False) - sub.active = False + if user is not None: + # release_connection_before_brevo=True: only reads happened above, + # so committing just returns the pooled connection before the + # (possibly slow) Brevo call. Also clears the app_user opt-in flag. + set_announcements_optin( + db_session, + user, + subscribe=False, + subscription=sub, + release_connection_before_brevo=True, + ) + else: + # Orphaned subscription (no owning user): just disable it. + sub.active = False else: db_session.delete(sub) db_session.flush() diff --git a/api/src/user_service/impl/users_api_impl.py b/api/src/user_service/impl/users_api_impl.py index eb3f5a4c2..eaa2b7394 100644 --- a/api/src/user_service/impl/users_api_impl.py +++ b/api/src/user_service/impl/users_api_impl.py @@ -36,7 +36,7 @@ ) from user_service.impl.subscription_helpers import ( ANNOUNCEMENTS_NOTIFICATION_TYPE_ID, - sync_announcements, + set_announcements_optin, ) from user_service_gen.apis.users_api_base import BaseUsersApi from user_service_gen.models.create_notification_subscription_request import ( @@ -114,9 +114,28 @@ def update_user(self, update_user_request: UpdateUserRequest, db_session=None) - raise HTTPException(status_code=404, detail="User not found.") update_data = update_user_request.model_dump(exclude_unset=True) + + # Detect a real change to the announcements opt-in so we can reconcile the + # subscription + Brevo membership below. Writing the boolean alone would + # leave the DB flag, the notification_subscription row, and Brevo out of + # sync (the historical bug this closes). Only act on an actual change so + # unrelated profile edits never make a (slow, failable) Brevo call. + optin_field = "is_registered_to_receive_api_announcements" + optin_target = None + if optin_field in update_data: + new_value = bool(update_data[optin_field]) + if new_value != bool(user.is_registered_to_receive_api_announcements): + optin_target = new_value + for field, value in update_data.items(): setattr(user, field, value) user.updated_at = datetime.now(timezone.utc) + + if optin_target is not None: + # Holds the DB connection through the Brevo call so the profile write + # and the opt-in change commit (or roll back on a 502) atomically. + set_announcements_optin(db_session, user, subscribe=optin_target) + db_session.flush() all_flags = db_session.query(FeatureFlag).filter(FeatureFlag.disabled.is_(False)).order_by(FeatureFlag.id).all() @@ -151,28 +170,30 @@ def create_user_subscription( if user is None: raise HTTPException(status_code=404, detail="User not found.") - # Idempotent: reuse an existing subscription, reactivating if needed. - existing = ( - db_session.query(NotificationSubscriptionOrm) - .filter( - NotificationSubscriptionOrm.user_id == user_id, - NotificationSubscriptionOrm.notification_type_id == notification_id, - ) - .one_or_none() - ) - sub = existing or NotificationSubscriptionOrm( - id=generate_unique_id(), - user_id=user_id, - notification_type_id=notification_id, - created_at=datetime.now(timezone.utc), - ) - sub.active = True - if notification_id == ANNOUNCEMENTS_NOTIFICATION_TYPE_ID: - sync_announcements(user.email, subscribe=True, subscription_id=sub.id) + # Reconcile the subscription row, Brevo membership and the app_user + # opt-in flag together so all three stay consistent. + sub = set_announcements_optin(db_session, user, subscribe=True) + else: + # Idempotent: reuse an existing subscription, reactivating if needed. + existing = ( + db_session.query(NotificationSubscriptionOrm) + .filter( + NotificationSubscriptionOrm.user_id == user_id, + NotificationSubscriptionOrm.notification_type_id == notification_id, + ) + .one_or_none() + ) + sub = existing or NotificationSubscriptionOrm( + id=generate_unique_id(), + user_id=user_id, + notification_type_id=notification_id, + created_at=datetime.now(timezone.utc), + ) + sub.active = True + if existing is None: + db_session.add(sub) - if existing is None: - db_session.add(sub) db_session.flush() return NotificationSubscriptionImpl.from_orm(sub) @@ -187,9 +208,9 @@ def update_user_subscription( active = update_notification_subscription_request.active if sub.notification_type_id == ANNOUNCEMENTS_NOTIFICATION_TYPE_ID: user = db_session.get(AppUser, user_id) - sync_announcements(user.email, subscribe=active, subscription_id=sub.id) - - sub.active = active + set_announcements_optin(db_session, user, subscribe=active, subscription=sub) + else: + sub.active = active db_session.flush() return NotificationSubscriptionImpl.from_orm(sub) @@ -203,13 +224,17 @@ def delete_user_subscription(self, id: str, db_session=None) -> None: sub = self._get_owned_subscription(db_session, id, user_id) if sub.notification_type_id == ANNOUNCEMENTS_NOTIFICATION_TYPE_ID: - email = db_session.get(AppUser, user_id).email - # Release the pooled DB connection before the (potentially slow) Brevo call so a slow - # or unreachable provider never holds a connection while we talk to it. The reads above - # took no row locks, so committing here only returns the connection to the pool. - db_session.commit() - sync_announcements(email, subscribe=False) - sub.active = False + user = db_session.get(AppUser, user_id) + # release_connection_before_brevo=True: only reads happened above, so + # committing just returns the pooled connection before the (possibly + # slow) Brevo call. Also clears the app_user opt-in flag. + set_announcements_optin( + db_session, + user, + subscribe=False, + subscription=sub, + release_connection_before_brevo=True, + ) else: db_session.delete(sub) db_session.flush() diff --git a/api/tests/unittest/test_brevo_add_contact.py b/api/tests/unittest/test_brevo_add_contact.py new file mode 100644 index 000000000..82c9b3768 --- /dev/null +++ b/api/tests/unittest/test_brevo_add_contact.py @@ -0,0 +1,57 @@ +# +# MobilityData 2026 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +from unittest.mock import MagicMock, patch + +from shared.common import brevo + + +def _sent_contact(api_mock): + """Return the CreateContact object passed to create_contact().""" + return api_mock.create_contact.call_args[0][0] + + +def test_add_contact_sets_only_subscription_id_by_default(): + api = MagicMock() + with patch.object(brevo, "_get_contacts_api", return_value=api): + brevo.add_contact_to_list("a@b.com", 42, "sub-1") + + contact = _sent_contact(api) + assert contact.attributes == {"MDB_SUBSCRIPTION_ID": "sub-1"} + assert contact.list_ids == [42] + assert contact.update_enabled is True + + +def test_add_contact_includes_firstname_and_organization_when_provided(): + api = MagicMock() + with patch.object(brevo, "_get_contacts_api", return_value=api): + brevo.add_contact_to_list("a@b.com", 42, "sub-1", first_name="Jane Doe", organization="Acme Transit") + + assert _sent_contact(api).attributes == { + "MDB_SUBSCRIPTION_ID": "sub-1", + "FIRSTNAME": "Jane Doe", + "ORGANIZATION": "Acme Transit", + } + + +def test_add_contact_omits_none_attributes(): + api = MagicMock() + with patch.object(brevo, "_get_contacts_api", return_value=api): + brevo.add_contact_to_list("a@b.com", 42, "sub-1", first_name=None, organization="Acme Transit") + + attributes = _sent_contact(api).attributes + assert "FIRSTNAME" not in attributes # None is omitted, not blanked + assert attributes["ORGANIZATION"] == "Acme Transit" + assert attributes["MDB_SUBSCRIPTION_ID"] == "sub-1" diff --git a/api/tests/unittest/user_service/test_users_api_impl.py b/api/tests/unittest/user_service/test_users_api_impl.py index 04cc0f035..1ab26e8bd 100644 --- a/api/tests/unittest/user_service/test_users_api_impl.py +++ b/api/tests/unittest/user_service/test_users_api_impl.py @@ -167,16 +167,80 @@ def test_updates_full_name(self): self.mock_session.flush.assert_called_once() self.assertEqual(result.full_name, "New Name") - def test_updates_api_announcements_flag(self): + def test_opt_in_creates_subscription_and_syncs_brevo(self): + """Flipping the flag false→true creates the announcements subscription and + writes the contact to Brevo (the disconnect this fix closes).""" user = _make_user(is_registered_to_receive_api_announcements=False) - _mock_query_first(self.mock_session, user) + user_q = _mock_query_first(self.mock_session, user) + user_q.filter.return_value.one_or_none.return_value = None # no existing sub _set_context() req = self._make_request(is_registered_to_receive_api_announcements=True) - result = self.api.update_user(req, db_session=self.mock_session) + with patch.object(helpers, "add_contact_to_list") as add, patch.object( + helpers, "get_announcements_list_id", return_value=42 + ): + result = self.api.update_user(req, db_session=self.mock_session) self.assertTrue(user.is_registered_to_receive_api_announcements) self.assertTrue(result.is_registered_to_receive_api_announcements) + add.assert_called_once() # Brevo write-back happened + self.assertEqual(add.call_args[0][:2], ("user@example.com", 42)) + self.mock_session.add.assert_called_once() # subscription row created + + def test_opt_out_deactivates_subscription_and_removes_from_brevo(self): + """Flipping the flag true→false removes the contact from Brevo and clears + the flag.""" + user = _make_user(is_registered_to_receive_api_announcements=True) + user_q = _mock_query_first(self.mock_session, user) + existing_sub = MagicMock(active=True) + user_q.filter.return_value.one_or_none.return_value = existing_sub + _set_context() + + req = self._make_request(is_registered_to_receive_api_announcements=False) + with patch.object(helpers, "remove_contact_from_list") as rem, patch.object( + helpers, "get_announcements_list_id", return_value=42 + ): + result = self.api.update_user(req, db_session=self.mock_session) + + rem.assert_called_once_with("user@example.com", 42) + self.assertFalse(existing_sub.active) + self.assertFalse(user.is_registered_to_receive_api_announcements) + self.assertFalse(result.is_registered_to_receive_api_announcements) + + def test_flag_unchanged_does_not_touch_brevo(self): + """Re-sending the same flag value (or updating other fields) makes no + Brevo call.""" + user = _make_user(is_registered_to_receive_api_announcements=True) + _mock_query_first(self.mock_session, user) + _set_context() + + req = self._make_request(is_registered_to_receive_api_announcements=True) + with patch.object(helpers, "add_contact_to_list") as add, patch.object( + helpers, "remove_contact_from_list" + ) as rem: + self.api.update_user(req, db_session=self.mock_session) + + add.assert_not_called() + rem.assert_not_called() + + def test_opt_in_brevo_failure_raises_502(self): + """A Brevo failure while flipping the flag surfaces as 502 (and, in real + DB use, rolls the whole update back).""" + import sib_api_v3_sdk + + user = _make_user(is_registered_to_receive_api_announcements=False) + user_q = _mock_query_first(self.mock_session, user) + user_q.filter.return_value.one_or_none.return_value = None + _set_context() + + req = self._make_request(is_registered_to_receive_api_announcements=True) + with patch.object( + helpers, "add_contact_to_list", side_effect=sib_api_v3_sdk.rest.ApiException(status=500) + ), patch.object(helpers, "get_announcements_list_id", return_value=42): + with self.assertRaises(HTTPException) as ctx: + self.api.update_user(req, db_session=self.mock_session) + + self.assertEqual(ctx.exception.status_code, 502) def test_partial_update_leaves_other_fields_unchanged(self): user = _make_user(full_name="Unchanged", is_registered_to_receive_api_announcements=True) diff --git a/functions-python/tasks_executor/README.md b/functions-python/tasks_executor/README.md index 22c244b58..9ce560e89 100644 --- a/functions-python/tasks_executor/README.md +++ b/functions-python/tasks_executor/README.md @@ -176,7 +176,7 @@ If the header is not provided, the default response content type is `application ### `migrate_firebase_users` -Migrates Firebase Auth users into the `users.app_user` PostgreSQL table. This task is **insert-only** — existing rows are never modified. Brevo is the source of truth for `is_registered_to_receive_api_announcements`. +Migrates Firebase Auth users into the `users.app_user` PostgreSQL table. `app_user` is **insert-only** with a single exception: the task sets `app_user.brevo_synced_at` on an existing row after it writes that user's `MDB_SUBSCRIPTION_ID` back onto the Brevo contact. It also ensures each user has an `api.announcements` `notification_subscription`. Brevo is the source of truth for `is_registered_to_receive_api_announcements`. ```json { @@ -192,14 +192,21 @@ Migrates Firebase Auth users into the `users.app_user` PostgreSQL table. This ta | Parameter | Type | Default | Description | |---|---|---|---| -| `dry_run` | bool | `true` | Read and count without any DB writes. Brevo is still queried so counts are accurate | +| `dry_run` | bool | `true` | Read and count without any DB writes or Brevo write-back. Brevo status is still queried so counts are accurate | | `limit` | int \| null | `null` | Maximum number of users to process per run; `null` means no limit | | `user_ids` | list[str] \| null | `null` | If provided, only migrate these specific Firebase UIDs | | `only_not_migrated` | bool | `true` | Skip users that already have a row in `app_user` with `migrated_at` set | **Brevo subscription logic**: For each new user, `BREVO_API_ANNOUNCEMENTS_LIST_ID` is checked. If the contact is `SUBSCRIBED`, `is_registered_to_receive_api_announcements` is set to `true`; `UNSUBSCRIBED` sets it to `false`; `NOT_FOUND` leaves it at the DB default (`false`). -**Announcements subscription association**: Every migrated user is associated with the `api.announcements` notification type via a row in `notification_subscription`. New users get the subscription created alongside their `app_user` row; existing users (including already-migrated ones) are **backfilled** a subscription if they don't already have one — without modifying their `app_user` row. The subscription is created enabled (`active=true`) for all users **except** those explicitly `UNSUBSCRIBED` on Brevo, who get a disabled (`active=false`) subscription. Users that are `NOT_FOUND` on Brevo (or whose Brevo check failed) are treated as "not unsubscribed" and therefore enabled. The step is idempotent (users that already have the subscription are untouched). Counts are reported as `announcements_enabled` / `announcements_disabled`. +**Announcements subscription association**: Every migrated user is associated with the `api.announcements` notification type via a row in `notification_subscription`. New users get the subscription created alongside their `app_user` row; existing users (including already-migrated ones) are **backfilled** a subscription if they don't already have one — without modifying their `app_user` row. The subscription is created enabled (`active=true`) for all users **except** those explicitly `UNSUBSCRIBED` on Brevo, who get a disabled (`active=false`) subscription. Users that are `NOT_FOUND` on Brevo (or whose Brevo check failed) are treated as "not unsubscribed" and therefore enabled. New-subscription counts are reported as `announcements_enabled` / `announcements_disabled`. + +**Brevo contact write-back (`MDB_SUBSCRIPTION_ID`)**: For every user whose announcements subscription should be on the Brevo list, the contact is written via `add_contact_to_list` so its `MDB_SUBSCRIPTION_ID` attribute points at the subscription id. `app_user.brevo_synced_at` records that write — it lives on `app_user` (not `notification_subscription`) because `api.announcements` is the only Brevo-delivered notification type and a Brevo contact maps 1:1 to a user. The decision is evaluated identically on every run (no "first run" / "last run" logic): + +- `brevo_synced_at IS NULL` → the contact is assumed to lack `MDB_SUBSCRIPTION_ID`, so it is (re)written and `brevo_synced_at` is stamped; +- `brevo_synced_at` set → already synced, skipped (no Brevo call). + +The write-back is **skipped** for contacts Brevo reports as `UNSUBSCRIBED` (they are never re-added to the list) and whenever the Brevo check fails (retried on the next run). `brevo_synced_at` is stamped **only after a successful write** and **never in dry-run mode**. Setting `brevo_synced_at` is the only update the task makes to an existing `app_user` row. Counts are reported as `brevo_synced` / `brevo_sync_failed`. Requires `BREVO_API_ANNOUNCEMENTS_LIST_ID` to be set; without it the write-back is skipped. **Datastore entity lookup**: For each new user, the `web_api_users` kind is queried by the `uid` property to retrieve `fullName`, `organization`, and `registrationCompletionTime`. @@ -220,6 +227,8 @@ Migrates Firebase Auth users into the `users.app_user` PostgreSQL table. This ta | `brevo_unsubscribed` | Users found as unsubscribed in Brevo | | `brevo_not_found` | Users not found in Brevo | | `brevo_failed` | Users where the Brevo check failed (non-fatal; user is still inserted) | +| `brevo_synced` | Contacts written back to Brevo with `MDB_SUBSCRIPTION_ID` (in dry-run, contacts that would be written) | +| `brevo_sync_failed` | Contacts where the Brevo write-back failed (non-fatal; retried next run) | | `dry_run` | Whether the task ran in dry-run mode | ### `notifications_dispatch_batch` (+ `notifications_dispatch`, `notifications_dispatch_monitor`) diff --git a/functions-python/tasks_executor/src/main.py b/functions-python/tasks_executor/src/main.py index b791310c6..7f0b43c6f 100644 --- a/functions-python/tasks_executor/src/main.py +++ b/functions-python/tasks_executor/src/main.py @@ -173,10 +173,15 @@ }, "migrate_firebase_users": { "description": ( - "Insert-only migration of Firebase Auth users into users.app_user. " + "Insert-only migration of Firebase Auth users into users.app_user " + "(the only field updated on an existing row is brevo_synced_at). " "Reads profile fields (fullName, organization, registrationCompletionTime) " "from Datastore kind 'web_api_users' (queried by uid property). " "Uses Brevo as the source of truth for is_registered_to_receive_api_announcements. " + "Ensures each user has an api.announcements notification_subscription and " + "writes MDB_SUBSCRIPTION_ID back onto the Brevo contact, tracking that " + "write via app_user.brevo_synced_at (idempotent: users already synced " + "are skipped). " "Parameters: dry_run (default true), limit (default null), " "user_ids (default null), only_not_migrated (default true)." ), diff --git a/functions-python/tasks_executor/src/tasks/users/migrate_firebase_users.py b/functions-python/tasks_executor/src/tasks/users/migrate_firebase_users.py index ff900f6e8..b7f025bee 100644 --- a/functions-python/tasks_executor/src/tasks/users/migrate_firebase_users.py +++ b/functions-python/tasks_executor/src/tasks/users/migrate_firebase_users.py @@ -28,7 +28,9 @@ SUBSCRIBED → True UNSUBSCRIBED → False NOT_FOUND → field not set (left at DB default false for new rows, untouched for existing) - app_user.migrated_at <- now() (set by this task) + app_user.migrated_at <- now() (set by this task, on insert) + app_user.brevo_synced_at <- now() (set by this task once the Brevo + contact has been written with its MDB_SUBSCRIPTION_ID; see below) Announcements subscription (notification_subscription table): Every migrated user is associated with the ``api.announcements`` notification @@ -40,6 +42,25 @@ whose Brevo check failed) are treated as not unsubscribed and therefore enabled. This step is idempotent: users that already have the subscription are untouched. +Brevo contact write-back (issue #1733): + For every user whose announcements subscription should be on the Brevo list, + the contact is (re)written via ``add_contact_to_list`` so its + ``MDB_SUBSCRIPTION_ID`` attribute points at the subscription id. + ``app_user.brevo_synced_at`` records that this write happened. It lives on + ``app_user`` (not ``notification_subscription``) because api.announcements is + the only Brevo-delivered notification type and a Brevo contact maps 1:1 to a + user. A NULL value means the contact has not been synced yet (its + ``MDB_SUBSCRIPTION_ID`` is assumed unset) and it is (re)written on the next + run. This is evaluated identically on every run — no notion of a "first" or + "last" run: + - brevo_synced_at IS NULL → write the contact back and stamp brevo_synced_at + - brevo_synced_at IS set → already synced, skipped (no Brevo call) + The write-back is skipped for users Brevo reports as UNSUBSCRIBED (we must not + re-add them to the list) and whenever the Brevo check fails (retried next run). + ``brevo_synced_at`` is stamped only after a successful write, and never in + dry-run mode. This is the ONE case where the task updates an existing + ``app_user`` row; everything else about ``app_user`` is insert-only. + NOT set by migration (managed by the API layer): app_user.updated_at """ @@ -58,6 +79,7 @@ from shared.common.brevo import ( BrevoSubscriptionStatus, + add_contact_to_list, get_contact_subscription_status, ) from shared.database.database import generate_unique_id @@ -144,16 +166,22 @@ def _iter_users(user_ids: list[str] | None) -> Generator[auth.UserRecord, None, page = page.get_next_page() -def _has_announcements_subscription(db_session: Session, user_id: str) -> bool: - """Return True if the user already has an api.announcements subscription.""" +def _get_announcements_subscription( + db_session: Session, user_id: str +) -> NotificationSubscription | None: + """Return the user's api.announcements subscription row, or None if absent. + + The full row (not just a boolean) is needed downstream: the subscription id + is written to Brevo as MDB_SUBSCRIPTION_ID (app_user.brevo_synced_at drives + the write-back idempotency check). + """ return ( - db_session.query(NotificationSubscription.id) + db_session.query(NotificationSubscription) .filter( NotificationSubscription.user_id == user_id, NotificationSubscription.notification_type_id == API_ANNOUNCEMENTS_TYPE_ID, ) .first() - is not None ) @@ -164,7 +192,8 @@ def migrate_firebase_users( only_not_migrated: bool = True, db_session: Session | None = None, ) -> dict: - """Core migration logic. Only INSERTs new users; existing rows are never modified. + """Core migration logic. INSERTs new users; the only field it updates on an + existing app_user row is brevo_synced_at (after a successful Brevo write-back). Args: dry_run: When True (default), reads and counts without any DB writes. @@ -177,7 +206,8 @@ def migrate_firebase_users( Returns: Summary dict with counts: total, inserted, skipped, no_email_skipped, brevo_subscribed, brevo_unsubscribed, brevo_not_found, brevo_failed, - announcements_enabled, announcements_disabled, dry_run. + announcements_enabled, announcements_disabled, brevo_synced, + brevo_sync_failed, dry_run. """ _get_firebase_app() ds_client = datastore.Client() @@ -205,6 +235,8 @@ def migrate_firebase_users( "brevo_failed": 0, "announcements_enabled": 0, "announcements_disabled": 0, + "brevo_synced": 0, + "brevo_sync_failed": 0, "dry_run": dry_run, } processed = 0 @@ -232,15 +264,22 @@ def migrate_firebase_users( ): results["skipped"] += 1 - # Idempotency: every user must end up with exactly one announcements - # subscription. New users always need one; existing users only if absent. - needs_announcements_sub = ( - existing is None - or not _has_announcements_subscription(db_session, user_record.uid) + subscription = _get_announcements_subscription(db_session, user_record.uid) + + # Idempotency signals, evaluated the same way on every run (no notion of + # a "first" vs "later" run): + # - every user must have exactly one announcements subscription; + # - the user's Brevo contact must carry MDB_SUBSCRIPTION_ID, which + # app_user.brevo_synced_at records — NULL means it still needs + # writing. A user gaining a new subscription (needs_subscription) + # also needs its (new) id pushed to Brevo. + needs_subscription = subscription is None + needs_brevo_sync = ( + existing is None or needs_subscription or existing.brevo_synced_at is None ) - # Nothing to do for an existing user that already has the subscription. - if existing is not None and not needs_announcements_sub: + # Nothing to do for an existing, already-synced user that has a subscription. + if existing is not None and not needs_subscription and not needs_brevo_sync: continue # Datastore profile lookup is only needed when inserting a new app_user. @@ -265,9 +304,10 @@ def migrate_firebase_users( user_record.uid, ) - # Brevo is the source of truth for subscription status. Queried whenever - # we will create the announcements subscription (new user or backfill). - # In dry_run mode we still query Brevo so the counts are accurate. + # Brevo is the source of truth for subscription status. Queried whenever a + # subscription needs creating or its contact still needs a MDB_SUBSCRIPTION_ID + # write-back. Also gates the write-back so we never re-add an UNSUBSCRIBED + # user to the list. In dry_run mode we still query Brevo for accurate counts. brevo_status: BrevoSubscriptionStatus | None = None try: brevo_status = get_contact_subscription_status( @@ -295,10 +335,21 @@ def migrate_firebase_users( if announcements_active else "announcements_disabled" ) + # Write MDB_SUBSCRIPTION_ID back only onto contacts Brevo confirms are + # already on the list (SUBSCRIBED). This deliberately excludes: + # - UNSUBSCRIBED: never re-add an opted-out contact to the list; + # - NOT_FOUND: the user is not a Brevo contact, so there is nothing to + # update — we do not create new contacts / grow the list here; + # - a failed check (brevo_status is None): retried on the next run. + should_sync_brevo = ( + brevo_status == BrevoSubscriptionStatus.SUBSCRIBED + and announcements_list_id is not None + ) if not dry_run: + app_user_row = existing if existing is None: - new_user = AppUser( + app_user_row = AppUser( id=user_record.uid, email=user_record.email, email_verified=user_record.email_verified, @@ -318,27 +369,52 @@ def migrate_firebase_users( brevo_status is not None and brevo_status != BrevoSubscriptionStatus.NOT_FOUND ): - new_user.is_registered_to_receive_api_announcements = ( + app_user_row.is_registered_to_receive_api_announcements = ( brevo_status == BrevoSubscriptionStatus.SUBSCRIBED ) - db_session.add(new_user) - db_session.add( - NotificationSubscription( + db_session.add(app_user_row) + results["inserted"] += 1 + + if subscription is None: + subscription = NotificationSubscription( id=generate_unique_id(), user_id=user_record.uid, notification_type_id=API_ANNOUNCEMENTS_TYPE_ID, active=announcements_active, created_at=datetime.now(timezone.utc), ) - ) + db_session.add(subscription) + results[announcements_key] += 1 + + # Write MDB_SUBSCRIPTION_ID onto the Brevo contact and stamp the + # user's brevo_synced_at only on success, so an unsynced row is + # retried. This is the only update this task makes to an existing + # app_user row. + if should_sync_brevo: + try: + add_contact_to_list( + user_record.email, + announcements_list_id, + subscription.id, + first_name=app_user_row.full_name, + organization=app_user_row.legacy_org_name, + ) + app_user_row.brevo_synced_at = datetime.now(timezone.utc) + results["brevo_synced"] += 1 + except Exception: # noqa: BLE001 + logger.warning( + "Brevo write-back failed for uid=%s", user_record.uid + ) + results["brevo_sync_failed"] += 1 + db_session.flush() - if existing is None: - results["inserted"] += 1 - results[announcements_key] += 1 else: if existing is None: results["inserted"] += 1 - results[announcements_key] += 1 + if subscription is None: + results[announcements_key] += 1 + if should_sync_brevo: + results["brevo_synced"] += 1 processed += 1 diff --git a/functions-python/tasks_executor/tests/tasks/users/test_migrate_firebase_users.py b/functions-python/tasks_executor/tests/tasks/users/test_migrate_firebase_users.py index 95a2c9180..a747cba83 100644 --- a/functions-python/tasks_executor/tests/tasks/users/test_migrate_firebase_users.py +++ b/functions-python/tasks_executor/tests/tasks/users/test_migrate_firebase_users.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import os import unittest from datetime import datetime, timezone from unittest.mock import MagicMock, patch @@ -50,6 +51,41 @@ def _added_subscription(session) -> NotificationSubscription: return _added_of_type(session, NotificationSubscription) +_DEFAULT = object() # sentinel for _make_db_session's existing_sub + + +def _make_subscription(user_id, sub_id="sub-existing", active=True): + """Build an existing api.announcements subscription row for the DB mock.""" + return NotificationSubscription( + id=sub_id, + user_id=user_id, + notification_type_id="api.announcements", + active=active, + ) + + +def _make_app_user( + uid, + email=None, + migrated_at=_DEFAULT, + brevo_synced_at=None, + full_name=None, + legacy_org_name=None, +): + """Build an existing app_user row. brevo_synced_at drives the write-back + idempotency check (None = contact not yet synced).""" + if migrated_at is _DEFAULT: + migrated_at = datetime.now(timezone.utc) + return AppUser( + id=uid, + email=email or f"{uid}@example.com", + migrated_at=migrated_at, + brevo_synced_at=brevo_synced_at, + full_name=full_name, + legacy_org_name=legacy_org_name, + ) + + def _make_auth_user( uid, email="user@example.com", email_verified=True, created_ms=1_000_000_000_000 ): @@ -65,13 +101,23 @@ def _make_auth_user( return user -def _make_db_session(existing_user=None, has_announcements_sub=True): +def _make_db_session( + existing_user=None, has_announcements_sub=True, existing_sub=_DEFAULT +): session = MagicMock() session.get.return_value = existing_user - # Controls _has_announcements_subscription(): a truthy .first() means the - # user already has an api.announcements subscription. - first_return = MagicMock() if has_announcements_sub else None - session.query.return_value.filter.return_value.first.return_value = first_return + # Controls _get_announcements_subscription(): .first() returns the existing + # subscription row (or None). A brand-new user (existing_user is None) never + # has one. For an existing user, has_announcements_sub picks whether a + # subscription row exists; pass existing_sub to override with a specific row + # (e.g. to assert the id written back to Brevo). The write-back idempotency + # state now lives on the app_user (brevo_synced_at), not the subscription. + if existing_sub is _DEFAULT: + if existing_user is not None and has_announcements_sub: + existing_sub = _make_subscription(existing_user.id) + else: + existing_sub = None + session.query.return_value.filter.return_value.first.return_value = existing_sub return session @@ -310,12 +356,15 @@ def test_new_user_not_found_gets_enabled_announcements_subscription(self): self.assertTrue(sub.active) self.assertEqual(stats["announcements_enabled"], 1) - # --- Existing users are never updated --- + # --- Existing, already-synced users are skipped --- - def test_existing_user_skipped_no_db_write_no_brevo(self): - """Existing user (any state) → skipped, no DB write, no Brevo call.""" + def test_existing_synced_user_skipped_no_db_write_no_brevo(self): + """Existing user that already has a subscription and brevo_synced_at set + → skipped entirely: no DB write, no Brevo call.""" user = _make_auth_user("uid5") - existing = AppUser(id="uid5", email="e@example.com", migrated_at=None) + existing = _make_app_user( + "uid5", email="e@example.com", brevo_synced_at=datetime.now(timezone.utc) + ) session = _make_db_session(existing) ds_client = _make_ds_client({}) @@ -419,14 +468,14 @@ def test_already_migrated_user_without_sub_backfilled_and_counted_skipped(self): self.assertEqual(stats["inserted"], 0) self.assertEqual(stats["announcements_disabled"], 1) - def test_existing_user_with_sub_no_backfill_no_brevo(self): - """Existing user that already has the subscription → no-op (no Brevo call, - no subscription created).""" + def test_existing_synced_user_with_sub_no_backfill_no_brevo(self): + """Existing user that already has the subscription and is brevo-synced → + no-op (no Brevo call, no subscription created).""" user = _make_auth_user("uid-has", email="has@example.com") - existing = AppUser( - id="uid-has", + existing = _make_app_user( + "uid-has", email="has@example.com", - migrated_at=datetime.now(timezone.utc), + brevo_synced_at=datetime.now(timezone.utc), ) session = _make_db_session(existing, has_announcements_sub=True) @@ -450,6 +499,230 @@ def test_existing_user_with_sub_no_backfill_no_brevo(self): self.assertEqual(stats["announcements_enabled"], 0) self.assertEqual(stats["announcements_disabled"], 0) + # --- Brevo contact write-back (MDB_SUBSCRIPTION_ID / brevo_synced_at) --- + + def _run_writeback( + self, + user_records, + db_session, + brevo_status, + list_id="42", + dry_run=False, + add_side_effect=None, + **kwargs, + ): + """Run the task with add_contact_to_list patched and the Brevo list id set. + + Returns (stats, mock_add_contact_to_list). + """ + ds_client = _make_ds_client({}) + env = {"BREVO_API_ANNOUNCEMENTS_LIST_ID": list_id} if list_id else {} + with ( + patch("tasks.users.migrate_firebase_users._get_firebase_app"), + patch("tasks.users.migrate_firebase_users.datastore") as mock_datastore, + patch( + "tasks.users.migrate_firebase_users._iter_users", + return_value=iter(user_records), + ), + patch(BREVO_MODULE, return_value=brevo_status), + patch( + "tasks.users.migrate_firebase_users.add_contact_to_list", + side_effect=add_side_effect, + ) as mock_add, + patch.dict(os.environ, env, clear=False), + ): + mock_datastore.Client.return_value = ds_client + stats = migrate_firebase_users( + db_session=db_session, dry_run=dry_run, **kwargs + ) + return stats, mock_add + + def test_new_subscribed_user_written_back_and_stamped(self): + """New subscribed user → subscription created, MDB_SUBSCRIPTION_ID written + once with the new subscription id, app_user.brevo_synced_at stamped.""" + user = _make_auth_user("uid-wb1", email="wb1@example.com") + session = _make_db_session() + + stats, mock_add = self._run_writeback( + [user], session, BrevoSubscriptionStatus.SUBSCRIBED + ) + + sub = _added_subscription(session) + self.assertIsNotNone(sub) + # New user has no Datastore profile here, so name/org are forwarded as None. + mock_add.assert_called_once_with( + "wb1@example.com", 42, sub.id, first_name=None, organization=None + ) + added_user = _added_app_user(session) + self.assertIsNotNone(added_user.brevo_synced_at) + self.assertEqual(stats["brevo_synced"], 1) + self.assertEqual(stats["brevo_sync_failed"], 0) + + def test_existing_unsynced_user_written_back_without_new_row(self): + """Existing user with brevo_synced_at=None → no new subscription row, + contact written back with the existing subscription id, and the existing + app_user row stamped in place.""" + user = _make_auth_user("uid-wb2", email="wb2@example.com") + existing = _make_app_user( + "uid-wb2", + email="wb2@example.com", + brevo_synced_at=None, + full_name="Jane Doe", + legacy_org_name="Acme Transit", + ) + sub = _make_subscription("uid-wb2", sub_id="sub-wb2") + session = _make_db_session(existing, existing_sub=sub) + + stats, mock_add = self._run_writeback( + [user], + session, + BrevoSubscriptionStatus.SUBSCRIBED, + only_not_migrated=False, + ) + + self.assertIsNone(_added_subscription(session)) # no new subscription row + self.assertIsNone(_added_app_user(session)) # existing row not re-inserted + # Full name → FIRSTNAME, legacy org → ORGANIZATION are forwarded to Brevo. + mock_add.assert_called_once_with( + "wb2@example.com", + 42, + "sub-wb2", + first_name="Jane Doe", + organization="Acme Transit", + ) + self.assertIsNotNone(existing.brevo_synced_at) # stamped in place + self.assertEqual(stats["brevo_synced"], 1) + + def test_already_synced_user_is_skipped(self): + """Existing user already synced (app_user.brevo_synced_at set) with a + subscription → skipped entirely: no write-back, no new subscription row.""" + user = _make_auth_user("uid-wb3", email="wb3@example.com") + existing = _make_app_user( + "uid-wb3", + email="wb3@example.com", + brevo_synced_at=datetime.now(timezone.utc), + ) + session = _make_db_session(existing, has_announcements_sub=True) + + stats, mock_add = self._run_writeback( + [user], + session, + BrevoSubscriptionStatus.SUBSCRIBED, + only_not_migrated=False, + ) + + mock_add.assert_not_called() + self.assertIsNone(_added_subscription(session)) + self.assertEqual(stats["brevo_synced"], 0) + + def test_unsubscribed_existing_user_not_written_back(self): + """Existing unsynced user whose contact is UNSUBSCRIBED on Brevo → never + re-added to the list; app_user.brevo_synced_at stays None.""" + user = _make_auth_user("uid-wb4", email="wb4@example.com") + existing = _make_app_user( + "uid-wb4", email="wb4@example.com", brevo_synced_at=None + ) + sub = _make_subscription("uid-wb4", sub_id="sub-wb4") + session = _make_db_session(existing, existing_sub=sub) + + stats, mock_add = self._run_writeback( + [user], + session, + BrevoSubscriptionStatus.UNSUBSCRIBED, + only_not_migrated=False, + ) + + mock_add.assert_not_called() + self.assertIsNone(existing.brevo_synced_at) + self.assertEqual(stats["brevo_synced"], 0) + + def test_writeback_failure_is_counted_and_not_stamped(self): + """add_contact_to_list raising → counted as brevo_sync_failed, + app_user.brevo_synced_at left None (retried next run), task does not abort.""" + user = _make_auth_user("uid-wb5", email="wb5@example.com") + existing = _make_app_user( + "uid-wb5", email="wb5@example.com", brevo_synced_at=None + ) + sub = _make_subscription("uid-wb5", sub_id="sub-wb5") + session = _make_db_session(existing, existing_sub=sub) + + stats, mock_add = self._run_writeback( + [user], + session, + BrevoSubscriptionStatus.SUBSCRIBED, + add_side_effect=Exception("Brevo down"), + only_not_migrated=False, + ) + + mock_add.assert_called_once() + self.assertIsNone(existing.brevo_synced_at) + self.assertEqual(stats["brevo_synced"], 0) + self.assertEqual(stats["brevo_sync_failed"], 1) + + def test_dry_run_does_not_write_back_but_counts(self): + """dry_run=True → no add_contact_to_list call and no stamping, but the + would-be sync is counted.""" + user = _make_auth_user("uid-wb6", email="wb6@example.com") + session = _make_db_session() + + stats, mock_add = self._run_writeback( + [user], session, BrevoSubscriptionStatus.SUBSCRIBED, dry_run=True + ) + + mock_add.assert_not_called() + session.add.assert_not_called() + self.assertEqual(stats["brevo_synced"], 1) + + def test_missing_list_id_skips_write_back(self): + """No BREVO_API_ANNOUNCEMENTS_LIST_ID configured → write-back skipped, + subscription still created.""" + user = _make_auth_user("uid-wb7", email="wb7@example.com") + session = _make_db_session() + + stats, mock_add = self._run_writeback( + [user], session, BrevoSubscriptionStatus.SUBSCRIBED, list_id=None + ) + + mock_add.assert_not_called() + self.assertEqual(stats["brevo_synced"], 0) + self.assertIsNotNone(_added_subscription(session)) + + def test_write_back_is_idempotent_across_runs(self): + """Second run over an already-synced user is a no-op: the write happens + once, then app_user.brevo_synced_at short-circuits subsequent runs.""" + user = _make_auth_user("uid-wb8", email="wb8@example.com") + existing = _make_app_user( + "uid-wb8", email="wb8@example.com", brevo_synced_at=None + ) + sub = _make_subscription("uid-wb8", sub_id="sub-wb8") + session = _make_db_session(existing, existing_sub=sub) + + ds_client = _make_ds_client({}) + with ( + patch("tasks.users.migrate_firebase_users._get_firebase_app"), + patch("tasks.users.migrate_firebase_users.datastore") as mock_datastore, + patch( + "tasks.users.migrate_firebase_users._iter_users", + side_effect=lambda user_ids: iter([user]), + ), + patch(BREVO_MODULE, return_value=BrevoSubscriptionStatus.SUBSCRIBED), + patch("tasks.users.migrate_firebase_users.add_contact_to_list") as mock_add, + patch.dict( + os.environ, {"BREVO_API_ANNOUNCEMENTS_LIST_ID": "42"}, clear=False + ), + ): + mock_datastore.Client.return_value = ds_client + stats1 = migrate_firebase_users( + db_session=session, dry_run=False, only_not_migrated=False + ) + stats2 = migrate_firebase_users( + db_session=session, dry_run=False, only_not_migrated=False + ) + + self.assertEqual(mock_add.call_count, 1) + self.assertEqual(stats1["brevo_synced"], 1) + self.assertEqual(stats2["brevo_synced"], 0) + # --- Brevo failure on new user --- def test_brevo_failure_on_new_user_does_not_abort(self): diff --git a/liquibase/changelog_user.xml b/liquibase/changelog_user.xml index 885a4be50..dadd1ae01 100644 --- a/liquibase/changelog_user.xml +++ b/liquibase/changelog_user.xml @@ -27,4 +27,7 @@ + + diff --git a/liquibase/changes_user/feat_1733.sql b/liquibase/changes_user/feat_1733.sql new file mode 100644 index 000000000..a7f2e5edf --- /dev/null +++ b/liquibase/changes_user/feat_1733.sql @@ -0,0 +1,11 @@ +-- Issue #1733: Add brevo_synced_at to app_user. +-- Tracks when the user's Brevo contact was last written with its +-- MDB_SUBSCRIPTION_ID (the id of the user's api.announcements +-- notification_subscription) by the migrate_firebase_users task. +-- NULL means the Brevo contact has not been synced yet (MDB_SUBSCRIPTION_ID is +-- assumed unset), so the next task run will (re)write it. api.announcements is +-- the only Brevo-delivered notification type and a Brevo contact maps 1:1 to a +-- user, so this lives on app_user rather than on notification_subscription. +-- Idempotent and reusable to detect/repair contacts with a missing Brevo id. +ALTER TABLE app_user + ADD COLUMN IF NOT EXISTS brevo_synced_at TIMESTAMPTZ;