fix: scope /u/notifications + navbar badge to the current shop

Operator users may own multiple custom-domain shops (e.g. shop.unturf.com
and media.unturf.com). Notifications carried a shop_id at insert time
already (lib/notifications.py, views/offer.py:_drop_offer_notification),
but the listing + the navbar unread count queried by user_id only, so
each shop's site surfaced every other shop's offers / sales / auctions.

Both count_unread_notifications and get_notifications_for_user now take
an optional `shop` kwarg. When supplied, they filter to that shop_id
plus shop_id IS NULL (so account-level rows — logins, etc. — still
follow the user regardless of which shop's site they're on). The
notifications view passes request.shop; the navbar badge does too via
add_unread_notification_count.

Tests: TestNotifications gains test_notifications_isolated_per_shop —
two shops owned by the same user, three rows (one per shop + one
shop-less); shop_a context shows shop_a + global, shop_b context shows
shop_b + global, no-shop call still returns all three. 4 in class pass.
This commit is contained in:
russell@unturf.com 2026-05-15 09:25:17 -04:00
parent 1075004419
commit 4cbd7e4b20
No known key found for this signature in database
4 changed files with 119 additions and 9 deletions

View file

@ -176,24 +176,47 @@ def get_notification_by_id(dbsession, notification_id):
return get_object_by_id(dbsession, notification_id, MpsNotification)
def count_unread_notifications(dbsession, user):
"""Cheap unread-count for the profile badge — no rows fetched."""
def _scope_to_shop(q, shop):
"""Restrict a MpsNotification query to a single shop's context.
``shop=None`` → no shop filter (caller didn't pass one — show all).
``shop`` set → rows tagged with that shop's id, plus shop-less rows
(account-level notifications, e.g. logins, that should follow the
user regardless of which shop's site they're browsing).
Offers / sales / auctions all carry a shop_id at insert time
(see lib/notifications.py and views/offer.py:_drop_offer_notification),
so scoping by shop hides cross-shop noise — important on operator
deployments where one user owns multiple custom-domain shops."""
from sqlalchemy import or_
if shop is None:
return q
return q.filter(
or_(MpsNotification.shop_id == shop.id, MpsNotification.shop_id.is_(None))
)
def count_unread_notifications(dbsession, user, shop=None):
"""Cheap unread-count for the profile badge — no rows fetched.
Scoped to ``shop`` when supplied (matches what the user can see on
/u/notifications on that same shop's site)."""
if user is None:
return 0
return (
q = (
dbsession.query(MpsNotification)
.filter(
MpsNotification.user_id == user.id,
MpsNotification.read == False, # noqa: E712 — SQLAlchemy idiom
)
.count()
)
return _scope_to_shop(q, shop).count()
def get_notifications_for_user(
dbsession, user, limit=50, offset=0, unread_only=False,
dbsession, user, limit=50, offset=0, unread_only=False, shop=None,
):
"""Return the user's notifications, newest first."""
"""Return the user's notifications, newest first.
Pass ``shop`` to scope to that shop's context (current site)."""
q = (
dbsession.query(MpsNotification)
.filter(MpsNotification.user_id == user.id)
@ -201,4 +224,5 @@ def get_notifications_for_user(
)
if unread_only:
q = q.filter(MpsNotification.read == False) # noqa: E712
q = _scope_to_shop(q, shop)
return q.limit(limit).offset(offset).all()

View file

@ -526,12 +526,17 @@ def includeme(config):
def add_unread_notification_count(request):
"""In-app unread-notification count for the navbar badge.
Returns 0 for anonymous users."""
Returns 0 for anonymous users. Scoped to ``request.shop`` so the
badge on each shop's site only counts notifications from that
shop (cross-shop noise is hidden on operator deployments that
own multiple custom-domain shops)."""
from .models.notification import count_unread_notifications
if request.user is None:
return 0
return count_unread_notifications(request.dbsession, request.user)
return count_unread_notifications(
request.dbsession, request.user, shop=request.shop,
)
config.add_request_method(
add_unread_notification_count,

View file

@ -7624,6 +7624,84 @@ class TestNotifications(_AuthenticatedBase):
)
self.assertEqual(unread_after, 0)
def test_notifications_isolated_per_shop(self):
"""Cross-shop bleed regression: when a user owns multiple shops,
/u/notifications on shop A must NOT show shop B's rows. Rows with
shop_id=NULL (account-level) are still surfaced everywhere."""
from ..models.notification import (
MpsNotification,
NOTIFICATION_KIND_OFFER_RECEIVED,
count_unread_notifications,
get_notifications_for_user,
)
from ..models.shop import get_shop_by_id
from ..models.user import get_user_by_email
from ..models.shop import Shop
from ..models.user_shop import UserShop
shop_a = self._create_shop_helper(user_creds=self.user1_creds)
shop_a_id = shop_a.uuid_str
# Spin up a second shop owned by the same user via ORM — the
# _create_shop_helper UI flow is heavy and unnecessary for this
# test (we only need a second shop_id to scope against).
shop_b = Shop(
name="Second Shop", phone_number="+10000000000",
billing_address="n/a", description="n/a",
)
seller = get_user_by_email(self.dbsession, self.user1_creds[0])
self.dbsession.add(shop_b)
self.dbsession.flush()
self.dbsession.add(UserShop(user=seller, shop=shop_b, role_id=0))
self.dbsession.flush()
shop_b_id = shop_b.uuid_str
seller = get_user_by_email(self.dbsession, self.user1_creds[0])
n_a = MpsNotification(
user=seller, kind=NOTIFICATION_KIND_OFFER_RECEIVED,
subject="Offer on A", body="...", shop=shop_a,
link_url="/o/aaa",
)
n_b = MpsNotification(
user=seller, kind=NOTIFICATION_KIND_OFFER_RECEIVED,
subject="Offer on B", body="...", shop=shop_b,
link_url="/o/bbb",
)
n_global = MpsNotification(
user=seller, kind="account_event",
subject="Account login", body="...", shop=None,
link_url="/u/settings",
)
self.dbsession.add(n_a)
self.dbsession.add(n_b)
self.dbsession.add(n_global)
self.dbsession.flush()
transaction.commit()
seller = get_user_by_email(self.dbsession, self.user1_creds[0])
shop_a = get_shop_by_id(self.dbsession, shop_a_id)
shop_b = get_shop_by_id(self.dbsession, shop_b_id)
# Scoped to shop A → shop_a row + shop-less account row, no shop_b.
rows_a = get_notifications_for_user(self.dbsession, seller, shop=shop_a)
subjects_a = sorted(r.subject for r in rows_a)
self.assertEqual(subjects_a, ["Account login", "Offer on A"])
self.assertEqual(
count_unread_notifications(self.dbsession, seller, shop=shop_a), 2,
)
# Scoped to shop B → shop_b row + global, no shop_a.
rows_b = get_notifications_for_user(self.dbsession, seller, shop=shop_b)
subjects_b = sorted(r.subject for r in rows_b)
self.assertEqual(subjects_b, ["Account login", "Offer on B"])
# No-shop call (legacy behavior) still returns everything.
self.assertEqual(
len(get_notifications_for_user(self.dbsession, seller)),
3,
)
def test_read_notifications_remain_visible_but_faded(self):
"""Fox's call: read notifications are NOT deleted — they stay
in the list, just visually less prominent. The unread *count*

View file

@ -186,8 +186,11 @@ def user_notifications(request):
"""
from ..models.notification import get_notifications_for_user
# Scope to the current shop's site — operator users may own multiple
# custom-domain shops, and each site should only surface its own
# offers/sales/auctions plus account-level rows that aren't shop-bound.
rows = get_notifications_for_user(
request.dbsession, request.user, limit=200,
request.dbsession, request.user, limit=200, shop=request.shop,
)
import ago