[AC-154] implement fire and forget ads client async

components/ads-client/src/ads_cache.rs

@@ -0,0 +1,99 @@
+use crate::mars::ad_response::{AdImage, AdSpoc, AdTile};
+use std::{collections::HashMap, time::Duration};
+
+// TODO: This is an intentionally naive in-memory cache implementation of the ads cache.
+// It functions as a skeleton to store ads fetched in the background, and has a naive expiration mechanism.
+// The subsequent vertical slice will replace this in its entirety with the http_cache sqlite database instead, with TTLs, persistent storage, etc.
+const DEFAULT_TTL: Duration = Duration::from_secs(300);
+
+#[derive(Debug)]
+pub struct AdsCache {
+    image_ads: HashMap<String, (u64, AdImage)>,
+    spoc_ads: HashMap<String, (u64, Vec<AdSpoc>)>,
+    tile_ads: HashMap<String, (u64, AdTile)>,
+}
+
+impl Default for AdsCache {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AdsCache {
+    pub fn new() -> Self {
+        AdsCache {
+            image_ads: HashMap::new(),
+            spoc_ads: HashMap::new(),
+            tile_ads: HashMap::new(),
+        }
+    }
+
+    pub fn cache_ads<T: AdsCacheable>(
+        &mut self,
+        ads: HashMap<String, T::StorageType>,
+        timestamp: u64,
+    ) {
+        T::cache_ads(ads, self, timestamp);
+    }
+
+    pub fn get_cached_ads<'a, T: AdsCacheable>(
+        &'a self,
+        placement: &str,
+    ) -> Option<&'a T::StorageType> {
+        T::fetch_cached_ads(self, placement)
+    }
+}
+
+pub trait AdsCacheable: Sized {
+    // The cached ad(s) to store (eg: this may be a single ad, or an array of ads)
+    type StorageType;
+
+    fn cache_ads(ads: HashMap<String, Self::StorageType>, ads_cache: &mut AdsCache, timestamp: u64);
+    fn fetch_cached_ads<'a>(ads_cache: &'a AdsCache, id: &str) -> Option<&'a Self::StorageType>;
+}
+
+impl AdsCacheable for AdImage {
+    type StorageType = AdImage;
+    fn cache_ads(ads: HashMap<String, AdImage>, ads_cache: &mut AdsCache, timestamp: u64) {
+        ads_cache
+            .image_ads
+            .extend(ads.into_iter().map(|(key, ad)| (key, (timestamp, ad))));
+        ads_cache
+            .image_ads
+            .retain(|_, (x, _)| *x - timestamp < DEFAULT_TTL.as_secs());
+    }
+
+    fn fetch_cached_ads<'a>(ads_cache: &'a AdsCache, id: &str) -> Option<&'a AdImage> {
+        ads_cache.image_ads.get(id).map(|(_, ads)| ads)
+    }
+}
+
+impl AdsCacheable for AdSpoc {
+    type StorageType = Vec<AdSpoc>;
+    fn cache_ads(ads: HashMap<String, Vec<AdSpoc>>, ads_cache: &mut AdsCache, timestamp: u64) {
+        ads_cache
+            .spoc_ads
+            .extend(ads.into_iter().map(|(key, ad)| (key, (timestamp, ad))));
+        ads_cache
+            .spoc_ads
+            .retain(|_, (x, _)| *x - timestamp < DEFAULT_TTL.as_secs());
+    }
+    fn fetch_cached_ads<'a>(ads_cache: &'a AdsCache, id: &str) -> Option<&'a Vec<AdSpoc>> {
+        ads_cache.spoc_ads.get(id).map(|(_, ads)| ads)
+    }
+}
+
+impl AdsCacheable for AdTile {
+    type StorageType = AdTile;
+    fn cache_ads(ads: HashMap<String, AdTile>, ads_cache: &mut AdsCache, timestamp: u64) {
+        ads_cache
+            .tile_ads
+            .extend(ads.into_iter().map(|(key, ad)| (key, (timestamp, ad))));
+        ads_cache
+            .tile_ads
+            .retain(|_, (x, _)| *x - timestamp < DEFAULT_TTL.as_secs());
+    }
+    fn fetch_cached_ads<'a>(ads_cache: &'a AdsCache, id: &str) -> Option<&'a AdTile> {
+        ads_cache.tile_ads.get(id).map(|(_, ads)| ads)
+    }
+}

