changes

crates/worker/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "worker"
+edition.workspace = true
+version.workspace = true
+
+[[bin]]
+name = "k-mood-worker"
+path = "src/main.rs"
+
+[dependencies]
+bootstrap.workspace = true
+domain.workspace = true
+application.workspace = true
+config.workspace = true
+tokio.workspace = true
+tracing.workspace = true

crates/worker/src/loops.rs

@@ -0,0 +1,147 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use application::job::use_cases::{run_due_jobs, sweep_recording_backlog, sweep_weather_backlog};
+use application::reminder::use_cases::process_due_reminders;
+use config::WorkerConfig;
+use domain::ports::{
+    EntryDimensionPort, JobQueueCommandPort, RecordingBackfillQueryPort, RecordingLookupPort,
+    RefreshSessionCommandPort, ReminderQueryPort, ReminderSenderPort, UserQueryPort,
+    WeatherBacklogQueryPort, WeatherLookupPort,
+};
+
+pub fn every(seconds: u64) -> tokio::time::Interval {
+    let mut interval = tokio::time::interval(Duration::from_secs(seconds.max(1)));
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+
+    interval
+}
+
+pub struct QueueDependencies {
+    pub queue: Arc<dyn JobQueueCommandPort>,
+    pub backfill: Arc<dyn RecordingBackfillQueryPort>,
+    pub recordings: Arc<dyn RecordingLookupPort>,
+    pub places: Arc<dyn WeatherBacklogQueryPort>,
+    pub weather: Option<Arc<dyn WeatherLookupPort>>,
+    pub weather_store: Arc<dyn EntryDimensionPort>,
+}
+
+pub async fn work_the_queue(held: QueueDependencies, worker: WorkerConfig) {
+    let queue = held.queue.clone();
+    let deps = run_due_jobs::Deps {
+        queue: held.queue,
+        backfill: held.backfill,
+        recordings: held.recordings,
+        places: held.places,
+        weather: held.weather,
+        weather_store: held.weather_store,
+    };
+
+    let mut ticks = every(worker.poll_seconds);
+
+    loop {
+        ticks.tick().await;
+
+        match queue.reclaim_stalled(worker.stalled_after_seconds).await {
+            Ok(reclaimed) if reclaimed > 0 => {
+                tracing::info!(reclaimed, "reclaimed jobs left running by a stopped worker")
+            }
+            Ok(_) => {}
+            Err(error) => tracing::error!(%error, "could not reclaim stalled jobs"),
+        }
+
+        match run_due_jobs::execute(worker.jobs_per_poll, worker.most_attempts, &deps).await {
+            Ok(worked) if worked.finished + worked.failed > 0 => {
+                tracing::info!(
+                    finished = worked.finished,
+                    failed = worked.failed,
+                    "worked the queue"
+                )
+            }
+            Ok(_) => {}
+            Err(error) => tracing::error!(%error, "could not work the queue"),
+        }
+    }
+}
+
+pub async fn sweep_for_work(
+    recordings: Arc<dyn RecordingBackfillQueryPort>,
+    places: Option<Arc<dyn WeatherBacklogQueryPort>>,
+    queue: Arc<dyn JobQueueCommandPort>,
+    worker: WorkerConfig,
+) {
+    let for_recordings = sweep_recording_backlog::Deps {
+        backlog: recordings,
+        queue: queue.clone(),
+    };
+
+    let for_weather = places.map(|backlog| sweep_weather_backlog::Deps { backlog, queue });
+
+    if for_weather.is_none() {
+        tracing::info!("not sweeping for weather, because weather cannot be looked up");
+    }
+
+    let mut ticks = every(worker.sweep_seconds);
+
+    loop {
+        ticks.tick().await;
+
+        if let Err(error) =
+            sweep_recording_backlog::execute(worker.enqueued_per_sweep, &for_recordings).await
+        {
+            tracing::error!(%error, "could not sweep for songs with no recording identity");
+        }
+
+        let Some(for_weather) = &for_weather else {
+            continue;
+        };
+
+        if let Err(error) =
+            sweep_weather_backlog::execute(worker.enqueued_per_sweep, for_weather).await
+        {
+            tracing::error!(%error, "could not sweep for places with no weather");
+        }
+    }
+}
+
+pub async fn send_due_reminders(
+    reminder_query: Arc<dyn ReminderQueryPort>,
+    user_query: Arc<dyn UserQueryPort>,
+    sender: Arc<dyn ReminderSenderPort>,
+    worker: WorkerConfig,
+) {
+    let deps = process_due_reminders::Deps {
+        reminder_query,
+        user_query,
+        sender,
+    };
+
+    let mut ticks = every(worker.reminder_seconds);
+
+    loop {
+        ticks.tick().await;
+
+        match process_due_reminders::execute(&deps).await {
+            Ok(sent) if sent > 0 => tracing::info!(sent, "reminders sent"),
+            Ok(_) => {}
+            Err(error) => tracing::error!(%error, "could not process due reminders"),
+        }
+    }
+}
+
+pub async fn clear_expired_sessions(
+    sessions: Arc<dyn RefreshSessionCommandPort>,
+    worker: WorkerConfig,
+) {
+    let mut ticks = every(worker.session_cleanup_seconds);
+
+    loop {
+        ticks.tick().await;
+
+        match sessions.delete_expired().await {
+            Ok(deleted) if deleted > 0 => tracing::info!(deleted, "expired sessions cleared"),
+            Ok(_) => {}
+            Err(error) => tracing::error!(%error, "could not clear expired sessions"),
+        }
+    }
+}

