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 { summary } = undo
261        {
262            record_tui_undo(
263                &mut tx,
264                &workspace.id,
265                &summary,
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