components/ads-client/src/client.rs

@@ -3,9 +3,7 @@
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */
 
-use std::collections::HashMap;
-use std::time::Duration;
-
+use crate::ads_cache::{AdsCache, AdsCacheable};
 use crate::http_cache::{ByteSize, CachePolicy, HttpCache};
 use crate::mars::ad_request::{AdPlacementRequest, AdRequestFlags};
 use crate::mars::ad_response::{AdImage, AdResponse, AdResponseValue, AdSpoc, AdTile};
@@ -15,6 +13,8 @@ use crate::telemetry::Telemetry;
 use config::AdsClientConfig;
 use context_id::{ContextIDComponent, DefaultContextIdCallback};
 use error::RequestAdsError;
+use std::collections::HashMap;
+use std::time::Duration;
 use url::Url;
 use uuid::Uuid;
 
@@ -42,6 +42,7 @@ where
     client: MARSClient<T>,
     context_id_provider: Box<dyn ContextIdProvider>,
     telemetry: T,
+    ads_cache: AdsCache,
 }
 
 impl<T> AdsClient<T>
@@ -91,6 +92,7 @@ where
             client,
             context_id_provider,
             telemetry: telemetry.clone(),
+            ads_cache: AdsCache::new(),
         }
     }
 
@@ -110,6 +112,15 @@ where
         Ok(())
     }
 
+    pub fn cache_ads<A: AdsCacheable>(&mut self, ads: HashMap<String, A::StorageType>) {
+        let now = chrono::Utc::now().timestamp().unsigned_abs();
+        self.ads_cache.cache_ads::<A>(ads, now);
+    }
+
+    pub fn get_cached_ads<A: AdsCacheable>(&self, placement_id: &str) -> Option<&A::StorageType> {
+        self.ads_cache.get_cached_ads::<A>(placement_id)
+    }
+
     pub fn get_context_id(&self) -> context_id::ApiResult<String> {
         self.context_id_provider.context_id()
     }
@@ -264,6 +275,7 @@ where
     }
 }
 
+// Event fires in both sync and background strategies.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum ClientOperationEvent {
     New,
@@ -273,6 +285,35 @@ pub enum ClientOperationEvent {
     RequestAds,
 }
 
+// Event fires when dispatch is fired, not when the event resolves.
+pub enum CommandDispatchedOperationEvent {
+    RecordClick,
+    RecordImpression,
+    ReportAd,
+    RequestAds,
+}
+
+// Event fires when the corresponding background event resolves.
+pub enum CommandProcessedOperationEvent {
+    RecordClick,
+    RecordImpression,
+    ReportAd,
+    RequestAds,
+}
+
+// Event fires when the corresponding background event fails to resolve.
+pub enum CommandFailedOperationEvent {
+    RecordClick,
+    RecordImpression,
+    ReportAd,
+    RequestAds,
+}
+
+pub enum WorkerMetaEvent {
+    Start,
+    Stop,
+}
+
 #[cfg(test)]
 mod tests {
     use std::{assert_eq, assert_ne, sync::Arc};
@@ -301,6 +342,7 @@ mod tests {
                 Box::new(DefaultContextIdCallback),
             )),
             telemetry,
+            ads_cache: AdsCache::new(),
         }
     }
 

components/ads-client/src/client/error.rs

@@ -3,7 +3,11 @@
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */
 
