aven_core/operations/
recurrence.rs

1use anyhow::{Context, Result, ensure};
2use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
3use sqlx::{Row, SqliteConnection};
4
5#[cfg(test)]
6use crate::choices::{TaskPriority, TaskStatus};
7use crate::db::{
8    Database, begin_immediate, entity_conflict_exists, recurrence_occurrence_from_row,
9    recurrence_series_from_row,
10};
11use crate::error::CoreError;
12use crate::ids::{TaskId, WorkspaceId, now, now_utc};
13#[cfg(test)]
14use crate::recurrence::RecurrenceProjectionState;
15#[cfg(test)]
16use crate::recurrence::RecurrenceSeriesState;
17use crate::recurrence::{
18    RecurrenceDuePolicy, RecurrenceOutcome, RecurrenceSchedule, RecurrenceSeriesId,
19    SERIES_REF_PREFIX, recurrence_series_display_ref,
20};
21use crate::refs::get_task_in_workspace;
22#[cfg(test)]
23use crate::task_fields::TaskField;
24use crate::types::{MutableEntityType, RecurrenceOccurrence, RecurrenceSeries, Task};
25use crate::undo::{UndoCommand, UndoContext, UndoPayload, record_tui_undo};
26use crate::workspaces::Workspace;
27
28#[cfg(test)]
29#[path = "recurrence_tests.rs"]
30mod tests;
31
32mod lifecycle;
33mod projection;
34mod resolution;
35mod template;
36
37pub use lifecycle::{pause_recurrence_series, resume_recurrence_series, stop_recurrence_series};
38pub(crate) use lifecycle::{
39    stop_recurrence_series_for_project_delete_in_transaction, timestamp_strictly_after,
40};
41pub(crate) use projection::reconcile_recurrence_series_in_transaction;
42use projection::{materialize_occurrence, verify_materialized_occurrence};
43use projection::{reconcile_recurrence_series_once, retryable_reconcile_error};
44pub(crate) use resolution::resolve_recurrence_occurrence_in_transaction;
45pub(crate) use resolution::{
46    RecurrenceMutationOutcome, RecurrenceStructuralMutation, RecurrenceTaskMutation,
47    route_recurrence_task_mutation, undo_recurrence_resolution,
48    validate_consumer_recurrence_mutation,
49};
50pub use template::{create_recurrence_series, update_recurrence_template};
51
52const RECONCILE_ATTEMPTS: usize = 3;
53const SERIES_TEMPLATE_FIELDS: &[&str] = &[
54    "title",
55    "description",
56    "project",
57    "priority",
58    "initial_status",
59    "labels",
60    "available_local_time",
61    "due_policy",
62    "state",
63    "stopped_at",
64    "deleted",
65];
66
67#[derive(Debug, Clone)]
68pub struct RecurrenceSeriesDraft {
69    pub title: String,
70    pub description: String,
71    pub project: String,
72    pub priority: String,
73    pub initial_status: String,
74    pub labels: Vec<String>,
75    pub metadata: Vec<crate::metadata::TaskMetadataInput>,
76    pub schedule: RecurrenceSchedule,
77}
78
79#[derive(Debug, Clone)]
80pub struct CreateRecurrenceSeriesParams {
81    pub draft: RecurrenceSeriesDraft,
82    at: Option<DateTime<Utc>>,
83    create_missing_labels: bool,
84}
85
86impl CreateRecurrenceSeriesParams {
87    pub fn new(draft: RecurrenceSeriesDraft) -> Self {
88        Self {
89            draft,
90            at: None,
91            create_missing_labels: false,
92        }
93    }
94
95    pub fn at(mut self, at: DateTime<Utc>) -> Self {
96        self.at = Some(at);
97        self
98    }
99
100    pub fn with_create_missing_labels(mut self) -> Self {
101        self.create_missing_labels = true;
102        self
103    }
104
105    fn resolve_at_with(mut self, clock: impl FnOnce() -> Result<DateTime<Utc>>) -> Result<Self> {
106        if self.at.is_none() {
107            self.at = Some(clock()?);
108        }
109        Ok(self)
110    }
111}
112
113#[derive(Debug, Clone, Default)]
114pub struct RecurrenceTemplateUpdate {
115    pub title: Option<String>,
116    pub description: Option<String>,
117    pub project: Option<String>,
118    pub priority: Option<String>,
119    pub initial_status: Option<String>,
120    pub labels: Option<Vec<String>>,
121    pub set_metadata: Vec<crate::metadata::TaskMetadataInput>,
122    pub remove_metadata: Vec<String>,
123    pub available_local_time: Option<Option<NaiveTime>>,
124    pub due_policy: Option<RecurrenceDuePolicy>,
125}
126
127#[derive(Debug, Clone)]
128pub struct UpdateRecurrenceTemplateParams {
129    pub update: RecurrenceTemplateUpdate,
130    create_missing_labels: bool,
131}
132
133impl UpdateRecurrenceTemplateParams {
134    pub fn new(update: RecurrenceTemplateUpdate) -> Self {
135        Self {
136            update,
137            create_missing_labels: false,
138        }
139    }
140
141    pub fn with_create_missing_labels(mut self) -> Self {
142        self.create_missing_labels = true;
143        self
144    }
145}
146
147#[derive(Debug, Clone)]
148pub struct RecurrenceCreateOutcome {
149    pub series: RecurrenceSeries,
150    pub series_ref: String,
151    pub occurrence: RecurrenceOccurrence,
152    pub task: Task,
153}
154
155#[derive(Debug, Clone)]
156pub struct RecurrenceTemplateUpdateOutcome {
157    pub series: RecurrenceSeries,
158    pub changed: bool,
159}
160
161#[derive(Debug, Clone)]
162pub struct RecurrenceReconcileOutcome {
163    pub series: RecurrenceSeries,
164    pub occurrence: Option<RecurrenceOccurrence>,
165    pub changed: bool,
166    pub lifecycle_blocked: bool,
167}
168
169#[derive(Debug, Clone)]
170pub struct RecurrenceResolveOutcome {
171    pub series: RecurrenceSeries,
172    pub resolved: RecurrenceOccurrence,
173    pub task: Task,
174    pub successor: Option<Task>,
175}
176
177#[derive(Debug, Clone)]
178pub struct RecurrenceStateOutcome {
179    pub series: RecurrenceSeries,
180    pub occurrence: Option<RecurrenceOccurrence>,
181}
182
183impl Database {
184    pub async fn create_recurrence_series(
185        &self,
186        workspace: &Workspace,
187        params: CreateRecurrenceSeriesParams,
188    ) -> Result<RecurrenceCreateOutcome> {
189        self.create_recurrence_series_with_clock(workspace, params, utc_now)
190            .await
191    }
192
193    async fn create_recurrence_series_with_clock(
194        &self,
195        workspace: &Workspace,
196        params: CreateRecurrenceSeriesParams,
197        clock: impl FnOnce() -> Result<DateTime<Utc>>,
198    ) -> Result<RecurrenceCreateOutcome> {
199        let params = params.resolve_at_with(clock)?;
200        let mut conn = self.acquire_writer().await?;
201        create_recurrence_series(&mut conn, workspace, params).await
202    }
203
204    pub async fn update_recurrence_template(
205        &self,
206        workspace: &Workspace,
207        series_id: &RecurrenceSeriesId,
208        params: UpdateRecurrenceTemplateParams,
209    ) -> Result<RecurrenceTemplateUpdateOutcome> {
210        let mut conn = self.acquire_writer().await?;
211        update_recurrence_template(&mut conn, workspace, series_id, params).await
212    }
213
214    pub async fn reconcile_recurrence_series(
215        &self,
216        workspace: &Workspace,
217        series_id: &RecurrenceSeriesId,
218        at: DateTime<Utc>,
219    ) -> Result<RecurrenceReconcileOutcome> {
220        let mut conn = self.acquire_writer().await?;
221        let mut last_error = None;
222        for _ in 0..RECONCILE_ATTEMPTS {
223            match reconcile_recurrence_series_once(&mut conn, workspace, series_id, at).await {
224                Ok(outcome) => return Ok(outcome),
225                Err(error) if retryable_reconcile_error(&error) => last_error = Some(error),
226                Err(error) => return Err(error),
227            }
228        }
229        Err(last_error.expect("bounded reconciliation records retry errors"))
230    }
231
232    pub async fn resolve_recurrence_occurrence(
233        &self,
234        workspace: &Workspace,
235        task_id: &TaskId,
236        outcome: RecurrenceOutcome,
237    ) -> Result<RecurrenceResolveOutcome> {
238        self.resolve_recurrence_occurrence_with_undo(workspace, task_id, outcome, UndoContext::None)
239            .await
240    }
241
242    pub async fn resolve_recurrence_occurrence_with_undo(
243        &self,
244        workspace: &Workspace,
245        task_id: &TaskId,
246        outcome: RecurrenceOutcome,
247        undo: UndoContext,
248    ) -> Result<RecurrenceResolveOutcome> {
249        let mut conn = self.acquire_writer().await?;
250        let mut tx = begin_immediate(&mut conn).await?;
251        let before = get_task_in_workspace(&mut tx, workspace, task_id).await?;
252        let result = resolve_recurrence_occurrence_in_transaction(
253            &mut tx,
254            workspace,
255            task_id,
256            outcome,
257            &now(),
258        )
259        .await?;
260        if before.status != result.task.status
261            && let UndoContext::Tui = undo
262        {
263            record_tui_undo(
264                &mut tx,
265                &workspace.id,
266                UndoPayload {
267                    commands: vec![UndoCommand::SetTaskField {
268                        task_id: task_id.clone(),
269                        field: "status".to_string(),
270                        before: before.status.as_str().to_string(),
271                        after: result.task.status.as_str().to_string(),
272                        queue_activity_before: Some(before.queue_activity_at.clone()),
273                        queue_activity_after: Some(result.task.queue_activity_at.clone()),
274                    }],
275                },
276            )
277            .await?;
278        }
279        tx.commit().await?;
280        Ok(result)
281    }
282
283    pub async fn pause_recurrence_series(
284        &self,
285        workspace: &Workspace,
286        series_id: &RecurrenceSeriesId,
287    ) -> Result<RecurrenceStateOutcome> {
288        let mut conn = self.acquire_writer().await?;
289        pause_recurrence_series(&mut conn, workspace, series_id, &now()).await
290    }
291
292    pub async fn resume_recurrence_series(
293        &self,
294        workspace: &Workspace,
295        series_id: &RecurrenceSeriesId,
296        at: DateTime<Utc>,
297    ) -> Result<RecurrenceStateOutcome> {
298        let mut conn = self.acquire_writer().await?;
299        resume_recurrence_series(&mut conn, workspace, series_id, at).await
300    }
301
302    pub async fn stop_recurrence_series(
303        &self,
304        workspace: &Workspace,
305        series_id: &RecurrenceSeriesId,
306        skip_current: bool,
307    ) -> Result<RecurrenceStateOutcome> {
308        let mut conn = self.acquire_writer().await?;
309        stop_recurrence_series(&mut conn, workspace, series_id, skip_current, &now()).await
310    }
311
312    pub async fn resolve_recurrence_ref(
313        &self,
314        workspace: &Workspace,
315        input: &str,
316    ) -> Result<RecurrenceSeries> {
317        let mut conn = self.acquire_reader().await?;
318        resolve_recurrence_ref(&mut conn, workspace, input).await
319    }
320
321    pub async fn recurrence_series_ref(
322        &self,
323        workspace_id: &WorkspaceId,
324        series_id: &RecurrenceSeriesId,
325    ) -> Result<String> {
326        let mut conn = self.acquire_reader().await?;
327        recurrence_series_ref(&mut conn, workspace_id, series_id).await
328    }
329}
330
331async fn load_series(
332    conn: &mut SqliteConnection,
333    workspace_id: &WorkspaceId,
334    series_id: &RecurrenceSeriesId,
335) -> Result<RecurrenceSeries> {
336    let row = sqlx::query(
337        "SELECT workspace_id, id, title, description, project_id, priority, initial_status,
338                frequency, interval, weekdays, timezone, start_on, available_local_time,
339                due_policy, state, stopped_at, created_at, updated_at, deleted
340         FROM recurrence_series WHERE workspace_id = ? AND id = ?",
341    )
342    .bind(workspace_id)
343    .bind(series_id)
344    .fetch_optional(&mut *conn)
345    .await?
346    .ok_or_else(|| {
347        CoreError::not_found(format!(
348            "error recurrence-series-not-found series_id={series_id}"
349        ))
350    })?;
351    recurrence_series_from_row(&row)
352}
353
354async fn load_series_labels(
355    conn: &mut SqliteConnection,
356    workspace_id: &WorkspaceId,
357    series_id: &RecurrenceSeriesId,
358) -> Result<Vec<String>> {
359    Ok(sqlx::query_scalar(
360        "SELECT label FROM recurrence_series_labels
361         WHERE workspace_id = ? AND series_id = ? ORDER BY label",
362    )
363    .bind(workspace_id)
364    .bind(series_id)
365    .fetch_all(&mut *conn)
366    .await?)
367}
368
369async fn load_series_metadata(
370    conn: &mut SqliteConnection,
371    workspace_id: &WorkspaceId,
372    series_id: &RecurrenceSeriesId,
373) -> Result<Vec<crate::metadata::ResolvedMetadataValue>> {
374    let rows = sqlx::query(
375        "SELECT m.field_id, f.key, m.value
376         FROM recurrence_series_metadata m
377         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
378         WHERE m.workspace_id = ? AND m.series_id = ? ORDER BY f.key",
379    )
380    .bind(workspace_id)
381    .bind(series_id)
382    .fetch_all(&mut *conn)
383    .await?;
384    Ok(rows
385        .into_iter()
386        .map(|row| crate::metadata::ResolvedMetadataValue {
387            field_id: row.get("field_id"),
388            key: row.get("key"),
389            value: row.get("value"),
390        })
391        .collect())
392}
393
394async fn load_occurrence(
395    conn: &mut SqliteConnection,
396    workspace_id: &WorkspaceId,
397    series_id: &RecurrenceSeriesId,
398    slot_on: NaiveDate,
399) -> Result<Option<RecurrenceOccurrence>> {
400    let row = sqlx::query(
401        "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at,
402                outcome_change_id, projection_state, archived_at
403         FROM recurrence_occurrences
404         WHERE workspace_id = ? AND series_id = ? AND slot_on = ?",
405    )
406    .bind(workspace_id)
407    .bind(series_id)
408    .bind(slot_on.format("%Y-%m-%d").to_string())
409    .fetch_optional(&mut *conn)
410    .await?;
411    row.as_ref().map(recurrence_occurrence_from_row).transpose()
412}
413
414async fn load_projected_occurrence(
415    conn: &mut SqliteConnection,
416    workspace_id: &WorkspaceId,
417    series_id: &RecurrenceSeriesId,
418) -> Result<Option<RecurrenceOccurrence>> {
419    let row = sqlx::query(
420        "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at,
421                outcome_change_id, projection_state, archived_at
422         FROM recurrence_occurrences
423         WHERE workspace_id = ? AND series_id = ? AND projection_state = 'projected'",
424    )
425    .bind(workspace_id)
426    .bind(series_id)
427    .fetch_optional(&mut *conn)
428    .await?;
429    row.as_ref().map(recurrence_occurrence_from_row).transpose()
430}
431
432async fn load_occurrence_for_task(
433    conn: &mut SqliteConnection,
434    workspace_id: &WorkspaceId,
435    task_id: &TaskId,
436) -> Result<Option<RecurrenceOccurrence>> {
437    let row = sqlx::query(
438        "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at,
439                outcome_change_id, projection_state, archived_at
440         FROM recurrence_occurrences WHERE workspace_id = ? AND task_id = ?",
441    )
442    .bind(workspace_id)
443    .bind(task_id)
444    .fetch_optional(&mut *conn)
445    .await?;
446    row.as_ref().map(recurrence_occurrence_from_row).transpose()
447}
448
449async fn lifecycle_conflict_exists(
450    conn: &mut SqliteConnection,
451    workspace_id: &WorkspaceId,
452    series_id: &RecurrenceSeriesId,
453) -> Result<bool> {
454    for field in ["state", "stopped_at"] {
455        if entity_conflict_exists(
456            conn,
457            workspace_id,
458            MutableEntityType::RecurrenceSeries,
459            series_id.as_str(),
460            field,
461        )
462        .await?
463        {
464            return Ok(true);
465        }
466    }
467    Ok(false)
468}
469
470async fn ensure_no_series_conflict(
471    conn: &mut SqliteConnection,
472    workspace_id: &WorkspaceId,
473    series_id: &RecurrenceSeriesId,
474    field: &str,
475) -> Result<()> {
476    ensure!(
477        !entity_conflict_exists(
478            conn,
479            workspace_id,
480            MutableEntityType::RecurrenceSeries,
481            series_id.as_str(),
482            field,
483        )
484        .await?,
485        CoreError::open_conflict(format!(
486            "error recurrence-conflicted-field series_id={series_id} field={field}"
487        ))
488    );
489    Ok(())
490}
491
492pub async fn resolve_recurrence_ref(
493    conn: &mut SqliteConnection,
494    workspace: &Workspace,
495    input: &str,
496) -> Result<RecurrenceSeries> {
497    let (hint, suffix) = split_ref(input);
498    if suffix.len() < 3 {
499        return Err(CoreError::validation(format!(
500            "error recurrence-ref-too-short input={input} minimum=3"
501        ))
502        .into());
503    }
504    let series_rows = sqlx::query(
505        "SELECT workspace_id, id, title, description, project_id, priority, initial_status,
506                frequency, interval, weekdays, timezone, start_on, available_local_time,
507                due_policy, state, stopped_at, created_at, updated_at, deleted
508         FROM recurrence_series WHERE workspace_id = ? AND id LIKE ? || '%' ORDER BY id",
509    )
510    .bind(&workspace.id)
511    .bind(&suffix)
512    .fetch_all(&mut *conn)
513    .await?;
514    let mut series = series_rows
515        .iter()
516        .map(recurrence_series_from_row)
517        .collect::<Result<Vec<_>>>()?;
518    if hint
519        .as_deref()
520        .is_some_and(|value| value != SERIES_REF_PREFIX)
521    {
522        series.clear();
523    }
524    if series.len() == 1 {
525        return Ok(series.remove(0));
526    }
527    if series.len() > 1 {
528        return Err(
529            CoreError::validation(format!("error ambiguous-recurrence-ref input={input}")).into(),
530        );
531    }
532    let task = crate::refs::resolve_task_ref_in_workspace(conn, workspace, input).await?;
533    let occurrence = load_occurrence_for_task(conn, &workspace.id, &task.id)
534        .await?
535        .context("error task-is-not-recurring")?;
536    load_series(conn, &workspace.id, &occurrence.series_id).await
537}
538
539async fn recurrence_series_ref(
540    conn: &mut SqliteConnection,
541    workspace_id: &WorkspaceId,
542    series_id: &RecurrenceSeriesId,
543) -> Result<String> {
544    let ids = sqlx::query_scalar::<_, RecurrenceSeriesId>(
545        "SELECT id FROM recurrence_series WHERE workspace_id = ? ORDER BY id",
546    )
547    .bind(workspace_id)
548    .fetch_all(&mut *conn)
549    .await?;
550    ensure!(
551        ids.iter().any(|candidate| candidate == series_id),
552        CoreError::not_found(format!(
553            "error recurrence-series-not-found series_id={series_id}"
554        ))
555    );
556    Ok(recurrence_series_display_ref(series_id, &ids))
557}
558
559fn split_ref(input: &str) -> (Option<String>, String) {
560    let (hint, suffix) = input
561        .split_once('-')
562        .map_or((None, input), |(hint, suffix)| (Some(hint), suffix));
563    let suffix = suffix
564        .chars()
565        .filter(|value| value.is_ascii_alphanumeric())
566        .map(|value| match value.to_ascii_uppercase() {
567            'O' => '0',
568            'I' | 'L' => '1',
569            value => value,
570        })
571        .collect();
572    (hint.map(|value| value.to_ascii_uppercase()), suffix)
573}
574
575fn format_local_time(value: Option<NaiveTime>) -> String {
576    value
577        .map(|value| value.format("%H:%M:%S").to_string())
578        .unwrap_or_default()
579}
580
581fn format_utc(value: DateTime<Utc>) -> String {
582    value.format("%Y-%m-%dT%H:%M:%SZ").to_string()
583}
584
585fn utc_now() -> Result<DateTime<Utc>> {
586    Ok(now_utc())
587}
aven_core/operations/recurrence.rs

aven_core/operations/
recurrence.rs
