aven_core/
query.rs

1use anyhow::{Result, bail};
2use chrono::{DateTime, Utc};
3
4use crate::choices::TaskStatus;
5use crate::db::Database;
6use crate::ids::{TaskId, WorkspaceId};
7use crate::recurrence::RecurrenceSeriesId;
8use crate::refs::DisplayRefContext;
9use crate::types::Task;
10
11mod database_stats;
12mod dependencies;
13mod details;
14mod doctor;
15pub mod fragments;
16mod hydration;
17mod projects;
18mod recent_actions;
19mod recurrence;
20mod related;
21mod search;
22mod sidebar;
23mod sorting;
24mod sync_history;
25mod tasks;
26#[cfg(test)]
27mod test_support;
28mod types;
29
30pub use database_stats::{DatabaseStats, DatabaseStatsPriorityCounts, DatabaseStatsStatusCounts};
31pub use dependencies::{TaskDependencyItem, TaskDependencySummary};
32pub(crate) use dependencies::{task_dependency_summary, task_dependency_summary_with_display_refs};
33pub use details::{TaskDetail, TaskDetailConflict};
34pub(crate) use details::{conflict_display_value, task_detail, task_detail_with_display_refs};
35pub use doctor::WorkspaceTaskCounts;
36pub(crate) use doctor::{
37    unresolved_conflict_count, unresolved_conflict_count_in_workspace, workspace_task_counts,
38};
39pub(crate) use projects::list_project_items_in_workspace;
40pub(crate) use recent_actions::{
41    list_recent_actions_in_workspace, task_activity_for_tasks_in_workspace,
42};
43pub(crate) use recurrence::task_recurrence_summaries;
44pub use related::TaskRelatedLink;
45pub(crate) use related::{related_links_for_tasks, task_related_links};
46pub use search::{
47    SearchMatchedField, TaskSearchPreviewResultSet, TaskSearchQuery, TaskSearchResult,
48};
49pub(crate) use search::{
50    search_task_items_in_workspace, search_task_occurrence_items_in_workspace,
51    search_task_preview_set_in_workspace,
52};
53pub(crate) use sidebar::sidebar_counts_for_scope_in_workspace;
54pub use sync_history::SyncHistoryStats;
55pub(crate) use sync_history::sync_history_stats;
56pub(crate) use tasks::{
57    list_bulk_update_task_items_in_workspace, list_task_items_in_workspace,
58    list_task_items_with_display_refs, list_task_items_without_activity_in_workspace,
59    list_task_summary_items_in_workspace,
60};
61pub use types::RecentActionTarget;
62pub use types::{
63    AttachmentMetadata, EpicRollup, MetadataFilter, ProjectListItem, RecentActionItem,
64    RecurrenceCounts, RecurrenceHistoryEntry, RecurrenceHistoryKind, RecurrenceHistoryPage,
65    RecurrenceOccurrenceLink, RecurrenceReconciliation, RecurrenceSeriesConflict,
66    RecurrenceSeriesDetail, RecurrenceSeriesLifecycleFilter, RecurrenceSeriesListItem,
67    RecurrenceSeriesListQuery, RecurrenceSeriesSummary, RecurrenceTaskGroup, SidebarCounts,
68    SortDirection, TaskAvailabilityFilter, TaskDependencyLink, TaskFilters, TaskIdFilter,
69    TaskItemHydration, TaskListItem, TaskNote, TaskQueryMode, TaskRecurrenceSummary, TaskSort,
70};
71
72pub const MAX_RECURRENCE_HISTORY_LIMIT: usize = 500;
73
74pub(crate) fn status_order(status: TaskStatus) -> u8 {
75    match status {
76        TaskStatus::Active => 0,
77        TaskStatus::Todo => 1,
78        TaskStatus::Inbox => 2,
79        TaskStatus::Backlog => 3,
80        TaskStatus::Done => 4,
81        TaskStatus::Canceled => 5,
82    }
83}
84
85fn validate_search_limit(limit: usize) -> Result<()> {
86    if limit == 0 {
87        bail!("error search-limit-invalid limit=0 min=1 hint=\"pass a limit of 1 or greater\"");
88    }
89    Ok(())
90}
91
92fn validate_recurrence_history_limit(limit: usize) -> Result<()> {
93    if !(1..=MAX_RECURRENCE_HISTORY_LIMIT).contains(&limit) {
94        bail!(
95            "error recurrence-history-limit-invalid limit={} min=1 max={} hint=\"pass a limit between 1 and {}\"",
96            limit,
97            MAX_RECURRENCE_HISTORY_LIMIT,
98            MAX_RECURRENCE_HISTORY_LIMIT
99        );
100    }
101    Ok(())
102}
103
104impl Database {
105    /// Reconciles recurrence projections before a report reads current state.
106    ///
107    /// The candidate count is bounded. Each changed series writes one atomic projection
108    /// transaction that can archive one superseded task and materialize one successor.
109    pub async fn reconcile_recurrence_reports_at(
110        &self,
111        workspace_id: &WorkspaceId,
112        at: DateTime<Utc>,
113    ) -> Result<RecurrenceReconciliation> {
114        let (workspace, candidates, incomplete) = {
115            let mut conn = self.acquire_reader().await?;
116            let workspace = crate::workspaces::workspace_for_id(&mut conn, workspace_id).await?;
117            let (candidates, incomplete) =
118                recurrence::recurrence_reconciliation_candidates(&mut conn, workspace_id, at)
119                    .await?;
120            (workspace, candidates, incomplete)
121        };
122        let mut report = RecurrenceReconciliation {
123            workspace_id: Some(workspace_id.clone()),
124            examined: candidates.len(),
125            incomplete,
126            ..RecurrenceReconciliation::default()
127        };
128        for series_id in candidates {
129            let result = self
130                .reconcile_recurrence_series(&workspace, &series_id, at)
131                .await?;
132            report.changed += usize::from(result.changed);
133            report.lifecycle_blocked += usize::from(result.lifecycle_blocked);
134        }
135        recurrence::validate_reconciliation(&report)?;
136        Ok(report)
137    }
138
139    pub async fn reconcile_recurrence_reports(
140        &self,
141        workspace_id: &WorkspaceId,
142    ) -> Result<RecurrenceReconciliation> {
143        let at = crate::ids::now_utc();
144        self.reconcile_recurrence_reports_at(workspace_id, at).await
145    }
146
147    pub async fn list_recurrence_series_view_at(
148        &self,
149        workspace_id: &WorkspaceId,
150        at: DateTime<Utc>,
151        query: RecurrenceSeriesListQuery,
152    ) -> Result<Vec<RecurrenceSeriesListItem>> {
153        self.reconcile_recurrence_reports_at(workspace_id, at)
154            .await?;
155        self.list_recurrence_series_view_from_current_projection(workspace_id, query)
156            .await
157    }
158
159    /// Reads the persisted recurrence projection as-is.
160    ///
161    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
162    #[doc(hidden)]
163    pub async fn list_recurrence_series_view_from_current_projection(
164        &self,
165        workspace_id: &WorkspaceId,
166        query: RecurrenceSeriesListQuery,
167    ) -> Result<Vec<RecurrenceSeriesListItem>> {
168        let mut conn = self.acquire_reader().await?;
169        recurrence::list_recurrence_series_view(&mut conn, workspace_id, &query).await
170    }
171
172    pub async fn list_recurrence_series_view(
173        &self,
174        workspace_id: &WorkspaceId,
175        query: RecurrenceSeriesListQuery,
176    ) -> Result<Vec<RecurrenceSeriesListItem>> {
177        let at = crate::ids::now_utc();
178        self.list_recurrence_series_view_at(workspace_id, at, query)
179            .await
180    }
181
182    pub async fn list_recurrence_series_at(
183        &self,
184        workspace_id: &WorkspaceId,
185        at: DateTime<Utc>,
186    ) -> Result<Vec<RecurrenceSeriesSummary>> {
187        self.reconcile_recurrence_reports_at(workspace_id, at)
188            .await?;
189        let mut conn = self.acquire_reader().await?;
190        recurrence::list_recurrence_series(&mut conn, workspace_id, at).await
191    }
192
193    pub async fn recurrence_series_detail_at(
194        &self,
195        workspace_id: &WorkspaceId,
196        series_id: &RecurrenceSeriesId,
197        at: DateTime<Utc>,
198    ) -> Result<RecurrenceSeriesDetail> {
199        self.reconcile_recurrence_reports_at(workspace_id, at)
200            .await?;
201        self.recurrence_series_detail_at_from_current_projection(workspace_id, series_id, at)
202            .await
203    }
204
205    /// Reads the persisted recurrence projection as-is.
206    ///
207    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
208    #[doc(hidden)]
209    pub async fn recurrence_series_detail_at_from_current_projection(
210        &self,
211        workspace_id: &WorkspaceId,
212        series_id: &RecurrenceSeriesId,
213        at: DateTime<Utc>,
214    ) -> Result<RecurrenceSeriesDetail> {
215        let mut conn = self.acquire_reader().await?;
216        recurrence::recurrence_series_detail(&mut conn, workspace_id, series_id, at).await
217    }
218
219    pub async fn recurrence_history_at(
220        &self,
221        workspace_id: &WorkspaceId,
222        series_id: &RecurrenceSeriesId,
223        at: DateTime<Utc>,
224        offset: usize,
225        limit: usize,
226    ) -> Result<RecurrenceHistoryPage> {
227        validate_recurrence_history_limit(limit)?;
228        self.reconcile_recurrence_reports_at(workspace_id, at)
229            .await?;
230        let mut conn = self.acquire_reader().await?;
231        recurrence::recurrence_history(&mut conn, workspace_id, series_id, at, offset, limit).await
232    }
233
234    pub async fn list_recurrence_series(
235        &self,
236        workspace_id: &WorkspaceId,
237    ) -> Result<Vec<RecurrenceSeriesSummary>> {
238        let at = crate::ids::now_utc();
239        self.list_recurrence_series_at(workspace_id, at).await
240    }
241
242    pub async fn recurrence_series_detail(
243        &self,
244        workspace_id: &WorkspaceId,
245        series_id: &RecurrenceSeriesId,
246    ) -> Result<RecurrenceSeriesDetail> {
247        let at = crate::ids::now_utc();
248        self.recurrence_series_detail_at(workspace_id, series_id, at)
249            .await
250    }
251
252    /// Reads the persisted recurrence projection as-is.
253    ///
254    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
255    #[doc(hidden)]
256    pub async fn recurrence_series_detail_from_current_projection(
257        &self,
258        workspace_id: &WorkspaceId,
259        series_id: &RecurrenceSeriesId,
260    ) -> Result<RecurrenceSeriesDetail> {
261        let at = crate::ids::now_utc();
262        self.recurrence_series_detail_at_from_current_projection(workspace_id, series_id, at)
263            .await
264    }
265
266    pub async fn recurrence_history(
267        &self,
268        workspace_id: &WorkspaceId,
269        series_id: &RecurrenceSeriesId,
270        offset: usize,
271        limit: usize,
272    ) -> Result<RecurrenceHistoryPage> {
273        let at = crate::ids::now_utc();
274        self.recurrence_history_at(workspace_id, series_id, at, offset, limit)
275            .await
276    }
277
278    pub async fn list_project_items(
279        &self,
280        workspace_id: &WorkspaceId,
281    ) -> Result<Vec<ProjectListItem>> {
282        self.reconcile_recurrence_reports(workspace_id).await?;
283        self.list_project_items_from_current_projection(workspace_id)
284            .await
285    }
286
287    /// Reads the persisted recurrence projection as-is.
288    ///
289    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
290    #[doc(hidden)]
291    pub async fn list_project_items_from_current_projection(
292        &self,
293        workspace_id: &WorkspaceId,
294    ) -> Result<Vec<ProjectListItem>> {
295        let mut conn = self.acquire_reader().await?;
296        list_project_items_in_workspace(&mut conn, workspace_id).await
297    }
298
299    pub(crate) async fn list_base_tasks(
300        &self,
301        workspace_id: &WorkspaceId,
302        filters: TaskFilters,
303        sort: TaskSort,
304        direction: SortDirection,
305    ) -> Result<Vec<Task>> {
306        self.reconcile_recurrence_reports(workspace_id).await?;
307        let mut conn = self.acquire_reader().await?;
308        tasks::list_base_tasks_in_workspace(&mut conn, workspace_id, filters, sort, direction, None)
309            .await
310    }
311
312    pub async fn list_task_items(
313        &self,
314        workspace_id: &WorkspaceId,
315        filters: TaskFilters,
316        mode: TaskQueryMode,
317        sort: TaskSort,
318        direction: SortDirection,
319    ) -> Result<Vec<TaskListItem>> {
320        self.reconcile_recurrence_reports(workspace_id).await?;
321        self.list_task_items_from_current_projection(workspace_id, filters, mode, sort, direction)
322            .await
323    }
324
325    pub async fn list_bulk_update_task_items(
326        &self,
327        workspace_id: &WorkspaceId,
328        filters: TaskFilters,
329        mode: TaskQueryMode,
330        sort: TaskSort,
331        direction: SortDirection,
332    ) -> Result<Vec<TaskListItem>> {
333        self.reconcile_recurrence_reports(workspace_id).await?;
334        let mut conn = self.acquire_reader().await?;
335        list_bulk_update_task_items_in_workspace(
336            &mut conn,
337            workspace_id,
338            filters,
339            mode,
340            sort,
341            direction,
342        )
343        .await
344    }
345
346    /// Reads the persisted recurrence projection as-is.
347    ///
348    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
349    #[doc(hidden)]
350    pub async fn list_task_items_from_current_projection(
351        &self,
352        workspace_id: &WorkspaceId,
353        filters: TaskFilters,
354        mode: TaskQueryMode,
355        sort: TaskSort,
356        direction: SortDirection,
357    ) -> Result<Vec<TaskListItem>> {
358        let mut conn = self.acquire_reader().await?;
359        list_task_items_in_workspace(&mut conn, workspace_id, filters, mode, sort, direction).await
360    }
361
362    pub async fn list_task_items_without_activity_from_current_projection(
363        &self,
364        workspace_id: &WorkspaceId,
365        filters: TaskFilters,
366        mode: TaskQueryMode,
367        sort: TaskSort,
368        direction: SortDirection,
369    ) -> Result<Vec<TaskListItem>> {
370        let mut conn = self.acquire_reader().await?;
371        list_task_items_without_activity_in_workspace(
372            &mut conn,
373            workspace_id,
374            filters,
375            mode,
376            sort,
377            direction,
378        )
379        .await
380    }
381
382    pub async fn list_task_summary_items(
383        &self,
384        workspace_id: &WorkspaceId,
385        filters: TaskFilters,
386        mode: TaskQueryMode,
387        sort: TaskSort,
388        direction: SortDirection,
389        limit: Option<usize>,
390    ) -> Result<Vec<TaskListItem>> {
391        self.reconcile_recurrence_reports(workspace_id).await?;
392        self.list_task_summary_items_from_current_projection(
393            workspace_id,
394            filters,
395            mode,
396            sort,
397            direction,
398            limit,
399        )
400        .await
401    }
402
403    /// Reads the persisted task projection as-is.
404    ///
405    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
406    #[doc(hidden)]
407    pub async fn list_task_summary_items_from_current_projection(
408        &self,
409        workspace_id: &WorkspaceId,
410        filters: TaskFilters,
411        mode: TaskQueryMode,
412        sort: TaskSort,
413        direction: SortDirection,
414        limit: Option<usize>,
415    ) -> Result<Vec<TaskListItem>> {
416        let mut conn = self.acquire_reader().await?;
417        list_task_summary_items_in_workspace(
418            &mut conn,
419            workspace_id,
420            filters,
421            mode,
422            sort,
423            direction,
424            limit,
425        )
426        .await
427    }
428
429    pub async fn sidebar_counts_for_scope(
430        &self,
431        workspace_id: &WorkspaceId,
432        project_key: Option<&str>,
433    ) -> Result<SidebarCounts> {
434        self.reconcile_recurrence_reports(workspace_id).await?;
435        self.sidebar_counts_for_scope_from_current_projection(workspace_id, project_key)
436            .await
437    }
438
439    /// Reads the persisted recurrence projection as-is.
440    ///
441    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
442    #[doc(hidden)]
443    pub async fn sidebar_counts_for_scope_from_current_projection(
444        &self,
445        workspace_id: &WorkspaceId,
446        project_key: Option<&str>,
447    ) -> Result<SidebarCounts> {
448        let mut conn = self.acquire_reader().await?;
449        sidebar_counts_for_scope_in_workspace(&mut conn, workspace_id, project_key).await
450    }
451
452    pub async fn list_recent_actions(
453        &self,
454        workspace_id: &WorkspaceId,
455        project_scope: Option<&str>,
456    ) -> Result<Vec<RecentActionItem>> {
457        self.reconcile_recurrence_reports(workspace_id).await?;
458        self.list_recent_actions_from_current_projection(workspace_id, project_scope)
459            .await
460    }
461
462    /// Reads the persisted recurrence projection as-is.
463    ///
464    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
465    #[doc(hidden)]
466    pub async fn list_recent_actions_from_current_projection(
467        &self,
468        workspace_id: &WorkspaceId,
469        project_scope: Option<&str>,
470    ) -> Result<Vec<RecentActionItem>> {
471        let mut conn = self.acquire_reader().await?;
472        list_recent_actions_in_workspace(&mut conn, workspace_id, project_scope).await
473    }
474
475    pub async fn task_activity_for_task(
476        &self,
477        workspace_id: &WorkspaceId,
478        task_id: &TaskId,
479    ) -> Result<Vec<RecentActionItem>> {
480        let mut conn = self.acquire_reader().await?;
481        Ok(task_activity_for_tasks_in_workspace(
482            &mut conn,
483            workspace_id,
484            std::slice::from_ref(task_id),
485        )
486        .await?
487        .remove(task_id)
488        .unwrap_or_default())
489    }
490
491    pub async fn search_task_items(
492        &self,
493        workspace_id: &WorkspaceId,
494        query: TaskSearchQuery,
495    ) -> Result<Vec<TaskSearchResult>> {
496        validate_search_limit(query.limit)?;
497        self.reconcile_recurrence_reports(workspace_id).await?;
498        let mut conn = self.acquire_reader().await?;
499        search_task_items_in_workspace(&mut conn, workspace_id, query).await
500    }
501
502    pub async fn search_task_occurrence_items(
503        &self,
504        workspace_id: &WorkspaceId,
505        query: TaskSearchQuery,
506    ) -> Result<Vec<TaskSearchResult>> {
507        validate_search_limit(query.limit)?;
508        self.reconcile_recurrence_reports(workspace_id).await?;
509        let mut conn = self.acquire_reader().await?;
510        search_task_occurrence_items_in_workspace(&mut conn, workspace_id, query).await
511    }
512
513    pub async fn search_task_preview_set(
514        &self,
515        workspace_id: &WorkspaceId,
516        query: TaskSearchQuery,
517    ) -> Result<TaskSearchPreviewResultSet> {
518        validate_search_limit(query.limit)?;
519        self.reconcile_recurrence_reports(workspace_id).await?;
520        self.search_task_preview_set_from_current_projection(workspace_id, query)
521            .await
522    }
523
524    /// Reads the persisted recurrence projection as-is.
525    ///
526    /// Callers establish projection currency by reconciling the workspace for the report snapshot.
527    #[doc(hidden)]
528    pub async fn search_task_preview_set_from_current_projection(
529        &self,
530        workspace_id: &WorkspaceId,
531        query: TaskSearchQuery,
532    ) -> Result<TaskSearchPreviewResultSet> {
533        validate_search_limit(query.limit)?;
534        let mut conn = self.acquire_reader().await?;
535        search_task_preview_set_in_workspace(&mut conn, workspace_id, query).await
536    }
537
538    pub async fn task_detail(&self, task: &Task) -> Result<TaskDetail> {
539        self.reconcile_recurrence_reports(&task.workspace_id)
540            .await?;
541        let mut conn = self.acquire_reader().await?;
542        task_detail(&mut conn, task).await
543    }
544
545    pub async fn task_detail_with_display_refs(
546        &self,
547        task: &Task,
548        display_refs: &DisplayRefContext,
549    ) -> Result<TaskDetail> {
550        self.reconcile_recurrence_reports(&task.workspace_id)
551            .await?;
552        let mut conn = self.acquire_reader().await?;
553        task_detail_with_display_refs(&mut conn, task, display_refs).await
554    }
555
556    pub async fn conflict_display_value(
557        &self,
558        workspace_id: &WorkspaceId,
559        field: &str,
560        value: &str,
561    ) -> Result<String> {
562        let mut conn = self.acquire_reader().await?;
563        conflict_display_value(&mut conn, workspace_id, field, value).await
564    }
565
566    pub async fn task_dependency_summary(
567        &self,
568        workspace_id: &WorkspaceId,
569        task_id: &TaskId,
570    ) -> Result<TaskDependencySummary> {
571        let mut conn = self.acquire_reader().await?;
572        task_dependency_summary(&mut conn, workspace_id, task_id).await
573    }
574
575    pub async fn task_related_links(
576        &self,
577        workspace_id: &WorkspaceId,
578        task_id: &TaskId,
579    ) -> Result<Vec<TaskRelatedLink>> {
580        let mut conn = self.acquire_reader().await?;
581        task_related_links(&mut conn, workspace_id, task_id, None).await
582    }
583
584    pub async fn workspace_task_counts(
585        &self,
586        workspace_id: &WorkspaceId,
587    ) -> Result<WorkspaceTaskCounts> {
588        let mut conn = self.acquire_reader().await?;
589        workspace_task_counts(&mut conn, workspace_id).await
590    }
591
592    pub async fn unresolved_conflict_count(&self) -> Result<i64> {
593        let mut conn = self.acquire_reader().await?;
594        unresolved_conflict_count(&mut conn).await
595    }
596
597    pub async fn unresolved_conflict_count_in_workspace(
598        &self,
599        workspace_id: &WorkspaceId,
600    ) -> Result<i64> {
601        let mut conn = self.acquire_reader().await?;
602        unresolved_conflict_count_in_workspace(&mut conn, workspace_id).await
603    }
604
605    pub async fn sync_history_stats(&self) -> Result<SyncHistoryStats> {
606        let mut conn = self.acquire_reader().await?;
607        sync_history_stats(&mut conn).await
608    }
609
610    pub async fn database_stats(
611        &self,
612        workspace: &crate::workspaces::Workspace,
613    ) -> Result<DatabaseStats> {
614        let mut conn = self.acquire_reader().await?;
615        database_stats::database_stats(&mut conn, workspace).await
616    }
617}
aven_core/query.rs

aven_core/
query.rs
