feat: Adds the background worker for fire-and-forget ads-client

components/ads-client/src/ads_store.rs

@@ -7,9 +7,12 @@ use serde::{Deserialize, Serialize};
 use crate::{
     ads_store::{builder::AdsStoreBuilder, store::AdsStoreHolder},
     common::bytesize::ByteSize,
-    mars::ad_response::{AdImage, AdSpoc, AdTile},
+    mars::{
+        ad_response::{AdImage, AdSpoc, AdTile},
+        error::FetchAdsError,
+    },
 };
-use std::path::Path;
+use std::{collections::HashMap, path::Path};
 
 /// Identification of placement sent and returned from MARS (eg: `mock_spoc_1`)
 #[derive(Debug, Hash, PartialEq, Eq, Clone)]
@@ -19,9 +22,6 @@ impl PlacementId {
     pub fn new(s: &str) -> PlacementId {
         PlacementId(s.to_string())
     }
-    pub fn into_inner(self) -> String {
-        self.0
-    }
 }
 
 impl AsRef<str> for PlacementId {
@@ -30,13 +30,51 @@ impl AsRef<str> for PlacementId {
     }
 }
 
+impl From<String> for PlacementId {
+    fn from(value: String) -> Self {
+        PlacementId(value)
+    }
+}
+
+impl From<PlacementId> for String {
+    fn from(value: PlacementId) -> Self {
+        value.0
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub enum StorableAd {
     Image(AdImage),
-    Spoc(AdSpoc),
+    Spoc(Vec<AdSpoc>),
     Tile(AdTile),
 }
 
+impl StorableAd {
+    pub fn into_image(self) -> Option<AdImage> {
+        if let StorableAd::Image(image) = self {
+            Some(image)
+        } else {
+            None
+        }
+    }
+
+    pub fn into_spocs(self) -> Option<Vec<AdSpoc>> {
+        if let StorableAd::Spoc(spocs) = self {
+            Some(spocs)
+        } else {
+            None
+        }
+    }
+
+    pub fn into_tile(self) -> Option<AdTile> {
+        if let StorableAd::Tile(tile) = self {
+            Some(tile)
+        } else {
+            None
+        }
+    }
+}
+
 pub struct AdsStore {
     holder: AdsStoreHolder,
     #[allow(dead_code)]
@@ -61,6 +99,17 @@ impl AdsStore {
         self.holder.invalidate_ad_by_id(placement_id)?;
         Ok(())
     }
+
+    pub fn lookup(&self, placement_id: &PlacementId) -> Result<Option<StorableAd>, FetchAdsError> {
+        self.holder.lookup(placement_id)
+    }
+
+    pub fn store_ads(&self, ads: HashMap<PlacementId, StorableAd>) -> Result<(), FetchAdsError> {
+        for (placement_id, ad) in ads {
+            self.holder.store_ad(&placement_id, ad)?;
+        }
+        Ok(())
+    }
 }
 
 #[cfg(test)]

components/ads-client/src/client.rs

@@ -4,11 +4,13 @@
 */
 
 #[cfg(feature = "stateful")]
-use crate::ads_store::AdsStore;
+use crate::ads_store::{AdsStore, PlacementId, StorableAd};
 use crate::common::bytesize::ByteSize;
 use crate::http_cache::{CachePolicy, HttpCache};
 use crate::mars::ad_request::{AdPlacementRequest, AdRequestFlags};
 use crate::mars::ad_response::{AdImage, AdResponse, AdResponseValue, AdSpoc, AdTile};
+#[cfg(feature = "stateful")]
+use crate::mars::error::FetchAdsError;
 use crate::mars::error::{RecordClickError, RecordImpressionError, ReportAdError};
 use crate::mars::{MARSClient, ReportReason};
 #[cfg(feature = "stateful")]
@@ -111,6 +113,71 @@ where
         self.client.clear_cache()
     }
 
+    #[cfg(feature = "stateful")]
+    pub fn store_ads(
+        &mut self,
+        ads: HashMap<PlacementId, StorableAd>,
+    ) -> Result<(), FetchAdsError> {
+        let ads_store = self.ads_store.lock();
+        if let Some(ads_store) = ads_store.as_ref() {
+            ads_store.store_ads(ads)?;
+            Ok(())
+        } else {
+            Err(FetchAdsError::SqliteShutdown)
+        }
+    }
+
+    #[cfg(feature = "stateful")]
+    pub fn get_stored_ad_images(&self, placement_id: &PlacementId) -> Option<AdImage> {
+        let ads_store = self.ads_store.lock();
+        if let Some(ads_store) = ads_store.as_ref() {
+            match ads_store.lookup(placement_id) {
+                Ok(ad) => ad.and_then(|ad| ad.into_image()),
+                Err(_) => {
+                    // TODO: Telemetry should return an error here (eg: some internal sqlite error)
+                    None
+                }
+            }
+        } else {
+            // TODO: Telemetry should be added here for the database being shut down.
+            None
+        }
+    }
+
+    #[cfg(feature = "stateful")]
+    pub fn get_stored_ad_spocs(&self, placement_id: &PlacementId) -> Option<Vec<AdSpoc>> {
+        let ads_store = self.ads_store.lock();
+        if let Some(ads_store) = ads_store.as_ref() {
+            match ads_store.lookup(placement_id) {
+                Ok(ad) => ad.and_then(|ad| ad.into_spocs()),
+                Err(_) => {
+                    // TODO: Telemetry should return an error here (eg: some internal sqlite error)
+                    None
+                }
+            }
+        } else {
+            // TODO: Telemetry should be added here for the database being shut down.
+            None
+        }
+    }
+
+    #[cfg(feature = "stateful")]
+    pub fn get_stored_ad_tile(&self, placement_id: &PlacementId) -> Option<AdTile> {
+        let ads_store = self.ads_store.lock();
+        if let Some(ads_store) = ads_store.as_ref() {
+            match ads_store.lookup(placement_id) {
+                Ok(ad) => ad.and_then(|ad| ad.into_tile()),
+                Err(_) => {
+                    // TODO: Telemetry should return an error here (eg: some internal sqlite error)
+                    None
+                }
+            }
+        } else {
+            // TODO: Telemetry should be added here for the database being shut down.
+            None
+        }
+    }
+
     pub fn get_context_id(&self) -> context_id::ApiResult<String> {
         self.context_id_component.request(DEFAULT_ROTATION_DAYS)
     }

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

@@ -3,10 +3,19 @@
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */
 
+#[cfg(feature = "stateful")]
+use std::sync::mpsc::{RecvTimeoutError, TrySendError};
+
 use crate::mars::error::{FetchAdsError, RecordClickError, RecordImpressionError, ReportAdError};
+#[cfg(feature = "stateful")]
+use crate::worker::command;
 
 #[derive(Debug, thiserror::Error)]
 pub enum ComponentError {
+    #[cfg(feature = "stateful")]
+    #[error("Error requesting ads from worker: {0}")]
+    BackgroundWorker(#[from] BackgroundWorkerError),
+
     #[error("Error recording a click for a placement: {0}")]
     RecordClick(#[from] RecordClickError),
 
@@ -28,3 +37,26 @@ pub enum RequestAdsError {
     #[error("Error requesting ads from MARS: {0}")]
     FetchAds(#[from] FetchAdsError),
 }
+
+#[cfg(feature = "stateful")]
+#[derive(Debug, thiserror::Error)]
+pub enum BackgroundWorkerError {
+    #[error("Error requesting new ads from the background worker: worker closed")]
+    Closed,
+
+    #[error("Error requesting new ads from the background worker: worker full")]
+    Full,
+
+    #[error("Background worker timed out waiting for response: {0}")]
+    TimedOut(#[from] RecvTimeoutError),
+}
+
+#[cfg(feature = "stateful")]
+impl From<TrySendError<command::DispatchCommand>> for BackgroundWorkerError {
+    fn from(value: TrySendError<command::DispatchCommand>) -> Self {
+        match value {
+            TrySendError::Disconnected(_) => BackgroundWorkerError::Closed,
+            TrySendError::Full(_) => BackgroundWorkerError::Full,
+        }
+    }
+}

components/ads-client/src/ffi.rs

@@ -20,6 +20,8 @@ use crate::mars::ad_response::{
 };
 use crate::mars::Environment;
 use crate::mars::ReportReason;
+#[cfg(feature = "stateful")]
+use crate::worker;
 use crate::AdsClientUrl;
 use crate::MozAdsClient;
 use parking_lot::Mutex;
@@ -107,6 +109,13 @@ impl MozAdsClientBuilder {
             .take()
             .map(MozAdsTelemetryWrapper::new)
             .unwrap_or_else(MozAdsTelemetryWrapper::noop);
+        #[cfg(feature = "stateful")]
+        let store_set = inner.store_config.is_some();
+        #[cfg(feature = "stateful")]
+        let worker_buffer_size = inner
+            .store_config
+            .as_ref()
+            .and_then(|x| x.worker_buffer_size);
         let client_config = AdsClientConfig {
             cache_config: inner.cache_config.clone().map(Into::into),
             environment: inner.environment.clone().unwrap_or_default().into(),
@@ -116,9 +125,18 @@ impl MozAdsClientBuilder {
         };
         let client = AdsClient::new(client_config);
         let shutdown_references = client.shutdown_references();
+        let inner = Arc::new(Mutex::new(client));
+        #[cfg(feature = "stateful")]
+        let worker = if store_set {
+            worker::BackgroundWorker::new(inner.clone(), worker_buffer_size)
+        } else {
+            worker::BackgroundWorker::new_empty()
+        };
         MozAdsClient {
-            inner: Mutex::new(client),
+            inner,
             shutdown_references,
+            #[cfg(feature = "stateful")]
+            _worker: worker,
         }
     }
 
@@ -172,6 +190,7 @@ pub struct MozAdsCacheConfig {
 #[derive(Clone, uniffi::Record)]
 pub struct MozAdsStoreConfig {
     pub db_path: String,
+    pub worker_buffer_size: Option<u32>,
 }
 
 #[derive(Debug, PartialEq, uniffi::Record)]

components/ads-client/src/lib.rs

@@ -3,7 +3,7 @@
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */
 
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
 use client::error::ComponentError;
 use error_support::handle_error;
@@ -23,9 +23,13 @@ pub mod http_cache;
 mod mars;
 pub mod shutdown;
 pub mod telemetry;
+#[cfg(feature = "stateful")]
+pub mod worker;
 
 pub use ffi::*;
 
+#[cfg(feature = "stateful")]
+use crate::worker::BackgroundWorker;
 use crate::{ffi::telemetry::MozAdsTelemetryWrapper, shutdown::ShutdownReferences};
 
 #[cfg(test)]
@@ -39,10 +43,13 @@ uniffi::custom_type!(AdsClientUrl, String, {
     lower: |obj| obj.as_str().to_string(),
 });
 
+pub type MozAdsClientInner = Arc<Mutex<AdsClient<MozAdsTelemetryWrapper>>>;
 #[derive(uniffi::Object)]
 pub struct MozAdsClient {
-    inner: Mutex<AdsClient<MozAdsTelemetryWrapper>>,
+    inner: MozAdsClientInner,
     shutdown_references: ShutdownReferences<MozAdsTelemetryWrapper>,
+    #[cfg(feature = "stateful")]
+    _worker: BackgroundWorker,
 }
 
 #[uniffi::export]

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

@@ -46,6 +46,9 @@ pub enum FetchAdsError {
     #[error("Internal database error: {0}")]
     Sqlite(#[from] rusqlite::Error),
 
+    #[error("Internal database error: database shut down or uninitialized")]
+    SqliteShutdown,
+
     #[error("Error sending request: {0}")]
     Request(#[from] viaduct::ViaductError),