diff --git a/templates/twitch/badge_list.html b/templates/twitch/badge_list.html
index a91b3cc..ea15c09 100644
--- a/templates/twitch/badge_list.html
+++ b/templates/twitch/badge_list.html
@@ -4,11 +4,11 @@
Chat Badges
{% endblock title %}
{% block content %}
- {{ badge_sets.count }} Twitch Chat Badges
+ {{ badge_data|length }} Twitch Chat Badges
- {% if badge_sets %}
+ {% if badge_data %}
{% for data in badge_data %}
{{ data.set.set_id }}
diff --git a/twitch/migrations/0017_dropbenefit_twitch_drop_distrib_70d961_idx.py b/twitch/migrations/0017_dropbenefit_twitch_drop_distrib_70d961_idx.py
new file mode 100644
index 0000000..ae8e9b2
--- /dev/null
+++ b/twitch/migrations/0017_dropbenefit_twitch_drop_distrib_70d961_idx.py
@@ -0,0 +1,22 @@
+# Generated by Django 6.0.4 on 2026-04-10 23:02
+
+from django.db import migrations
+from django.db import models
+
+
+class Migration(migrations.Migration):
+ """Add an index on DropBenefit for distribution_type and name."""
+
+ dependencies = [
+ ("twitch", "0016_mark_all_drops_fully_imported"),
+ ]
+
+ operations = [
+ migrations.AddIndex(
+ model_name="dropbenefit",
+ index=models.Index(
+ fields=["distribution_type", "name"],
+ name="twitch_drop_distrib_70d961_idx",
+ ),
+ ),
+ ]
diff --git a/twitch/models.py b/twitch/models.py
index 3f0f77f..73daa0e 100644
--- a/twitch/models.py
+++ b/twitch/models.py
@@ -8,6 +8,7 @@ import auto_prefetch
from django.conf import settings
from django.contrib.postgres.indexes import GinIndex
from django.db import models
+from django.db.models import Prefetch
from django.urls import reverse
from django.utils import timezone
from django.utils.html import format_html
@@ -17,6 +18,7 @@ from twitch.utils import normalize_twitch_box_art_url
if TYPE_CHECKING:
import datetime
+ from django.db.models import QuerySet
logger: logging.Logger = logging.getLogger("ttvdrops")
@@ -882,6 +884,8 @@ class DropBenefit(auto_prefetch.Model):
models.Index(fields=["is_ios_available"]),
models.Index(fields=["added_at"]),
models.Index(fields=["updated_at"]),
+ # Composite index for badge award lookups (distribution_type="BADGE", name__in=titles)
+ models.Index(fields=["distribution_type", "name"]),
]
def __str__(self) -> str:
@@ -1261,6 +1265,20 @@ class ChatBadgeSet(auto_prefetch.Model):
"""Return a string representation of the badge set."""
return self.set_id
+ @classmethod
+ def for_list_view(cls) -> QuerySet[ChatBadgeSet]:
+ """Return all badge sets with badges prefetched, ordered by set_id."""
+ return cls.objects.prefetch_related(
+ Prefetch("badges", queryset=ChatBadge.objects.order_by("badge_id")),
+ ).order_by("set_id")
+
+ @classmethod
+ def for_detail_view(cls, set_id: str) -> ChatBadgeSet:
+ """Return a single badge set with badges prefetched."""
+ return cls.objects.prefetch_related(
+ Prefetch("badges", queryset=ChatBadge.objects.order_by("badge_id")),
+ ).get(set_id=set_id)
+
# MARK: ChatBadge
class ChatBadge(auto_prefetch.Model):
@@ -1355,3 +1373,43 @@ class ChatBadge(auto_prefetch.Model):
def __str__(self) -> str:
"""Return a string representation of the badge."""
return f"{self.badge_set.set_id}/{self.badge_id}: {self.title}"
+
+ @staticmethod
+ def award_campaigns_by_title(titles: list[str]) -> dict[str, list[DropCampaign]]:
+ """Batch-fetch DropCampaigns that award badges matching the given titles.
+
+ Avoids N+1 queries: one query traverses DropBenefit → TimeBasedDrop → DropCampaign
+ to get (benefit_name, campaign_pk) pairs, then one more query fetches the campaigns.
+
+ Returns:
+ Mapping of badge title to a list of DropCampaigns awarding it.
+ Titles with no matching campaigns are omitted.
+ """
+ if not titles:
+ return {}
+
+ # Single JOIN query: (benefit_name, campaign_pk) via the M2M chain
+ # DropBenefit -> DropBenefitEdge -> TimeBasedDrop -> DropCampaign (FK column)
+ pairs: list[tuple[str, int | None]] = list(
+ DropBenefit.objects
+ .filter(distribution_type="BADGE", name__in=titles)
+ .values_list("name", "drops__campaign_id")
+ .distinct(),
+ )
+
+ title_to_campaign_pks: dict[str, set[int]] = {}
+ for name, campaign_pk in pairs:
+ if campaign_pk is not None:
+ title_to_campaign_pks.setdefault(name, set()).add(campaign_pk)
+
+ if not title_to_campaign_pks:
+ return {}
+
+ all_campaign_pks = {pk for pks in title_to_campaign_pks.values() for pk in pks}
+ campaigns_by_pk: dict[int, DropCampaign] = {
+ c.pk: c for c in DropCampaign.objects.filter(pk__in=all_campaign_pks)
+ }
+ return {
+ title: [campaigns_by_pk[pk] for pk in sorted(pks) if pk in campaigns_by_pk]
+ for title, pks in title_to_campaign_pks.items()
+ }
diff --git a/twitch/tests/test_views.py b/twitch/tests/test_views.py
index c393247..2297d33 100644
--- a/twitch/tests/test_views.py
+++ b/twitch/tests/test_views.py
@@ -2761,3 +2761,347 @@ class TestImageObjectStructuredData:
schema: dict[str, Any] = json.loads(response.context["schema_data"])
assert schema["image"]["creditText"] == "Real Campaign Publisher"
assert schema["organizer"]["name"] == "Real Campaign Publisher"
+
+
+@pytest.mark.django_db
+class TestBadgeListView:
+ """Tests for the badge_list_view function."""
+
+ def test_badge_list_returns_200(self, client: Client) -> None:
+ """Badge list view renders successfully with no badge sets."""
+ response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:badge_list"))
+ assert response.status_code == 200
+
+ def test_badge_list_context_has_badge_data(self, client: Client) -> None:
+ """Badge list view passes badge_data list (not badge_sets queryset) to template."""
+ badge_set: ChatBadgeSet = ChatBadgeSet.objects.create(set_id="test_vip")
+ ChatBadge.objects.create(
+ badge_set=badge_set,
+ badge_id="1",
+ image_url_1x="https://example.com/1x.png",
+ image_url_2x="https://example.com/2x.png",
+ image_url_4x="https://example.com/4x.png",
+ title="VIP",
+ description="VIP badge",
+ )
+
+ response: _MonkeyPatchedWSGIResponse = client.get(reverse("twitch:badge_list"))
+ context: ContextList | dict[str, Any] = response.context # type: ignore[assignment]
+ if isinstance(context, list):
+ context = context[-1]
+
+ assert "badge_data" in context
+ assert len(context["badge_data"]) == 1
+ assert context["badge_data"][0]["set"].set_id == "test_vip"
+ assert len(context["badge_data"][0]["badges"]) == 1
+ assert "badge_sets" not in context
+
+ def test_badge_list_query_count_stays_flat(self, client: Client) -> None:
+ """badge_list_view should not issue N+1 queries as badge set count grows."""
+ for i in range(3):
+ bs: ChatBadgeSet = ChatBadgeSet.objects.create(set_id=f"set_flat_{i}")
+ for j in range(4):
+ ChatBadge.objects.create(
+ badge_set=bs,
+ badge_id=str(j),
+ image_url_1x="https://example.com/1x.png",
+ image_url_2x="https://example.com/2x.png",
+ image_url_4x="https://example.com/4x.png",
+ title=f"Badge {i}-{j}",
+ description="desc",
+ )
+
+ def _count_selects() -> int:
+ with CaptureQueriesContext(connection) as ctx:
+ resp: _MonkeyPatchedWSGIResponse = client.get(
+ reverse("twitch:badge_list"),
+ )
+ assert resp.status_code == 200
+ return sum(
+ 1
+ for q in ctx.captured_queries
+ if q["sql"].lstrip().upper().startswith("SELECT")
+ )
+
+ baseline: int = _count_selects()
+
+ # Add 10 more badge sets with badges
+ for i in range(3, 13):
+ bs = ChatBadgeSet.objects.create(set_id=f"set_flat_{i}")
+ for j in range(4):
+ ChatBadge.objects.create(
+ badge_set=bs,
+ badge_id=str(j),
+ image_url_1x="https://example.com/1x.png",
+ image_url_2x="https://example.com/2x.png",
+ image_url_4x="https://example.com/4x.png",
+ title=f"Badge {i}-{j}",
+ description="desc",
+ )
+
+ scaled: int = _count_selects()
+ assert scaled <= baseline + 1, (
+ f"badge_list_view SELECT count grew with data; possible N+1. "
+ f"baseline={baseline}, scaled={scaled}"
+ )
+
+
+@pytest.mark.django_db
+class TestBadgeSetDetailView:
+ """Tests for the badge_set_detail_view function."""
+
+ @pytest.fixture
+ def badge_set_with_badges(self) -> dict[str, Any]:
+ """Create a badge set with numeric badge IDs and a campaign awarding one badge.
+
+ Returns:
+ Dict with badge_set, badge1-3, campaign, and benefit instances.
+ """
+ org: Organization = Organization.objects.create(
+ twitch_id="org_badge_test",
+ name="Badge Test Org",
+ )
+ game: Game = Game.objects.create(
+ twitch_id="game_badge_test",
+ name="badge_test_game",
+ display_name="Badge Test Game",
+ )
+ game.owners.add(org)
+
+ badge_set: ChatBadgeSet = ChatBadgeSet.objects.create(set_id="drops")
+ badge1: ChatBadge = ChatBadge.objects.create(
+ badge_set=badge_set,
+ badge_id="1",
+ image_url_1x="https://example.com/1x.png",
+ image_url_2x="https://example.com/2x.png",
+ image_url_4x="https://example.com/4x.png",
+ title="Drop 1",
+ description="First drop badge",
+ )
+ badge2: ChatBadge = ChatBadge.objects.create(
+ badge_set=badge_set,
+ badge_id="10",
+ image_url_1x="https://example.com/1x.png",
+ image_url_2x="https://example.com/2x.png",
+ image_url_4x="https://example.com/4x.png",
+ title="Drop 10",
+ description="Tenth drop badge",
+ )
+ badge3: ChatBadge = ChatBadge.objects.create(
+ badge_set=badge_set,
+ badge_id="2",
+ image_url_1x="https://example.com/1x.png",
+ image_url_2x="https://example.com/2x.png",
+ image_url_4x="https://example.com/4x.png",
+ title="Drop 2",
+ description="Second drop badge",
+ )
+
+ campaign: DropCampaign = DropCampaign.objects.create(
+ twitch_id="badge_test_campaign",
+ name="Badge Test Campaign",
+ game=game,
+ operation_names=["DropCampaignDetails"],
+ )
+ drop: TimeBasedDrop = TimeBasedDrop.objects.create(
+ twitch_id="badge_test_drop",
+ name="Badge Test Drop",
+ campaign=campaign,
+ )
+ benefit: DropBenefit = DropBenefit.objects.create(
+ twitch_id="badge_test_benefit",
+ name="Drop 1",
+ distribution_type="BADGE",
+ )
+ drop.benefits.add(benefit)
+
+ return {
+ "badge_set": badge_set,
+ "badge1": badge1,
+ "badge2": badge2,
+ "badge3": badge3,
+ "campaign": campaign,
+ "benefit": benefit,
+ }
+
+ def test_badge_set_detail_returns_200(
+ self,
+ client: Client,
+ badge_set_with_badges: dict[str, Any],
+ ) -> None:
+ """Badge set detail view renders successfully."""
+ set_id: str = badge_set_with_badges["badge_set"].set_id
+ url: str = reverse("twitch:badge_set_detail", args=[set_id])
+ response: _MonkeyPatchedWSGIResponse = client.get(url)
+ assert response.status_code == 200
+
+ def test_badge_set_detail_404_for_missing_set(self, client: Client) -> None:
+ """Badge set detail view returns 404 for unknown set_id."""
+ url: str = reverse("twitch:badge_set_detail", args=["nonexistent"])
+ response: _MonkeyPatchedWSGIResponse = client.get(url)
+ assert response.status_code == 404
+
+ def test_badges_sorted_numerically(
+ self,
+ client: Client,
+ badge_set_with_badges: dict[str, Any],
+ ) -> None:
+ """Numeric badge_ids should be sorted as integers (1, 2, 10) not strings (1, 10, 2)."""
+ set_id: str = badge_set_with_badges["badge_set"].set_id
+ url: str = reverse("twitch:badge_set_detail", args=[set_id])
+ response: _MonkeyPatchedWSGIResponse = client.get(url)
+ context: ContextList | dict[str, Any] = response.context # type: ignore[assignment]
+ if isinstance(context, list):
+ context = context[-1]
+
+ badge_ids: list[str] = [b.badge_id for b in context["badges"]]
+ assert badge_ids == ["1", "2", "10"], (
+ f"Expected numeric sort order [1, 2, 10], got {badge_ids}"
+ )
+
+ def test_award_campaigns_attached_to_badges(
+ self,
+ client: Client,
+ badge_set_with_badges: dict[str, Any],
+ ) -> None:
+ """Badges with matching BADGE benefits should have award_campaigns populated."""
+ set_id: str = badge_set_with_badges["badge_set"].set_id
+ url: str = reverse("twitch:badge_set_detail", args=[set_id])
+ response: _MonkeyPatchedWSGIResponse = client.get(url)
+ context: ContextList | dict[str, Any] = response.context # type: ignore[assignment]
+ if isinstance(context, list):
+ context = context[-1]
+
+ badges: list[ChatBadge] = list(context["badges"])
+ badge_titled_drop1: ChatBadge = next(b for b in badges if b.title == "Drop 1")
+ badge_titled_drop2: ChatBadge = next(b for b in badges if b.title == "Drop 2")
+
+ assert len(badge_titled_drop1.award_campaigns) == 1 # pyright: ignore[reportAttributeAccessIssue]
+ assert badge_titled_drop1.award_campaigns[0].twitch_id == "badge_test_campaign" # pyright: ignore[reportAttributeAccessIssue]
+ assert len(badge_titled_drop2.award_campaigns) == 0 # pyright: ignore[reportAttributeAccessIssue]
+
+ def test_badge_set_detail_avoids_n_plus_one(
+ self,
+ client: Client,
+ ) -> None:
+ """badge_set_detail_view should not issue per-badge queries for award campaigns."""
+ org: Organization = Organization.objects.create(
+ twitch_id="org_n1_badge",
+ name="N+1 Badge Org",
+ )
+ game: Game = Game.objects.create(
+ twitch_id="game_n1_badge",
+ name="game_n1_badge",
+ display_name="N+1 Badge Game",
+ )
+ game.owners.add(org)
+
+ badge_set: ChatBadgeSet = ChatBadgeSet.objects.create(set_id="n1_test")
+
+ def _make_badge_and_campaign(idx: int) -> None:
+ badge: ChatBadge = ChatBadge.objects.create(
+ badge_set=badge_set,
+ badge_id=str(idx),
+ image_url_1x="https://example.com/1x.png",
+ image_url_2x="https://example.com/2x.png",
+ image_url_4x="https://example.com/4x.png",
+ title=f"N1 Badge {idx}",
+ description="desc",
+ )
+ campaign: DropCampaign = DropCampaign.objects.create(
+ twitch_id=f"n1_campaign_{idx}",
+ name=f"N+1 Campaign {idx}",
+ game=game,
+ operation_names=["DropCampaignDetails"],
+ )
+ drop: TimeBasedDrop = TimeBasedDrop.objects.create(
+ twitch_id=f"n1_drop_{idx}",
+ name=f"N+1 Drop {idx}",
+ campaign=campaign,
+ )
+ benefit: DropBenefit = DropBenefit.objects.create(
+ twitch_id=f"n1_benefit_{idx}",
+ name=badge.title,
+ distribution_type="BADGE",
+ )
+ drop.benefits.add(benefit)
+
+ for i in range(3):
+ _make_badge_and_campaign(i)
+
+ url: str = reverse("twitch:badge_set_detail", args=[badge_set.set_id])
+
+ def _count_selects() -> int:
+ with CaptureQueriesContext(connection) as ctx:
+ resp: _MonkeyPatchedWSGIResponse = client.get(url)
+ assert resp.status_code == 200
+ return sum(
+ 1
+ for q in ctx.captured_queries
+ if q["sql"].lstrip().upper().startswith("SELECT")
+ )
+
+ baseline: int = _count_selects()
+
+ # Add 10 more badges, each with their own campaigns
+ for i in range(3, 13):
+ _make_badge_and_campaign(i)
+
+ scaled: int = _count_selects()
+ assert scaled <= baseline + 1, (
+ f"badge_set_detail_view SELECT count grew with badge count; possible N+1. "
+ f"baseline={baseline}, scaled={scaled}"
+ )
+
+ def test_drop_benefit_index_used_for_badge_award_lookup(self) -> None:
+ """DropBenefit queries filtering by distribution_type+name should use indexes."""
+ org: Organization = Organization.objects.create(
+ twitch_id="org_benefit_idx",
+ name="Benefit Index Org",
+ )
+ game: Game = Game.objects.create(
+ twitch_id="game_benefit_idx",
+ name="game_benefit_idx",
+ display_name="Benefit Index Game",
+ )
+ game.owners.add(org)
+
+ # Create enough non-BADGE benefits so the planner has reason to use an index
+ for i in range(300):
+ DropBenefit.objects.create(
+ twitch_id=f"non_badge_{i}",
+ name=f"Emote {i}",
+ distribution_type="EMOTE",
+ )
+
+ badge_titles: list[str] = []
+ for i in range(5):
+ DropBenefit.objects.create(
+ twitch_id=f"badge_benefit_idx_{i}",
+ name=f"Badge Title {i}",
+ distribution_type="BADGE",
+ )
+ badge_titles.append(f"Badge Title {i}")
+
+ qs = DropBenefit.objects.filter(
+ distribution_type="BADGE",
+ name__in=badge_titles,
+ )
+ plan: str = qs.explain()
+
+ if connection.vendor == "sqlite":
+ uses_index: bool = "USING INDEX" in plan.upper()
+ elif connection.vendor == "postgresql":
+ uses_index = (
+ "INDEX SCAN" in plan.upper()
+ or "BITMAP INDEX SCAN" in plan.upper()
+ or "INDEX ONLY SCAN" in plan.upper()
+ )
+ else:
+ pytest.skip(
+ f"Unsupported DB vendor for index-plan assertion: {connection.vendor}",
+ )
+
+ assert uses_index, (
+ f"DropBenefit query on (distribution_type, name) did not use an index.\n{plan}"
+ )
diff --git a/twitch/views.py b/twitch/views.py
index f017dfd..2e9be5b 100644
--- a/twitch/views.py
+++ b/twitch/views.py
@@ -14,11 +14,9 @@ from django.core.paginator import EmptyPage
from django.core.paginator import Page
from django.core.paginator import PageNotAnInteger
from django.core.paginator import Paginator
-from django.db.models import Case
from django.db.models import Count
from django.db.models import Prefetch
from django.db.models import Q
-from django.db.models import When
from django.db.models.query import QuerySet
from django.http import Http404
from django.http import HttpResponse
@@ -1585,22 +1583,12 @@ def badge_list_view(request: HttpRequest) -> HttpResponse:
Returns:
HttpResponse: The rendered badge list page.
"""
- badge_sets: QuerySet[ChatBadgeSet] = (
- ChatBadgeSet.objects
- .all()
- .prefetch_related(
- Prefetch("badges", queryset=ChatBadge.objects.order_by("badge_id")),
- )
- .order_by("set_id")
- )
-
- # Group badges by set for easier display
badge_data: list[dict[str, Any]] = [
{
"set": badge_set,
"badges": list(badge_set.badges.all()), # pyright: ignore[reportAttributeAccessIssue]
}
- for badge_set in badge_sets
+ for badge_set in ChatBadgeSet.for_list_view()
]
# CollectionPage schema for badges list
@@ -1618,7 +1606,6 @@ def badge_list_view(request: HttpRequest) -> HttpResponse:
seo_meta={"schema_data": collection_schema},
)
context: dict[str, Any] = {
- "badge_sets": badge_sets,
"badge_data": badge_data,
**seo_context,
}
@@ -1641,52 +1628,30 @@ def badge_set_detail_view(request: HttpRequest, set_id: str) -> HttpResponse:
Http404: If the badge set is not found.
"""
try:
- badge_set: ChatBadgeSet = ChatBadgeSet.objects.prefetch_related(
- Prefetch("badges", queryset=ChatBadge.objects.order_by("badge_id")),
- ).get(set_id=set_id)
+ badge_set: ChatBadgeSet = ChatBadgeSet.for_detail_view(set_id)
except ChatBadgeSet.DoesNotExist as exc:
msg = "No badge set found matching the query"
raise Http404(msg) from exc
- def get_sorted_badges(badge_set: ChatBadgeSet) -> QuerySet[ChatBadge]:
- badges = badge_set.badges.all() # pyright: ignore[reportAttributeAccessIssue]
+ # Sort badges treating pure-numeric badge_ids as integers, strings alphabetically after
+ badges: list[ChatBadge] = sorted(
+ badge_set.badges.all(), # pyright: ignore[reportAttributeAccessIssue]
+ key=lambda b: (0, int(b.badge_id)) if b.badge_id.isdigit() else (1, b.badge_id),
+ )
- def sort_badges(badge: ChatBadge) -> tuple:
- """Sort badges by badge_id, treating numeric IDs as integers.
-
- Args:
- badge: The ChatBadge to sort.
-
- Returns:
- A tuple used for sorting, where numeric badge_ids are sorted as integers.
- """
- try:
- return (int(badge.badge_id),)
- except ValueError:
- return (badge.badge_id,)
-
- sorted_badges: list[ChatBadge] = sorted(badges, key=sort_badges)
- badge_ids: list[int] = [badge.pk for badge in sorted_badges]
- preserved_order = Case(
- *[When(pk=pk, then=pos) for pos, pk in enumerate(badge_ids)],
- )
- return ChatBadge.objects.filter(pk__in=badge_ids).order_by(preserved_order)
-
- badges: QuerySet[ChatBadge, ChatBadge] = get_sorted_badges(badge_set)
-
- # Attach award_campaigns attribute to each badge for template use
+ # Batch-fetch award campaigns for all badge titles (2 queries regardless of badge count)
+ award_map: dict[str, list[DropCampaign]] = ChatBadge.award_campaigns_by_title(
+ [b.title for b in badges],
+ )
for badge in badges:
- benefits: QuerySet[DropBenefit, DropBenefit] = DropBenefit.objects.filter(
- distribution_type="BADGE",
- name=badge.title,
- )
- campaigns: QuerySet[DropCampaign, DropCampaign] = DropCampaign.objects.filter(
- time_based_drops__benefits__in=benefits,
- ).distinct()
- badge.award_campaigns = list(campaigns) # pyright: ignore[reportAttributeAccessIssue]
+ badge.award_campaigns = award_map.get(badge.title, []) # pyright: ignore[reportAttributeAccessIssue]
badge_set_name: str = badge_set.set_id
- badge_set_description: str = f"Twitch chat badge set {badge_set_name} with {len(badges)} badge{'s' if len(badges) != 1 else ''} awarded through drop campaigns."
+ badge_count: int = len(badges)
+ badge_set_description: str = (
+ f"Twitch chat badge set {badge_set_name} with {badge_count} "
+ f"badge{'s' if badge_count != 1 else ''} awarded through drop campaigns."
+ )
badge_schema: dict[str, Any] = {
"@context": "https://schema.org",