-use crate::mars::error::{FetchAdsError, RecordClickError, RecordImpressionError, ReportAdError};
+use crate::{
+    mars::error::{FetchAdsError, RecordClickError, RecordImpressionError, ReportAdError},
+    worker::command,
+};
+use std::sync::mpsc::{RecvTimeoutError, TrySendError};
 
 #[derive(Debug, thiserror::Error)]
 pub enum ComponentError {
@@ -18,6 +22,9 @@ pub enum ComponentError {
 
     #[error("Error requesting ads: {0}")]
     RequestAds(#[from] RequestAdsError),
+
+    #[error("Error requesting ads from worker: {0}")]
+    BackgroundWorker(#[from] BackgroundWorkerError),
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -28,3 +35,28 @@ pub enum RequestAdsError {
     #[error("Error requesting ads from MARS: {0}")]
     FetchAds(#[from] FetchAdsError),
 }
+
+#[derive(Debug, thiserror::Error)]
+pub enum BackgroundWorkerError {
+    #[error("Error requesting new ads from the background worker: worker full")]
+    WorkerFull,
+
+    #[error("Error requesting new ads from the background worker: worker closed")]
+    WorkerClosed,
+
+    #[error("Worker timed out waiting for response: {0}")]
+    WorkerTimedOut(#[from] RecvTimeoutError),
+
+    #[error("Error sending pong back from background worker")]
+    PongFailure(Box<TrySendError<()>>),
+}
+
+impl From<TrySendError<command::DispatchCommand>> for BackgroundWorkerError {
+    // TODO: For future vertical slice (for retries), we may want to keep the failed dispatch for retrying
+    fn from(value: TrySendError<command::DispatchCommand>) -> Self {
+        match value {
+            TrySendError::Disconnected(_) => BackgroundWorkerError::WorkerClosed,
+            TrySendError::Full(_) => BackgroundWorkerError::WorkerFull,
+        }
+    }
+}

components/ads-client/src/ffi.rs

@@ -6,8 +6,6 @@
 pub mod error;
 pub mod telemetry;
 
-use std::sync::Arc;
-
 use crate::client::config::{AdsCacheConfig, AdsClientConfig};
 use crate::client::{AdsClient, ContextIdProvider};
 use crate::ffi::telemetry::MozAdsTelemetryWrapper;
@@ -20,14 +18,14 @@ use crate::mars::ad_response::{
 };
 use crate::mars::Environment;
 use crate::mars::ReportReason;
-use crate::AdsClientUrl;
 use crate::MozAdsClient;
+use crate::{worker, AdsClientUrl};
 use parking_lot::Mutex;
 use std::collections::HashMap;
+use std::sync::Arc;
 
 pub use error::{AdsClientApiResult, MozAdsClientApiError};
 pub use telemetry::MozAdsTelemetry;
-
 // TODO: Temporary workaround for HNT requirements — do not use for new integrations.
 // Context ID management should remain internal to the ads client and this interface should be removed.
 #[uniffi::export(with_foreign)]
@@ -55,7 +53,7 @@ impl From<MozAdsContextIdProviderWrapper> for Box<dyn ContextIdProvider> {
     }
 }
 
-#[derive(Default, uniffi::Record)]
+#[derive(Default, uniffi::Record, Clone)]
 pub struct MozAdsRequestOptions {
     pub cache_policy: Option<MozAdsCachePolicy>,
     #[uniffi(default)]
@@ -124,6 +122,11 @@ impl MozAdsClientBuilder {
 
     pub fn build(&self) -> MozAdsClient {
         let inner = self.0.lock();
+        let telemetry = inner
+            .telemetry
+            .clone()
+            .map(MozAdsTelemetryWrapper::new)
+            .unwrap_or_else(MozAdsTelemetryWrapper::noop);
         let client_config = AdsClientConfig {
             cache_config: inner.cache_config.clone().map(Into::into),
             context_id_provider: inner
@@ -132,16 +135,12 @@ impl MozAdsClientBuilder {
                 .map(MozAdsContextIdProviderWrapper::new)
                 .map(Into::into),
             environment: inner.environment.unwrap_or_default().into(),
-            telemetry: inner
-                .telemetry
-                .clone()
-                .map(MozAdsTelemetryWrapper::new)
-                .unwrap_or_else(MozAdsTelemetryWrapper::noop),
+            telemetry: telemetry.clone(),
         };
         let client = AdsClient::new(client_config);
-        MozAdsClient {
-            inner: Mutex::new(client),
-        }
+        let inner = Arc::new(Mutex::new(client));
+        let worker = worker::AdsClientWorkerWrapper::new(inner.clone(), telemetry);
+        MozAdsClient { inner, worker }
     }
 
     pub fn cache_config(self: Arc<Self>, cache_config: MozAdsCacheConfig) -> Arc<Self> {

components/ads-client/src/ffi/telemetry.rs

@@ -9,7 +9,10 @@ use std::sync::Arc;
 use parking_lot::RwLock;
 
 use crate::client::error::RequestAdsError;
-use crate::client::ClientOperationEvent;
+use crate::client::{
+    ClientOperationEvent, CommandDispatchedOperationEvent, CommandFailedOperationEvent,
+    CommandProcessedOperationEvent, WorkerMetaEvent,
+};
 use crate::http_cache::{CacheOutcome, HttpCacheBuilderError};
 use crate::mars::error::{RecordClickError, RecordImpressionError, ReportAdError};
 use crate::telemetry::Telemetry;
@@ -101,6 +104,55 @@ impl Telemetry for MozAdsTelemetryWrapper {
             });
             return;
         }
+
+        if let Some(client_op) = event.downcast_ref::<CommandDispatchedOperationEvent>() {
+            inner.record_client_operation_total(match client_op {
+                CommandDispatchedOperationEvent::RecordClick => {
+                    "cmd_dispatch_record_click".to_string()
+                }
+                CommandDispatchedOperationEvent::RecordImpression => {
+                    "cmd_dispatch_record_impression".to_string()
+                }
+                CommandDispatchedOperationEvent::ReportAd => "cmd_dispatch_report_ad".to_string(),
+                CommandDispatchedOperationEvent::RequestAds => "cmd_dispatch_report_ad".to_string(),
+            });
+            return;
+        }
+
+        if let Some(client_op) = event.downcast_ref::<CommandProcessedOperationEvent>() {
+            inner.record_client_operation_total(match client_op {
+                CommandProcessedOperationEvent::RecordClick => {
+                    "cmd_processed_record_click".to_string()
+                }
+                CommandProcessedOperationEvent::RecordImpression => {
+                    "cmd_processed_record_impression".to_string()
+                }
+                CommandProcessedOperationEvent::ReportAd => "cmd_processed_report_ad".to_string(),
+                CommandProcessedOperationEvent::RequestAds => "cmd_processed_report_ad".to_string(),
+            });
+            return;
+        }
+
+        if let Some(client_op) = event.downcast_ref::<CommandFailedOperationEvent>() {
+            inner.record_client_operation_total(match client_op {
+                CommandFailedOperationEvent::RecordClick => "cmd_failed_record_click".to_string(),
+                CommandFailedOperationEvent::RecordImpression => {
+                    "cmd_failed_record_impression".to_string()
+                }
+                CommandFailedOperationEvent::ReportAd => "cmd_failed_report_ad".to_string(),
+                CommandFailedOperationEvent::RequestAds => "cmd_failed_report_ad".to_string(),
+            });
+            return;
+        }
+
+        if let Some(client_op) = event.downcast_ref::<WorkerMetaEvent>() {
+            inner.record_client_operation_total(match client_op {
+                WorkerMetaEvent::Start => "worker_started".to_string(),
+                WorkerMetaEvent::Stop => "worker_ended".to_string(),
+            });
+            return;
+        }
+
         if let Some(cache_builder_error) = event.downcast_ref::<HttpCacheBuilderError>() {
             inner.record_build_cache_error(
                 match cache_builder_error {