Add jobs views tracker (#258)

Related to #250

Signed-off-by: Sergio Castaño Arteaga <[email protected]>
Signed-off-by: Cintia Sánchez García <[email protected]>
Co-authored-by: Cintia Sánchez García <[email protected]>

Author: tegioz

Cargo.lock

@@ -33,6 +33,7 @@ lettre = { version = "0.11.15", default-features = false, features = ["builder",
 markdown = "1.0.0-alpha.24"
 mime_guess = "2.0.5"
 minify-html = "0.16.4"
+mockall = "0.13.1"
 num-format = "0.4.4"
 oauth2 = "5.0.0"
 openidconnect = { version = "4.0.0", features = ["accept-rfc3339-timestamps"] }

Cargo.toml

@@ -2,6 +2,7 @@
 {{ template "dashboard/search_applications.sql" }}
 {{ template "img/get_image_version.sql" }}
 {{ template "jobboard/search_jobs.sql" }}
+{{ template "jobboard/update_jobs_views.sql" }}
 {{ template "misc/search_locations.sql" }}
 
 ---- create above / drop below ----

database/migrations/functions/001_load_functions.sql

@@ -0,0 +1,16 @@
+-- update_jobs_views updates the views of the jobs provided.
+create or replace function update_jobs_views(p_lock_key bigint, p_data jsonb)
+returns void as $$
+    -- Make sure only one batch of updates is processed at a time
+    select pg_advisory_xact_lock(p_lock_key);
+
+    -- Insert or update the corresponding views counters as needed
+    insert into job_views (job_id, day, total)
+    select
+        (value->>0)::uuid as job_id,
+        (value->>1)::date as day,
+        (value->>2)::integer as total
+    from jsonb_array_elements(p_data)
+    on conflict (job_id, day) do
+    update set total = job_views.total + excluded.total;
+$$ language sql;

database/migrations/functions/jobboard/update_jobs_views.sql

@@ -0,0 +1,10 @@
+create table if not exists job_views (
+     job_id uuid references job on delete set null,
+     day date not null,
+     total integer not null,
+     unique (job_id, day)
+ );
+
+---- create above / drop below ----
+
+drop table if exists job_views;

database/migrations/schema/0009_job_views.sql

@@ -58,6 +58,8 @@ unicode-segmentation = { workspace = true }
 uuid = { workspace = true }
 
 [dev-dependencies]
+futures = { workspace = true }
+mockall = { workspace = true }
 
 [build-dependencies]
 anyhow = { workspace = true }

gitjobs-server/Cargo.toml

@@ -17,6 +17,7 @@ use tokio::{select, time::sleep};
 use tokio_util::sync::CancellationToken;
 use tracing::instrument;
 use uuid::Uuid;
+use views::DBViews;
 use workers::DBWorkers;
 
 pub(crate) mod auth;
@@ -25,6 +26,7 @@ pub(crate) mod img;
 pub(crate) mod jobboard;
 pub(crate) mod misc;
 pub(crate) mod notifications;
+pub(crate) mod views;
 pub(crate) mod workers;
 
 /// Error message when a transaction client is not found.
@@ -40,7 +42,7 @@ const TXS_CLIENT_TIMEOUT: TimeDelta = TimeDelta::seconds(10);
 /// DB implementation must support.
 #[async_trait]
 pub(crate) trait DB:
-    DBJobBoard + DBDashBoard + DBAuth + DBImage + DBNotifications + DBWorkers + DBMisc
+    DBJobBoard + DBDashBoard + DBAuth + DBImage + DBNotifications + DBWorkers + DBViews + DBMisc
 {
     /// Begin transaction.
     async fn tx_begin(&self) -> Result<Uuid>;

gitjobs-server/src/db/mod.rs

@@ -0,0 +1,46 @@
+//! This module defines some database functionality used in the views tracker.
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use async_trait::async_trait;
+#[cfg(test)]
+use mockall::automock;
+use tokio_postgres::types::Json;
+use tracing::{instrument, trace};
+
+use crate::{
+    db::PgDB,
+    views::{Day, JobId, Total},
+};
+
+// Lock key used when updating the jobs views in the database.
+const LOCK_KEY_UPDATE_JOBS_VIEWS: i64 = 1;
+
+/// Trait that defines some database operations used in the views tracker.
+#[async_trait]
+#[cfg_attr(test, automock)]
+pub(crate) trait DBViews {
+    /// Update the number of views of the jobs provided.
+    async fn update_jobs_views(&self, data: Vec<(JobId, Day, Total)>) -> Result<()>;
+}
+
+/// Type alias to represent a `DBViews` trait object.
+pub(crate) type DynDBViews = Arc<dyn DBViews + Send + Sync>;
+
+#[async_trait]
+impl DBViews for PgDB {
+    #[instrument(skip(self), err)]
+    async fn update_jobs_views(&self, data: Vec<(JobId, Day, Total)>) -> Result<()> {
+        trace!("db: update jobs views");
+
+        let db = self.pool.get().await?;
+        db.execute(
+            "select update_jobs_views($1::bigint, $2::jsonb)",
+            &[&LOCK_KEY_UPDATE_JOBS_VIEWS, &Json(&data)],
+        )
+        .await?;
+
+        Ok(())
+    }
+}

gitjobs-server/src/db/views.rs

@@ -24,6 +24,7 @@ use crate::{
         jobboard::jobs::{ExploreSection, Filters, JobSection, JobsPage, ResultsSection},
         pagination::{NavigationLinks, build_url},
     },
+    views::DynViewsTracker,
 };
 
 // Pages and sections handlers.
@@ -132,3 +133,14 @@ pub(crate) async fn apply(
 
     Ok(StatusCode::NO_CONTENT)
 }
+
+/// Handler used to track a job view.
+#[instrument(skip_all, err)]
+pub(crate) async fn track_view(
+    State(views_tracker): State<DynViewsTracker>,
+    Path(job_id): Path<Uuid>,
+) -> Result<impl IntoResponse, HandlerError> {
+    views_tracker.track_view(job_id).await?;
+
+    Ok(StatusCode::NO_CONTENT)
+}

gitjobs-server/src/handlers/jobboard/jobs.rs

@@ -15,6 +15,7 @@ use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
+use views::ViewsTrackerDB;
 
 use crate::{
     config::{Config, LogFormat},
@@ -29,6 +30,7 @@ mod img;
 mod notifications;
 mod router;
 mod templates;
+mod views;
 mod workers;
 
 #[derive(Debug, Parser)]
@@ -82,19 +84,29 @@ async fn main() -> Result<()> {
     // Setup image store
     let image_store = Arc::new(DbImageStore::new(db.clone()));
 
-    // Run some workers
-    workers::run(db.clone(), &tracker, cancellation_token.clone());
-
-    // Setup and launch notifications manager
+    // Setup notifications manager
     let notifications_manager = Arc::new(PgNotificationsManager::new(
         db.clone(),
-        cfg.email,
-        tracker.clone(),
-        cancellation_token.clone(),
+        &cfg.email,
+        &tracker,
+        &cancellation_token,
     )?);
 
+    // Setup views tracker
+    let views_tracker = Arc::new(ViewsTrackerDB::new(db.clone(), &tracker, &cancellation_token));
+
+    // Run some other workers
+    workers::run(db.clone(), &tracker, cancellation_token.clone());
+
     // Setup and launch HTTP server
-    let router = router::setup(cfg.server.clone(), db, image_store, notifications_manager).await?;
+    let router = router::setup(
+        cfg.server.clone(),
+        db,
+        image_store,
+        notifications_manager,
+        views_tracker,
+    )
+    .await?;
     let listener = TcpListener::bind(&cfg.server.addr).await?;
     info!("server started");
     info!(%cfg.server.addr, "listening");

gitjobs-server/src/main.rs

@@ -48,54 +48,38 @@ pub(crate) type DynNotificationsManager = Arc<dyn NotificationsManager + Send +
 /// Notifications manager backed by `PostgreSQL`.
 pub(crate) struct PgNotificationsManager {
     db: DynDB,
-    cfg: EmailConfig,
-    tracker: TaskTracker,
-    cancellation_token: CancellationToken,
 }
 
 impl PgNotificationsManager {
     /// Create a new `PgNotificationsManager` instance.
     pub(crate) fn new(
         db: DynDB,
-        cfg: EmailConfig,
-        tracker: TaskTracker,
-        cancellation_token: CancellationToken,
+        cfg: &EmailConfig,
+        tracker: &TaskTracker,
+        cancellation_token: &CancellationToken,
     ) -> Result<Self> {
-        let notifications_manager = Self {
-            db,
-            cfg,
-            tracker,
-            cancellation_token,
-        };
-        notifications_manager.run()?;
-
-        Ok(notifications_manager)
-    }
-
-    /// Run notifications manager.
-    fn run(&self) -> Result<()> {
         // Setup smtp client
-        let smtp_client = AsyncSmtpTransport::<Tokio1Executor>::relay(&self.cfg.smtp.host)?
+        let smtp_client = AsyncSmtpTransport::<Tokio1Executor>::relay(&cfg.smtp.host)?
             .credentials(Credentials::new(
-                self.cfg.smtp.username.clone(),
-                self.cfg.smtp.password.clone(),
+                cfg.smtp.username.clone(),
+                cfg.smtp.password.clone(),
             ))
             .build();
 
         // Setup and run some workers to deliver notifications
         for _ in 1..=NUM_WORKERS {
             let mut worker = Worker {
-                db: self.db.clone(),
-                cfg: self.cfg.clone(),
+                db: db.clone(),
+                cfg: cfg.clone(),
                 smtp_client: smtp_client.clone(),
-                cancellation_token: self.cancellation_token.clone(),
+                cancellation_token: cancellation_token.clone(),
             };
-            self.tracker.spawn(async move {
+            tracker.spawn(async move {
                 worker.run().await;
             });
         }
 
-        Ok(())
+        Ok(Self { db })
     }
 }