[DISCO-3977] Fetch and store engagement data for ADM and Wikipedia (#1335)

* [DISCO-3977] Fetch and store engagement data for ADM and Wikipedia

* move blob name to config file

* consolidate adm and wiki filemanagers into one

Author: misaniwere

merino/configs/default.toml

@@ -303,6 +303,10 @@ gcs_bq_project = ""
 
 gcs_storage_project = ""
 
+# MERINO_ENGAGEMENT__BLOB_NAME
+# Path within the GCS bucket to the engagement data blob.
+blob_name = "suggest-merino-exports/engagement/latest.json"
+
 [default.icon]
 # Subdirectory within bucket for favicons
 favicons_root = "favicons"
@@ -574,6 +578,10 @@ cron_interval_sec = 60
 # Time between re-syncs of Remote Settings data, in seconds. Defaults to 1 hour.
 resync_interval_sec = 3600
 
+# MERINO_PROVIDERS__ADM__ENGAGEMENT_RESYNC_INTERVAL_SEC
+# Time between re-syncs of engagement data from GCS, in seconds. Defaults to 1 hour.
+engagement_resync_interval_sec = 3600
+
 # MERINO_PROVIDERS__ADM__SCORE
 # Ranking score for this provider as a floating point number. Defaults to 0.31.
 # This is intentionally set to be greater than the score in Remote Settings to fix
@@ -760,6 +768,15 @@ query_timeout_sec = 5.0
 # The ranking score for this provider as a floating point number. Defaults to 0.23.
 score = 0.23
 
+# MERINO_PROVIDERS__WIKIPEDIA__CRON_INTERVAL_SEC
+# The interval of the engagement data cron job tick (in seconds).
+# Should be more frequent than `engagement_resync_interval_sec` to allow retries.
+cron_interval_sec = 60
+
+# MERINO_PROVIDERS__WIKIPEDIA__ENGAGEMENT_RESYNC_INTERVAL_SEC
+# Time between re-syncs of Wikipedia engagement data from GCS, in seconds. Defaults to 1 hour.
+engagement_resync_interval_sec = 3600
+
 
 [default.providers.polygon]
 # MERINO_PROVIDERS__POLYGON__TYPE

merino/providers/suggest/adm/backends/protocol.py

@@ -17,6 +17,13 @@ class FormFactor(Enum):
     PHONE = 1
 
 
+class EngagementData(BaseModel):
+    """Model for engagement data file content"""
+
+    amp: dict[str, dict[str, str | int]]
+    amp_aggregated: dict[str, int]
+
+
 class SuggestionContent(BaseModel):
     """Class that holds the result from a fetch operation."""
 

merino/providers/suggest/adm/provider.py

@@ -12,10 +12,15 @@
 
 from merino.optimizers.models import EngagementMetrics, ThompsonCandidate
 from merino.optimizers.thompson import ThompsonSampler
-from merino.providers.suggest.adm.backends.protocol import FormFactor
+from merino.providers.suggest.adm.backends.protocol import EngagementData, FormFactor
+from merino.utils.gcs.engagement.filemanager import EngagementFilemanager
 from merino.utils import cron
 from merino.providers.suggest.adm.backends.protocol import AdmBackend, SuggestionContent
-from merino.providers.suggest.base import BaseProvider, BaseSuggestion, SuggestionRequest
+from merino.providers.suggest.base import (
+    BaseProvider,
+    BaseSuggestion,
+    SuggestionRequest,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -82,6 +87,11 @@ class Provider(BaseProvider):
     resync_interval_sec: float
     min_attempted_count: int
     thompson: ThompsonSampler | None = None
+    engagement_data: EngagementData
+    filemanager: EngagementFilemanager
+    engagement_resync_interval_sec: float
+    last_engagement_fetch_at: float
+    engagement_cron_task: asyncio.Task
 
     def __init__(
         self,
@@ -90,6 +100,9 @@ def __init__(
         name: str,
         resync_interval_sec: float,
         cron_interval_sec: float,
+        engagement_gcs_bucket: str,
+        engagement_blob_name: str,
+        engagement_resync_interval_sec: float,
         enabled_by_default: bool = True,
         min_attempted_count: int = 0,
         thompson: ThompsonSampler | None = None,
@@ -105,6 +118,13 @@ def __init__(
         self._enabled_by_default = enabled_by_default
         self.min_attempted_count = min_attempted_count
         self.thompson = thompson
+        self.engagement_data = EngagementData(amp={}, amp_aggregated={})
+        self.engagement_resync_interval_sec = engagement_resync_interval_sec
+        self.last_engagement_fetch_at = 0
+        self.filemanager = EngagementFilemanager(
+            gcs_bucket_path=engagement_gcs_bucket,
+            blob_name=engagement_blob_name,
+        )
         super().__init__(**kwargs)
 
     async def initialize(self) -> None:
@@ -132,15 +152,46 @@ async def initialize(self) -> None:
         # reference to it.
         self.cron_task = asyncio.create_task(cron_job())
 
+        engagement_cron_job = cron.Job(
+            name="resync_engagement_data",
+            interval=self.cron_interval_sec,
+            condition=self._should_fetch_engagement,
+            task=self._fetch_engagement_data,
+        )
+        self.engagement_cron_task = asyncio.create_task(engagement_cron_job())
+
     def _should_fetch(self) -> bool:
         """Check if it should fetch data from Remote Settings."""
         return (time.time() - self.last_fetch_at) >= self.resync_interval_sec
 
+    def _should_fetch_engagement(self) -> bool:
+        """Check if it should fetch engagement data from GCS."""
+        return (time.time() - self.last_engagement_fetch_at) >= self.engagement_resync_interval_sec
+
     async def _fetch(self) -> None:
         """Fetch suggestions, keywords, and icons from Remote Settings."""
         self.suggestion_content = await self.backend.fetch()
         self.last_fetch_at = time.time()
 
+    async def _fetch_engagement_data(self) -> None:
+        """Fetch engagement data from GCS and store it in memory.
+
+        If the fetch returns no data, `last_engagement_fetch_at` is not updated
+        so the cron job retries on the next tick.
+        """
+        try:
+            data = await self.filemanager.get_file()
+            if data is None:
+                logger.warning("Engagement data fetch returned None, will retry on next tick")
+                return
+            self.engagement_data = EngagementData.model_validate(data.model_dump())
+            self.last_engagement_fetch_at = time.time()
+        except Exception as e:
+            logger.warning(
+                "Failed to fetch engagement data from GCS",
+                extra={"error": str(e)},
+            )
+
     def hidden(self) -> bool:  # noqa: D102
         return False
 

merino/providers/suggest/manager.py

@@ -185,12 +185,14 @@ def _create_provider(provider_id: str, setting: Settings) -> BaseProvider:
             if settings.providers.adm.thompson.enabled:
                 thompson = ThompsonSampler(
                     config=ThompsonConfig(
-                        dummy_candidate=EngagementMetrics(
-                            engaged=settings.providers.adm.thompson.dummy_engaged_count,
-                            attempted=settings.providers.adm.thompson.dummy_attempted_count,
-                        )
-                        if settings.providers.adm.thompson.dummy_enabled
-                        else None,
+                        dummy_candidate=(
+                            EngagementMetrics(
+                                engaged=settings.providers.adm.thompson.dummy_engaged_count,
+                                attempted=settings.providers.adm.thompson.dummy_attempted_count,
+                            )
+                            if settings.providers.adm.thompson.dummy_enabled
+                            else None
+                        ),
                     )
                 )
             return AdmProvider(
@@ -202,6 +204,9 @@ def _create_provider(provider_id: str, setting: Settings) -> BaseProvider:
                 enabled_by_default=setting.enabled_by_default,
                 min_attempted_count=settings.providers.adm.thompson.min_attempted_count,
                 thompson=thompson,
+                engagement_gcs_bucket=settings.engagement.gcs_storage_bucket,
+                engagement_blob_name=settings.engagement.blob_name,
+                engagement_resync_interval_sec=setting.engagement_resync_interval_sec,
             )
         case ProviderType.GEOLOCATION:
             return GeolocationProvider(
@@ -236,6 +241,10 @@ def _create_provider(provider_id: str, setting: Settings) -> BaseProvider:
                 name=provider_id,
                 query_timeout_sec=setting.query_timeout_sec,
                 enabled_by_default=setting.enabled_by_default,
+                engagement_gcs_bucket=settings.engagement.gcs_storage_bucket,
+                engagement_blob_name=settings.engagement.blob_name,
+                engagement_resync_interval_sec=setting.engagement_resync_interval_sec,
+                cron_interval_sec=setting.cron_interval_sec,
             )
         case ProviderType.POLYGON:
             cache = (

merino/providers/suggest/wikipedia/backends/protocol.py

@@ -2,6 +2,14 @@
 
 from typing import Any, Protocol
 
+from pydantic import BaseModel
+
+
+class EngagementData(BaseModel):
+    """Model for Wikipedia engagement data from GCS."""
+
+    wiki_aggregated: dict[str, int]
+
 
 class WikipediaBackend(Protocol):
     """Protocol for a Wikipedia backend that this provider depends on.

merino/providers/suggest/wikipedia/provider.py

@@ -1,6 +1,8 @@
 """The provider for the dynamic Wikipedia integration."""
 
+import asyncio
 import logging
+import time
 from typing import Any, Final
 
 from pydantic import HttpUrl
@@ -13,8 +15,10 @@
     SuggestionRequest,
     Category,
 )
-from merino.providers.suggest.wikipedia.backends.protocol import WikipediaBackend
+from merino.providers.suggest.wikipedia.backends.protocol import EngagementData, WikipediaBackend
+from merino.utils.gcs.engagement.filemanager import EngagementFilemanager
 from merino.providers.suggest.wikipedia.backends.utils import get_language_code
+from merino.utils import cron
 
 # The Wikipedia icon backed by Merino's image CDN.
 # TODO: Use a better way to fetch this icon URL instead of hardcoding it here.
@@ -48,11 +52,21 @@ class Provider(BaseProvider):
     backend: WikipediaBackend
     score: float
     title_block_list: set[str]
+    engagement_data: EngagementData
+    filemanager: EngagementFilemanager
+    engagement_resync_interval_sec: float
+    cron_interval_sec: float
+    last_engagement_fetch_at: float
+    engagement_cron_task: asyncio.Task
 
     def __init__(
         self,
         backend: WikipediaBackend,
         title_block_list: set[str],
+        engagement_gcs_bucket: str,
+        engagement_blob_name: str,
+        engagement_resync_interval_sec: float,
+        cron_interval_sec: float,
         name: str = "wikipedia",
         enabled_by_default: bool = True,
         query_timeout_sec: float = settings.providers.wikipedia.query_timeout_sec,
@@ -67,11 +81,50 @@ def __init__(
         self._enabled_by_default = enabled_by_default
         self._query_timeout_sec = query_timeout_sec
         self.score = score
+        self.engagement_data = EngagementData(wiki_aggregated={})
+        self.engagement_resync_interval_sec = engagement_resync_interval_sec
+        self.cron_interval_sec = cron_interval_sec
+        self.last_engagement_fetch_at = 0
+        self.filemanager = EngagementFilemanager(
+            gcs_bucket_path=engagement_gcs_bucket,
+            blob_name=engagement_blob_name,
+        )
         super().__init__(**kwargs)
 
     async def initialize(self) -> None:
-        """Initialize Wikipedia provider."""
-        return
+        """Initialize Wikipedia provider and start engagement data cron job."""
+        engagement_cron_job = cron.Job(
+            name="resync_wikipedia_engagement_data",
+            interval=self.cron_interval_sec,
+            condition=self._should_fetch_engagement,
+            task=self._fetch_engagement_data,
+        )
+        self.engagement_cron_task = asyncio.create_task(engagement_cron_job())
+
+    def _should_fetch_engagement(self) -> bool:
+        """Check if it should fetch Wikipedia engagement data from GCS."""
+        return (time.time() - self.last_engagement_fetch_at) >= self.engagement_resync_interval_sec
+
+    async def _fetch_engagement_data(self) -> None:
+        """Fetch Wikipedia engagement data from GCS and store it in memory.
+
+        If the fetch returns no data, `last_engagement_fetch_at` is not updated
+        so the cron job retries on the next tick.
+        """
+        try:
+            data = await self.filemanager.get_file()
+            if data is None:
+                logger.warning(
+                    "Wikipedia engagement data fetch returned None, will retry on next tick"
+                )
+                return
+            self.engagement_data = EngagementData.model_validate(data.model_dump())
+            self.last_engagement_fetch_at = time.time()
+        except Exception as e:
+            logger.warning(
+                "Failed to fetch Wikipedia engagement data from GCS",
+                extra={"error": str(e)},
+            )
 
     def hidden(self) -> bool:  # noqa: D102
         """Whether this provider is hidden or not."""

merino/utils/gcs/engagement/__init__.py

@@ -0,0 +1,4 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+"""GCS engagement data utilities."""

merino/utils/gcs/engagement/filemanager.py

@@ -0,0 +1,63 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+"""Filemanager for retrieving engagement data from GCS."""
+
+import logging
+from json import JSONDecodeError
+
+import orjson
+from gcloud.aio.storage import Blob, Bucket, Storage
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+
+class EngagementData(BaseModel):
+    """Model for the full engagement data file stored in GCS."""
+
+    amp: dict[str, dict[str, str | int]] = {}
+    amp_aggregated: dict[str, int] = {}
+    wiki_aggregated: dict[str, int] = {}
+
+
+class EngagementFilemanager:
+    """Filemanager for fetching engagement data from GCS asynchronously."""
+
+    gcs_bucket_path: str
+    blob_name: str
+    gcs_client: Storage | None
+    bucket: Bucket | None
+
+    def __init__(self, gcs_bucket_path: str, blob_name: str) -> None:
+        """:param gcs_bucket_path: GCS bucket name to fetch from.
+        :param blob_name: Name of the blob in the GCS bucket.
+        """
+        self.gcs_bucket_path = gcs_bucket_path
+        self.blob_name = blob_name
+        self.gcs_client = None
+        self.bucket = None
+
+    def get_bucket(self) -> Bucket:
+        """Lazily instantiate the GCS client and return the configured bucket."""
+        if self.bucket is not None:
+            return self.bucket
+
+        if self.gcs_client is None:
+            self.gcs_client = Storage()
+
+        self.bucket = Bucket(storage=self.gcs_client, name=self.gcs_bucket_path)
+        return self.bucket
+
+    async def get_file(self) -> EngagementData | None:
+        """Fetch the engagement data file from GCS and return a validated model instance."""
+        try:
+            bucket = self.get_bucket()
+            blob: Blob = await bucket.get_blob(self.blob_name)
+            blob_data = await blob.download()
+            return EngagementData.model_validate(orjson.loads(blob_data))
+        except (JSONDecodeError, ValueError) as e:
+            logger.error(f"Failed to decode engagement data JSON: {e}")
+        except Exception as e:
+            logger.error(f"Error fetching engagement data file {self.blob_name}: {e}")
+        return None