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

aven_core/
query.rs
