apalis_core/backend/
expose.rs

1use std::str::FromStr;
2
3use crate::{
4    backend::{Backend, TaskSink, WireFormatBackend},
5    task::{Task, status::Status},
6};
7
8const DEFAULT_PAGE_SIZE: u32 = 10;
9/// Allows exposing additional functionality from the backend
10pub trait Expose<Args, Kind> {}
11
12impl<B, Args, Kind> Expose<Args, Kind> for B where
13    B: Backend
14        + Metrics
15        + ListWorkers
16        + ListQueues
17        + ListAllTasks
18        + ListTasks
19        + TaskSink<Args, Kind>
20{
21}
22
23/// Allows listing all queues available in the backend
24pub trait ListQueues: Backend {
25    /// List all available queues in the backend
26    fn list_queues(&self) -> impl Future<Output = Result<Vec<QueueInfo>, Self::Error>> + Send;
27}
28
29/// Allows listing all workers registered with the backend
30pub trait ListWorkers: Backend {
31    /// List all registered workers in the current queue
32    fn list_workers(&self) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send;
33
34    /// List all registered workers in all queues
35    fn list_all_workers(
36        &self,
37    ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send;
38}
39/// Allows listing tasks with optional filtering
40pub trait ListTasks: WireFormatBackend + Backend {
41    /// List tasks matching the given filter in the current queue
42    #[allow(clippy::type_complexity)]
43    fn list_tasks(
44        &self,
45        filter: &Filter,
46    ) -> impl Future<Output = Result<Vec<Task<Self::Compact>>, Self::Error>> + Send;
47}
48
49/// Allows listing tasks across all queues with optional filtering
50pub trait ListAllTasks: WireFormatBackend + Backend {
51    /// List tasks matching the given filter in all queues
52    #[allow(clippy::type_complexity)]
53    fn list_all_tasks(
54        &self,
55        filter: &Filter,
56    ) -> impl Future<Output = Result<Vec<Task<Self::Compact>>, Self::Error>> + Send;
57}
58
59/// Allows collecting metrics from the backend
60pub trait Metrics: Backend {
61    /// Collects and returns global statistics from the backend
62    fn global(&self) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send;
63
64    /// Collects and returns statistics for a specific queue
65    fn fetch_by_queue(&self) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send;
66}
67
68/// Represents information about a specific queue in the backend
69#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
70#[derive(Debug, Clone)]
71pub struct QueueInfo {
72    /// Name of the queue
73    pub name: String,
74    /// Statistics related to the queue
75    pub stats: Vec<Statistic>,
76    /// List of workers associated with the queue
77    pub workers: Vec<String>,
78    /// Last 7 days of activity in the queue
79    pub activity: Vec<usize>,
80}
81
82/// Represents a worker currently registered with the backend
83#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
84#[derive(Debug, Clone)]
85pub struct RunningWorker {
86    /// Unique identifier for the worker
87    pub id: String,
88    /// Queue the worker is processing tasks from
89    pub queue: String,
90    /// Backend of the worker
91    pub backend: String,
92    /// Timestamp when the worker was started
93    pub started_at: u64,
94    /// Timestamp of the last heartbeat received from the worker
95    pub last_heartbeat: u64,
96    /// Layers the worker is associated with
97    pub layers: String,
98}
99
100#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
101#[derive(Debug, Clone)]
102/// Filter criteria for listing tasks
103pub struct Filter {
104    /// Optional status to filter tasks by
105    #[cfg_attr(feature = "serde", serde(default))]
106    pub status: Option<Status>,
107    #[cfg_attr(feature = "serde", serde(default = "default_page"))]
108    /// Page number for pagination (default is 1)
109    pub page: u32,
110    /// Optional page size for pagination (default is 10)
111    #[cfg_attr(feature = "serde", serde(default))]
112    pub page_size: Option<u32>,
113}
114
115impl Filter {
116    /// Calculate the offset based on the current page and page size
117    #[must_use]
118    pub fn offset(&self) -> u32 {
119        (self.page - 1) * self.page_size.unwrap_or(DEFAULT_PAGE_SIZE)
120    }
121
122    /// Get the limit (page size) for the query
123    #[must_use]
124    pub fn limit(&self) -> u32 {
125        self.page_size.unwrap_or(DEFAULT_PAGE_SIZE)
126    }
127}
128
129#[cfg(feature = "serde")]
130fn default_page() -> u32 {
131    1
132}
133/// Represents an overview of the backend including queues, workers, and statistics
134#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
135#[derive(Debug, Clone)]
136pub struct Statistic {
137    /// Overall statistics of the backend
138    pub title: String,
139    /// The statistics type
140    pub stat_type: StatType,
141    /// The value of the statistic
142    pub value: String,
143    /// The priority of the statistic (lower number means higher priority)
144    pub priority: Option<u64>,
145}
146/// Statistics type
147#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
148#[derive(Debug, Clone, PartialEq, Eq, Default)]
149#[non_exhaustive]
150pub enum StatType {
151    /// Timestamp statistic
152    Timestamp,
153    /// Numeric statistic
154    #[default]
155    Number,
156    /// Decimal statistic
157    Decimal,
158    /// Percentage statistic
159    Percentage,
160}
161
162/// Error type for parsing StatType from a string
163#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
164#[non_exhaustive]
165pub enum ParseStatTypeError {
166    /// Error for invalid statistic type string
167    #[error("invalid stat type: `{0}`")]
168    InvalidStatType(String),
169}
170
171impl FromStr for StatType {
172    type Err = ParseStatTypeError;
173
174    fn from_str(s: &str) -> Result<Self, Self::Err> {
175        match s {
176            "Timestamp" => Ok(Self::Timestamp),
177            "Decimal" => Ok(Self::Decimal),
178            "Percentage" => Ok(Self::Percentage),
179            "Number" => Ok(Self::Number),
180            _ => Err(ParseStatTypeError::InvalidStatType(s.to_owned())),
181        }
182    }
183}
apalis_core/backend/expose.rs

apalis_core/backend/
expose.rs
