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

aven_core/operations/
recurrence.rs