crates/worker/src/main.rs

@@ -0,0 +1,102 @@
+mod loops;
+
+#[tokio::main]
+async fn main() {
+    bootstrap::setup_tracing();
+
+    if let Err(error) = run().await {
+        tracing::error!(%error, "the worker could not start");
+        std::process::exit(1);
+    }
+}
+
+async fn run() -> Result<(), Box<dyn std::error::Error>> {
+    let config = bootstrap::load()?;
+    let context = bootstrap::build(config).await?;
+    let worker = context.worker_config.clone();
+
+    tracing::info!(
+        poll_seconds = worker.poll_seconds,
+        sweep_seconds = worker.sweep_seconds,
+        most_attempts = worker.most_attempts,
+        "k-mood worker started"
+    );
+
+    report_abandoned_work(&context).await;
+
+    let queue = tokio::spawn(loops::work_the_queue(
+        loops::QueueDependencies {
+            queue: context.job_queue.clone(),
+            backfill: context.recording_backlog.clone(),
+            recordings: context.recordings.clone(),
+            places: context.weather_backlog.clone(),
+            weather: context.weather.clone(),
+            weather_store: context.weather_store.clone(),
+        },
+        worker.clone(),
+    ));
+
+    let sweeper = tokio::spawn(loops::sweep_for_work(
+        context.recording_backlog.clone(),
+        context
+            .weather
+            .as_ref()
+            .map(|_| context.weather_backlog.clone()),
+        context.job_queue.clone(),
+        worker.clone(),
+    ));
+
+    let sessions = tokio::spawn(loops::clear_expired_sessions(
+        context.state.refresh_session_command.clone(),
+        worker.clone(),
+    ));
+
+    let reminders = context.reminder_sender.clone().map(|sender| {
+        tokio::spawn(loops::send_due_reminders(
+            context.state.reminder_query.clone(),
+            context.state.user_query.clone(),
+            sender,
+            worker,
+        ))
+    });
+
+    if reminders.is_none() {
+        tracing::info!("push notifications are not configured, so no reminders will be sent");
+    }
+
+    bootstrap::shutdown::on_signal().await;
+
+    queue.abort();
+    sweeper.abort();
+    sessions.abort();
+    if let Some(reminders) = reminders {
+        reminders.abort();
+    }
+
+    tracing::info!("k-mood worker shut down");
+
+    Ok(())
+}
+
+async fn report_abandoned_work(context: &bootstrap::Context) {
+    match context.job_view.find_exhausted(50).await {
+        Ok(abandoned) if !abandoned.is_empty() => {
+            tracing::warn!(
+                count = abandoned.len(),
+                "jobs were given up on and are waiting to be looked at"
+            );
+
+            for job in abandoned {
+                tracing::warn!(
+                    job_id = %job.id(),
+                    kind = job.kind().name(),
+                    attempts = job.attempts(),
+                    reason = job.last_error().unwrap_or("unknown"),
+                    "abandoned job"
+                );
+            }
+        }
+        Ok(_) => {}
+        Err(error) => tracing::error!(%error, "could not read abandoned jobs"),
+    }
+}