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;