1use std::collections::BTreeMap;
2
3use anyhow::{Context, Result, bail, ensure};
4use chrono::{DateTime, NaiveDate, NaiveTime, Utc};
5use serde_json::Value;
6use sqlx::{Row, SqliteConnection};
7
8use crate::change_log::{ChangeEntity, ChangePayload, append_change, op_type};
9use crate::choices::{TaskPriority, TaskStatus};
10use crate::db::{
11 Database, IdentifiedChange, begin_immediate, entity_conflict_exists, entity_field_version,
12 insert_change, insert_change_with_identity, recurrence_occurrence_from_row,
13 recurrence_series_from_row, set_entity_field_version, set_field_version,
14};
15use crate::error::CoreError;
16use crate::ids::{TaskId, WorkspaceId, new_id, now};
17use crate::labels::{resolve_labels_in_workspace, resolve_or_create_labels_in_workspace};
18use crate::mutation::apply_field_value_in_workspace;
19use crate::projects::resolve_or_create_project_in_workspace;
20use crate::recurrence::{
21 RecurrenceDuePolicy, RecurrenceOutcome, RecurrenceProjectionState, RecurrenceSchedule,
22 RecurrenceSeriesId, RecurrenceSeriesState, derive_occurrence_identity, next_slot_after,
23 projection_slot_at, recurrence_series_display_ref, slot_cutoff, slot_values,
24};
25use crate::refs::get_task_in_workspace;
26use crate::task_fields::TaskField;
27use crate::types::{MutableEntityType, RecurrenceOccurrence, RecurrenceSeries, Task};
28use crate::undo::{UndoCommand, UndoContext, UndoPayload, record_tui_undo};
29use crate::workspaces::Workspace;
30
31#[cfg(test)]
32#[path = "recurrence_tests.rs"]
33mod tests;
34
35const RECONCILE_ATTEMPTS: usize = 3;
36const SERIES_REF_PREFIX: &str = "RCR";
37const SERIES_TEMPLATE_FIELDS: &[&str] = &[
38 "title",
39 "description",
40 "project",
41 "priority",
42 "initial_status",
43 "labels",
44 "available_local_time",
45 "due_policy",
46 "state",
47 "stopped_at",
48 "deleted",
49];
50
51#[derive(Debug, Clone)]
52pub struct RecurrenceSeriesDraft {
53 pub title: String,
54 pub description: String,
55 pub project: String,
56 pub priority: String,
57 pub initial_status: String,
58 pub labels: Vec<String>,
59 pub schedule: RecurrenceSchedule,
60}
61
62#[derive(Debug, Clone)]
63pub struct CreateRecurrenceSeriesParams {
64 pub draft: RecurrenceSeriesDraft,
65 at: Option<DateTime<Utc>>,
66 create_missing_labels: bool,
67}
68
69impl CreateRecurrenceSeriesParams {
70 pub fn new(draft: RecurrenceSeriesDraft) -> Self {
71 Self {
72 draft,
73 at: None,
74 create_missing_labels: false,
75 }
76 }
77
78 pub fn at(mut self, at: DateTime<Utc>) -> Self {
79 self.at = Some(at);
80 self
81 }
82
83 pub fn with_create_missing_labels(mut self) -> Self {
84 self.create_missing_labels = true;
85 self
86 }
87
88 fn resolve_at_with(mut self, clock: impl FnOnce() -> Result<DateTime<Utc>>) -> Result<Self> {
89 if self.at.is_none() {
90 self.at = Some(clock()?);
91 }
92 Ok(self)
93 }
94}
95
96#[derive(Debug, Clone, Default)]
97pub struct RecurrenceTemplateUpdate {
98 pub title: Option<String>,
99 pub description: Option<String>,
100 pub project: Option<String>,
101 pub priority: Option<String>,
102 pub initial_status: Option<String>,
103 pub labels: Option<Vec<String>>,
104 pub available_local_time: Option<Option<NaiveTime>>,
105 pub due_policy: Option<RecurrenceDuePolicy>,
106}
107
108#[derive(Debug, Clone)]
109pub struct UpdateRecurrenceTemplateParams {
110 pub update: RecurrenceTemplateUpdate,
111 create_missing_labels: bool,
112}
113
114impl UpdateRecurrenceTemplateParams {
115 pub fn new(update: RecurrenceTemplateUpdate) -> Self {
116 Self {
117 update,
118 create_missing_labels: false,
119 }
120 }
121
122 pub fn with_create_missing_labels(mut self) -> Self {
123 self.create_missing_labels = true;
124 self
125 }
126}
127
128#[derive(Debug, Clone)]
129pub struct RecurrenceCreateOutcome {
130 pub series: RecurrenceSeries,
131 pub series_ref: String,
132 pub occurrence: RecurrenceOccurrence,
133 pub task: Task,
134}
135
136#[derive(Debug, Clone)]
137pub struct RecurrenceTemplateUpdateOutcome {
138 pub series: RecurrenceSeries,
139 pub changed: bool,
140}
141
142#[derive(Debug, Clone)]
143pub struct RecurrenceReconcileOutcome {
144 pub series: RecurrenceSeries,
145 pub occurrence: Option<RecurrenceOccurrence>,
146 pub changed: bool,
147 pub lifecycle_blocked: bool,
148}
149
150#[derive(Debug, Clone)]
151pub struct RecurrenceResolveOutcome {
152 pub series: RecurrenceSeries,
153 pub resolved: RecurrenceOccurrence,
154 pub task: Task,
155 pub successor: Option<Task>,
156}
157
158#[derive(Debug, Clone)]
159pub struct RecurrenceStateOutcome {
160 pub series: RecurrenceSeries,
161 pub occurrence: Option<RecurrenceOccurrence>,
162}
163
164impl Database {
165 pub async fn create_recurrence_series(
166 &self,
167 workspace: &Workspace,
168 params: CreateRecurrenceSeriesParams,
169 ) -> Result<RecurrenceCreateOutcome> {
170 self.create_recurrence_series_with_clock(workspace, params, utc_now)
171 .await
172 }
173
174 async fn create_recurrence_series_with_clock(
175 &self,
176 workspace: &Workspace,
177 params: CreateRecurrenceSeriesParams,
178 clock: impl FnOnce() -> Result<DateTime<Utc>>,
179 ) -> Result<RecurrenceCreateOutcome> {
180 let params = params.resolve_at_with(clock)?;
181 let mut conn = self.acquire_writer().await?;
182 create_recurrence_series(&mut conn, workspace, params).await
183 }
184
185 pub async fn update_recurrence_template(
186 &self,
187 workspace: &Workspace,
188 series_id: &RecurrenceSeriesId,
189 params: UpdateRecurrenceTemplateParams,
190 ) -> Result<RecurrenceTemplateUpdateOutcome> {
191 let mut conn = self.acquire_writer().await?;
192 update_recurrence_template(&mut conn, workspace, series_id, params).await
193 }
194
195 pub async fn reconcile_recurrence_series(
196 &self,
197 workspace: &Workspace,
198 series_id: &RecurrenceSeriesId,
199 at: DateTime<Utc>,
200 ) -> Result<RecurrenceReconcileOutcome> {
201 let mut conn = self.acquire_writer().await?;
202 let mut last_error = None;
203 for _ in 0..RECONCILE_ATTEMPTS {
204 match reconcile_recurrence_series_once(&mut conn, workspace, series_id, at).await {
205 Ok(outcome) => return Ok(outcome),
206 Err(error) if retryable_reconcile_error(&error) => last_error = Some(error),
207 Err(error) => return Err(error),
208 }
209 }
210 Err(last_error.expect("bounded reconciliation records retry errors"))
211 }
212
213 pub async fn resolve_recurrence_occurrence(
214 &self,
215 workspace: &Workspace,
216 task_id: &TaskId,
217 outcome: RecurrenceOutcome,
218 ) -> Result<RecurrenceResolveOutcome> {
219 self.resolve_recurrence_occurrence_with_undo(workspace, task_id, outcome, UndoContext::None)
220 .await
221 }
222
223 pub async fn resolve_recurrence_occurrence_with_undo(
224 &self,
225 workspace: &Workspace,
226 task_id: &TaskId,
227 outcome: RecurrenceOutcome,
228 undo: UndoContext,
229 ) -> Result<RecurrenceResolveOutcome> {
230 let mut conn = self.acquire_writer().await?;
231 let mut tx = begin_immediate(&mut conn).await?;
232 let before = get_task_in_workspace(&mut tx, workspace, task_id).await?;
233 let result = resolve_recurrence_occurrence_in_transaction(
234 &mut tx,
235 workspace,
236 task_id,
237 outcome,
238 &now(),
239 )
240 .await?;
241 if before.status != result.task.status
242 && let UndoContext::Tui { summary } = undo
243 {
244 record_tui_undo(
245 &mut tx,
246 &workspace.id,
247 &summary,
248 UndoPayload {
249 commands: vec![UndoCommand::SetTaskField {
250 task_id: task_id.clone(),
251 field: "status".to_string(),
252 before: before.status.as_str().to_string(),
253 after: result.task.status.as_str().to_string(),
254 }],
255 },
256 )
257 .await?;
258 }
259 tx.commit().await?;
260 Ok(result)
261 }
262
263 pub async fn pause_recurrence_series(
264 &self,
265 workspace: &Workspace,
266 series_id: &RecurrenceSeriesId,
267 ) -> Result<RecurrenceStateOutcome> {
268 let mut conn = self.acquire_writer().await?;
269 pause_recurrence_series(&mut conn, workspace, series_id, &now()).await
270 }
271
272 pub async fn resume_recurrence_series(
273 &self,
274 workspace: &Workspace,
275 series_id: &RecurrenceSeriesId,
276 at: DateTime<Utc>,
277 ) -> Result<RecurrenceStateOutcome> {
278 let mut conn = self.acquire_writer().await?;
279 resume_recurrence_series(&mut conn, workspace, series_id, at).await
280 }
281
282 pub async fn stop_recurrence_series(
283 &self,
284 workspace: &Workspace,
285 series_id: &RecurrenceSeriesId,
286 skip_current: bool,
287 ) -> Result<RecurrenceStateOutcome> {
288 let mut conn = self.acquire_writer().await?;
289 stop_recurrence_series(&mut conn, workspace, series_id, skip_current, &now()).await
290 }
291
292 pub async fn resolve_recurrence_ref(
293 &self,
294 workspace: &Workspace,
295 input: &str,
296 ) -> Result<RecurrenceSeries> {
297 let mut conn = self.acquire_reader().await?;
298 resolve_recurrence_ref(&mut conn, workspace, input).await
299 }
300
301 pub async fn recurrence_series_ref(
302 &self,
303 workspace_id: &WorkspaceId,
304 series_id: &RecurrenceSeriesId,
305 ) -> Result<String> {
306 let mut conn = self.acquire_reader().await?;
307 recurrence_series_ref(&mut conn, workspace_id, series_id).await
308 }
309}
310
311pub async fn create_recurrence_series(
312 conn: &mut SqliteConnection,
313 workspace: &Workspace,
314 params: CreateRecurrenceSeriesParams,
315) -> Result<RecurrenceCreateOutcome> {
316 let CreateRecurrenceSeriesParams {
317 draft,
318 at,
319 create_missing_labels,
320 } = params;
321 let at = at.map_or_else(utc_now, Ok)?;
322 let priority = TaskPriority::parse(&draft.priority)?;
323 let initial_status = TaskStatus::parse(&draft.initial_status)?;
324 ensure!(
325 initial_status.is_open(),
326 CoreError::validation(format!(
327 "error recurrence-initial-status-terminal status={}",
328 initial_status.as_str()
329 ))
330 );
331 let creation_date = at
332 .with_timezone(&draft.schedule.timezone.timezone())
333 .date_naive();
334 let first_slot = draft
335 .schedule
336 .slots_on_or_after(draft.schedule.start_on.max(creation_date))
337 .next()
338 .context("recurrence schedule has no representable first slot")?;
339 let series_id = RecurrenceSeriesId::new();
340 let created_at = format_utc(at);
341
342 let mut tx = begin_immediate(conn).await?;
343 let project =
344 resolve_or_create_project_in_workspace(&mut tx, &workspace.id, draft.project.as_str())
345 .await?;
346 let labels = if create_missing_labels {
347 resolve_or_create_labels_in_workspace(&mut tx, workspace, &draft.labels)
348 .await?
349 .names
350 } else {
351 resolve_labels_in_workspace(&mut tx, &workspace.id, &draft.labels).await?
352 };
353 sqlx::query(
354 "INSERT INTO recurrence_series(
355 workspace_id, id, title, description, project_id, priority, initial_status,
356 frequency, interval, weekdays, timezone, start_on, available_local_time,
357 due_policy, state, stopped_at, created_at, updated_at, deleted
358 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', '', ?, ?, 0)",
359 )
360 .bind(&workspace.id)
361 .bind(&series_id)
362 .bind(&draft.title)
363 .bind(&draft.description)
364 .bind(&project.id)
365 .bind(priority.as_str())
366 .bind(initial_status.as_str())
367 .bind(draft.schedule.rule.frequency().as_str())
368 .bind(i64::from(draft.schedule.rule.interval()))
369 .bind(draft.schedule.rule.weekdays_set().to_string())
370 .bind(draft.schedule.timezone.as_str())
371 .bind(draft.schedule.start_on.format("%Y-%m-%d").to_string())
372 .bind(format_local_time(draft.schedule.available_local_time))
373 .bind(draft.schedule.due_policy.as_str())
374 .bind(&created_at)
375 .bind(&created_at)
376 .execute(&mut *tx)
377 .await?;
378 for label in &labels {
379 sqlx::query(
380 "INSERT INTO recurrence_series_labels(workspace_id, series_id, label)
381 VALUES (?, ?, ?)",
382 )
383 .bind(&workspace.id)
384 .bind(&series_id)
385 .bind(label)
386 .execute(&mut *tx)
387 .await?;
388 }
389 let create_change_id = append_change(
390 &mut tx,
391 ChangeEntity::RecurrenceSeries,
392 series_id.as_str(),
393 None,
394 op_type::CREATE_RECURRENCE_SERIES,
395 ChangePayload::workspace(workspace)
396 .set("series_id", series_id.as_str())
397 .set("title", &draft.title)
398 .set("description", &draft.description)
399 .set("project_id", project.id.as_str())
400 .set("project_key", &project.key)
401 .set("project_name", &project.name)
402 .set("project_prefix", &project.prefix)
403 .set("priority", priority.as_str())
404 .set("initial_status", initial_status.as_str())
405 .set("frequency", draft.schedule.rule.frequency().as_str())
406 .set("interval", draft.schedule.rule.interval())
407 .set("weekdays", draft.schedule.rule.weekdays_set().to_string())
408 .set("timezone", draft.schedule.timezone.as_str())
409 .set(
410 "start_on",
411 draft.schedule.start_on.format("%Y-%m-%d").to_string(),
412 )
413 .set(
414 "available_local_time",
415 format_local_time(draft.schedule.available_local_time),
416 )
417 .set("due_policy", draft.schedule.due_policy.as_str())
418 .set("labels", &labels)
419 .set("state", "active")
420 .set("stopped_at", "")
421 .set("created_at", &created_at)
422 .set("updated_at", &created_at),
423 )
424 .await?;
425 for field in SERIES_TEMPLATE_FIELDS {
426 set_entity_field_version(
427 &mut tx,
428 &workspace.id,
429 MutableEntityType::RecurrenceSeries,
430 series_id.as_str(),
431 field,
432 &create_change_id,
433 )
434 .await?;
435 }
436 let series = load_series(&mut tx, &workspace.id, &series_id).await?;
437 let occurrence =
438 materialize_occurrence(&mut tx, workspace, &series, &labels, first_slot).await?;
439 let task_id = occurrence
440 .task_id
441 .as_ref()
442 .expect("materialized occurrence has a task")
443 .clone();
444 let task = get_task_in_workspace(&mut tx, workspace, &task_id).await?;
445 let series_ref = recurrence_series_ref(&mut tx, &workspace.id, &series_id).await?;
446 tx.commit().await?;
447 Ok(RecurrenceCreateOutcome {
448 series,
449 series_ref,
450 occurrence,
451 task,
452 })
453}
454
455pub async fn update_recurrence_template(
456 conn: &mut SqliteConnection,
457 workspace: &Workspace,
458 series_id: &RecurrenceSeriesId,
459 params: UpdateRecurrenceTemplateParams,
460) -> Result<RecurrenceTemplateUpdateOutcome> {
461 let UpdateRecurrenceTemplateParams {
462 update,
463 create_missing_labels,
464 } = params;
465 if let Some(priority) = update.priority.as_deref() {
466 TaskPriority::parse(priority)?;
467 }
468 if let Some(status) = update.initial_status.as_deref() {
469 let status = TaskStatus::parse(status)?;
470 ensure!(
471 status.is_open(),
472 CoreError::validation(format!(
473 "error recurrence-initial-status-terminal status={}",
474 status.as_str()
475 ))
476 );
477 }
478
479 let mut tx = begin_immediate(conn).await?;
480 let current = load_series(&mut tx, &workspace.id, series_id).await?;
481 let mut values = Vec::<(&str, String)>::new();
482 if let Some(title) = update.title
483 && title != current.title
484 {
485 values.push(("title", title));
486 }
487 if let Some(description) = update.description
488 && description != current.description
489 {
490 values.push(("description", description));
491 }
492 if let Some(priority) = update.priority
493 && priority != current.priority.as_str()
494 {
495 values.push(("priority", priority));
496 }
497 if let Some(status) = update.initial_status
498 && status != current.initial_status.as_str()
499 {
500 values.push(("initial_status", status));
501 }
502 if let Some(time) = update.available_local_time {
503 let value = format_local_time(time);
504 if value != format_local_time(current.available_local_time) {
505 values.push(("available_local_time", value));
506 }
507 }
508 if let Some(due_policy) = update.due_policy
509 && due_policy != current.due_policy
510 {
511 values.push(("due_policy", due_policy.as_str().to_string()));
512 }
513 if let Some(project) = update.project {
514 let project =
515 resolve_or_create_project_in_workspace(&mut tx, &workspace.id, &project).await?;
516 if project.id != current.project_id {
517 values.push(("project", project.id.to_string()));
518 }
519 }
520
521 let current_labels = load_series_labels(&mut tx, &workspace.id, series_id).await?;
522 let target_labels = if let Some(labels) = update.labels {
523 if create_missing_labels {
524 resolve_or_create_labels_in_workspace(&mut tx, workspace, &labels)
525 .await?
526 .names
527 } else {
528 resolve_labels_in_workspace(&mut tx, &workspace.id, &labels).await?
529 }
530 } else {
531 current_labels.clone()
532 };
533 let labels_changed = current_labels != target_labels;
534 if values.is_empty() && !labels_changed {
535 tx.commit().await?;
536 return Ok(RecurrenceTemplateUpdateOutcome {
537 series: current,
538 changed: false,
539 });
540 }
541
542 for (field, _) in &values {
543 ensure_no_series_conflict(&mut tx, &workspace.id, series_id, field).await?;
544 }
545 if labels_changed {
546 ensure_no_series_conflict(&mut tx, &workspace.id, series_id, "labels").await?;
547 }
548 let mut base_versions = BTreeMap::new();
549 for (field, _) in &values {
550 base_versions.insert(
551 (*field).to_string(),
552 entity_field_version(
553 &mut tx,
554 &workspace.id,
555 MutableEntityType::RecurrenceSeries,
556 series_id.as_str(),
557 field,
558 )
559 .await?,
560 );
561 }
562 if labels_changed {
563 base_versions.insert(
564 "labels".to_string(),
565 entity_field_version(
566 &mut tx,
567 &workspace.id,
568 MutableEntityType::RecurrenceSeries,
569 series_id.as_str(),
570 "labels",
571 )
572 .await?,
573 );
574 }
575 let updated_at = now();
576 for (field, value) in &values {
577 update_series_template_scalar(&mut tx, &workspace.id, series_id, field, value, &updated_at)
578 .await?;
579 }
580 if labels_changed {
581 sqlx::query(
582 "DELETE FROM recurrence_series_labels WHERE workspace_id = ? AND series_id = ?",
583 )
584 .bind(&workspace.id)
585 .bind(series_id)
586 .execute(&mut *tx)
587 .await?;
588 for label in &target_labels {
589 sqlx::query(
590 "INSERT INTO recurrence_series_labels(workspace_id, series_id, label)
591 VALUES (?, ?, ?)",
592 )
593 .bind(&workspace.id)
594 .bind(series_id)
595 .bind(label)
596 .execute(&mut *tx)
597 .await?;
598 }
599 }
600 let change_id = append_change(
601 &mut tx,
602 ChangeEntity::RecurrenceSeries,
603 series_id.as_str(),
604 None,
605 op_type::UPDATE_RECURRENCE_TEMPLATE,
606 ChangePayload::workspace(workspace)
607 .set("fields", &values)
608 .set("base_versions", &base_versions)
609 .set("labels_changed", labels_changed)
610 .set("labels", &target_labels)
611 .set("updated_at", &updated_at),
612 )
613 .await?;
614 for (field, _) in &values {
615 set_entity_field_version(
616 &mut tx,
617 &workspace.id,
618 MutableEntityType::RecurrenceSeries,
619 series_id.as_str(),
620 field,
621 &change_id,
622 )
623 .await?;
624 }
625 if labels_changed {
626 set_entity_field_version(
627 &mut tx,
628 &workspace.id,
629 MutableEntityType::RecurrenceSeries,
630 series_id.as_str(),
631 "labels",
632 &change_id,
633 )
634 .await?;
635 }
636 let series = load_series(&mut tx, &workspace.id, series_id).await?;
637 tx.commit().await?;
638 Ok(RecurrenceTemplateUpdateOutcome {
639 series,
640 changed: true,
641 })
642}
643
644async fn reconcile_recurrence_series_once(
645 conn: &mut SqliteConnection,
646 workspace: &Workspace,
647 series_id: &RecurrenceSeriesId,
648 at: DateTime<Utc>,
649) -> Result<RecurrenceReconcileOutcome> {
650 let mut tx = begin_immediate(conn).await?;
651 let outcome =
652 reconcile_recurrence_series_in_transaction(&mut tx, workspace, series_id, at).await?;
653 tx.commit().await?;
654 Ok(outcome)
655}
656
657pub(crate) async fn reconcile_recurrence_series_in_transaction(
658 conn: &mut SqliteConnection,
659 workspace: &Workspace,
660 series_id: &RecurrenceSeriesId,
661 at: DateTime<Utc>,
662) -> Result<RecurrenceReconcileOutcome> {
663 let series = load_series(conn, &workspace.id, series_id).await?;
664 let projected = load_projected_occurrence(conn, &workspace.id, series_id).await?;
665 if !matches!(series.state, RecurrenceSeriesState::Active) {
666 return Ok(RecurrenceReconcileOutcome {
667 series,
668 occurrence: projected,
669 changed: false,
670 lifecycle_blocked: false,
671 });
672 }
673 if lifecycle_conflict_exists(conn, &workspace.id, series_id).await? {
674 return Ok(RecurrenceReconcileOutcome {
675 series,
676 occurrence: projected,
677 changed: false,
678 lifecycle_blocked: true,
679 });
680 }
681
682 let schedule = series.schedule();
683 let target = projection_slot_at(&schedule, at)?;
684 if projected
685 .as_ref()
686 .is_some_and(|occurrence| occurrence.slot_on >= target)
687 {
688 return Ok(RecurrenceReconcileOutcome {
689 series,
690 occurrence: projected,
691 changed: false,
692 lifecycle_blocked: false,
693 });
694 }
695
696 let changed_at = format_utc(at);
697 if let Some(projected) = projected {
698 sqlx::query(
699 "UPDATE recurrence_occurrences
700 SET projection_state = 'archived', archived_at = ?
701 WHERE workspace_id = ? AND series_id = ? AND slot_on = ?
702 AND projection_state = 'projected'",
703 )
704 .bind(&changed_at)
705 .bind(&workspace.id)
706 .bind(series_id)
707 .bind(projected.slot_on.format("%Y-%m-%d").to_string())
708 .execute(&mut *conn)
709 .await?;
710 }
711 if let Some(existing) = load_occurrence(conn, &workspace.id, series_id, target).await?
712 && matches!(
713 existing.projection_state,
714 RecurrenceProjectionState::Archived
715 )
716 && existing.outcome.is_none()
717 {
718 let expected = derive_occurrence_identity(&workspace.id, series_id, &schedule, target)?;
719 ensure!(
720 existing.task_id.as_ref() == Some(&expected.task_id),
721 CoreError::generation_conflict(format!(
722 "error recurrence-generation-conflict slot={target} field=task_id"
723 ))
724 );
725 let task = get_task_in_workspace(conn, workspace, &expected.task_id).await?;
726 ensure!(
727 task.status.is_open() && !task.deleted,
728 CoreError::generation_conflict(format!(
729 "error recurrence-generation-conflict slot={target} field=task"
730 ))
731 );
732 sqlx::query(
733 "UPDATE recurrence_occurrences
734 SET projection_state = 'projected', archived_at = ''
735 WHERE workspace_id = ? AND series_id = ? AND slot_on = ?
736 AND projection_state = 'archived' AND outcome = ''",
737 )
738 .bind(&workspace.id)
739 .bind(series_id)
740 .bind(target.format("%Y-%m-%d").to_string())
741 .execute(&mut *conn)
742 .await?;
743 let occurrence = load_occurrence(conn, &workspace.id, series_id, target)
744 .await?
745 .context("promoted recurrence occurrence missing")?;
746 return Ok(RecurrenceReconcileOutcome {
747 series,
748 occurrence: Some(occurrence),
749 changed: true,
750 lifecycle_blocked: false,
751 });
752 }
753 let labels = load_series_labels(conn, &workspace.id, series_id).await?;
754 let occurrence = materialize_occurrence(conn, workspace, &series, &labels, target).await?;
755 Ok(RecurrenceReconcileOutcome {
756 series,
757 occurrence: Some(occurrence),
758 changed: true,
759 lifecycle_blocked: false,
760 })
761}
762
763pub(crate) async fn resolve_recurrence_occurrence_in_transaction(
764 conn: &mut SqliteConnection,
765 workspace: &Workspace,
766 task_id: &TaskId,
767 outcome: RecurrenceOutcome,
768 resolved_at: &str,
769) -> Result<RecurrenceResolveOutcome> {
770 let mut occurrence = load_occurrence_for_task(conn, &workspace.id, task_id)
771 .await?
772 .ok_or_else(|| CoreError::not_found("error recurrence-occurrence-not-found"))?;
773 if matches!(
774 occurrence.projection_state,
775 RecurrenceProjectionState::Projected
776 ) {
777 reconcile_recurrence_series_in_transaction(
778 conn,
779 workspace,
780 &occurrence.series_id,
781 DateTime::parse_from_rfc3339(resolved_at)
782 .context("invalid recurrence resolution time")?
783 .with_timezone(&Utc),
784 )
785 .await?;
786 occurrence = load_occurrence_for_task(conn, &workspace.id, task_id)
787 .await?
788 .ok_or_else(|| CoreError::not_found("error recurrence-occurrence-not-found"))?;
789 }
790 ensure!(
791 matches!(
792 occurrence.projection_state,
793 RecurrenceProjectionState::Projected
794 ),
795 CoreError::validation(format!(
796 "error recurrence-occurrence-not-current task_id={task_id}"
797 ))
798 );
799 let task = get_task_in_workspace(conn, workspace, task_id).await?;
800 ensure!(
801 task.status.is_open(),
802 CoreError::validation(format!(
803 "error recurrence-occurrence-terminal task_id={task_id}"
804 ))
805 );
806 let series = load_series(conn, &workspace.id, &occurrence.series_id).await?;
807 let target_status = match outcome {
808 RecurrenceOutcome::Completed => TaskStatus::Done,
809 RecurrenceOutcome::Skipped => TaskStatus::Canceled,
810 };
811 let status_change_id = write_task_status(
812 conn,
813 workspace,
814 task_id,
815 task.status,
816 target_status,
817 resolved_at,
818 )
819 .await?;
820
821 let successor_slot = if matches!(series.state, RecurrenceSeriesState::Active) {
822 Some(
823 next_slot_after(&series.rule, series.start_on, occurrence.slot_on)
824 .context("recurrence schedule has no representable successor")?,
825 )
826 } else {
827 None
828 };
829 let successor_task_id = successor_slot
830 .map(|slot| {
831 derive_occurrence_identity(&workspace.id, &series.id, &series.schedule(), slot)
832 .map(|identity| identity.task_id)
833 })
834 .transpose()?;
835 let outcome_change_id = append_change(
836 conn,
837 ChangeEntity::RecurrenceSeries,
838 series.id.as_str(),
839 Some("outcome"),
840 op_type::RESOLVE_RECURRENCE_OCCURRENCE,
841 ChangePayload::workspace(workspace)
842 .set("slot_on", occurrence.slot_on.format("%Y-%m-%d").to_string())
843 .set("task_id", task_id.as_str())
844 .set("outcome", outcome.as_str())
845 .set("task_status", target_status.as_str())
846 .set("resolved_at", resolved_at)
847 .set("task_status_change_id", &status_change_id)
848 .set(
849 "successor_task_id",
850 successor_task_id.as_ref().map(TaskId::as_str).unwrap_or(""),
851 )
852 .set("frequency", series.rule.frequency().as_str())
853 .set("interval", series.rule.interval())
854 .set("weekdays", series.rule.weekdays_set().to_string())
855 .set("timezone", series.timezone.as_str())
856 .set("start_on", series.start_on.format("%Y-%m-%d").to_string())
857 .set(
858 "available_local_time",
859 format_local_time(series.available_local_time),
860 )
861 .set("due_policy", series.due_policy.as_str()),
862 )
863 .await?;
864 sqlx::query(
865 "UPDATE recurrence_occurrences
866 SET outcome = ?, resolved_at = ?, outcome_change_id = ?, projection_state = 'resolved'
867 WHERE workspace_id = ? AND series_id = ? AND slot_on = ?
868 AND projection_state = 'projected'",
869 )
870 .bind(outcome.as_str())
871 .bind(resolved_at)
872 .bind(&outcome_change_id)
873 .bind(&workspace.id)
874 .bind(&series.id)
875 .bind(occurrence.slot_on.format("%Y-%m-%d").to_string())
876 .execute(&mut *conn)
877 .await?;
878
879 let labels = load_series_labels(conn, &workspace.id, &series.id).await?;
880 let successor = if let Some(next_slot) = successor_slot {
881 let successor_occurrence =
882 materialize_occurrence(conn, workspace, &series, &labels, next_slot).await?;
883 let successor_id = successor_occurrence
884 .task_id
885 .as_ref()
886 .expect("materialized successor has a task");
887 Some(get_task_in_workspace(conn, workspace, successor_id).await?)
888 } else {
889 None
890 };
891 let resolved = load_occurrence(conn, &workspace.id, &series.id, occurrence.slot_on)
892 .await?
893 .expect("resolved occurrence remains stored");
894 let task = get_task_in_workspace(conn, workspace, task_id).await?;
895 Ok(RecurrenceResolveOutcome {
896 series,
897 resolved,
898 task,
899 successor,
900 })
901}
902
903pub async fn pause_recurrence_series(
904 conn: &mut SqliteConnection,
905 workspace: &Workspace,
906 series_id: &RecurrenceSeriesId,
907 paused_at: &str,
908) -> Result<RecurrenceStateOutcome> {
909 let paused_at_utc = DateTime::parse_from_rfc3339(paused_at)
910 .context("invalid recurrence pause time")?
911 .with_timezone(&Utc);
912 let mut tx = begin_immediate(conn).await?;
913 let series = load_series(&mut tx, &workspace.id, series_id).await?;
914 ensure!(
915 matches!(series.state, RecurrenceSeriesState::Active),
916 CoreError::validation(format!(
917 "error recurrence-pause-invalid-state state={}",
918 series.state.as_str()
919 ))
920 );
921 ensure_no_series_conflict(&mut tx, &workspace.id, series_id, "state").await?;
922 reconcile_recurrence_series_in_transaction(&mut tx, workspace, series_id, paused_at_utc)
923 .await?;
924 let projected = load_projected_occurrence(&mut tx, &workspace.id, series_id).await?;
925 let change_id = set_series_state(
926 &mut tx,
927 workspace,
928 series_id,
929 RecurrenceSeriesState::Paused,
930 None,
931 paused_at,
932 op_type::SET_RECURRENCE_STATE,
933 )
934 .await?;
935 let interval_id = new_id();
936 sqlx::query(
937 "INSERT INTO recurrence_pause_intervals(
938 workspace_id, id, series_id, paused_at, resumed_at, suspended_slot_on,
939 suspended_task_id, created_by_change_id, resolved_by_change_id
940 ) VALUES (?, ?, ?, ?, '', ?, ?, ?, '')",
941 )
942 .bind(&workspace.id)
943 .bind(&interval_id)
944 .bind(series_id)
945 .bind(paused_at)
946 .bind(
947 projected
948 .as_ref()
949 .map(|occurrence| occurrence.slot_on.format("%Y-%m-%d").to_string())
950 .unwrap_or_default(),
951 )
952 .bind(
953 projected
954 .as_ref()
955 .and_then(|occurrence| occurrence.task_id.as_ref())
956 .map(TaskId::as_str)
957 .unwrap_or(""),
958 )
959 .bind(&change_id)
960 .execute(&mut *tx)
961 .await?;
962 append_change(
963 &mut tx,
964 ChangeEntity::RecurrenceSeries,
965 series_id.as_str(),
966 Some("pause"),
967 op_type::OPEN_RECURRENCE_PAUSE,
968 ChangePayload::workspace(workspace)
969 .set("interval_id", &interval_id)
970 .set("paused_at", paused_at)
971 .set(
972 "suspended_slot_on",
973 projected
974 .as_ref()
975 .map(|occurrence| occurrence.slot_on.format("%Y-%m-%d").to_string())
976 .unwrap_or_default(),
977 )
978 .set(
979 "suspended_task_id",
980 projected
981 .as_ref()
982 .and_then(|occurrence| occurrence.task_id.as_ref())
983 .map(TaskId::as_str)
984 .unwrap_or(""),
985 ),
986 )
987 .await?;
988 let series = load_series(&mut tx, &workspace.id, series_id).await?;
989 tx.commit().await?;
990 Ok(RecurrenceStateOutcome {
991 series,
992 occurrence: projected,
993 })
994}
995
996pub async fn resume_recurrence_series(
997 conn: &mut SqliteConnection,
998 workspace: &Workspace,
999 series_id: &RecurrenceSeriesId,
1000 at: DateTime<Utc>,
1001) -> Result<RecurrenceStateOutcome> {
1002 let mut resumed_at = format_utc(at);
1003 let mut tx = begin_immediate(conn).await?;
1004 let series = load_series(&mut tx, &workspace.id, series_id).await?;
1005 ensure!(
1006 matches!(series.state, RecurrenceSeriesState::Paused),
1007 CoreError::validation(format!(
1008 "error recurrence-resume-invalid-state state={}",
1009 series.state.as_str()
1010 ))
1011 );
1012 ensure_no_series_conflict(&mut tx, &workspace.id, series_id, "state").await?;
1013 let interval = sqlx::query(
1014 "SELECT id, paused_at, suspended_slot_on, suspended_task_id
1015 FROM recurrence_pause_intervals
1016 WHERE workspace_id = ? AND series_id = ? AND resumed_at = ''",
1017 )
1018 .bind(&workspace.id)
1019 .bind(series_id)
1020 .fetch_optional(&mut *tx)
1021 .await?
1022 .ok_or_else(|| CoreError::validation("error recurrence-open-pause-missing"))?;
1023 let interval_id: String = interval.get("id");
1024 let paused_at: String = interval.get("paused_at");
1025 resumed_at = timestamp_strictly_after(&paused_at, &resumed_at)?;
1026 let suspended_slot_text: String = interval.get("suspended_slot_on");
1027 let suspended_slot = (!suspended_slot_text.is_empty())
1028 .then(|| suspended_slot_text.parse::<NaiveDate>())
1029 .transpose()?;
1030
1031 let close_change_id = append_change(
1032 &mut tx,
1033 ChangeEntity::RecurrenceSeries,
1034 series_id.as_str(),
1035 Some("pause"),
1036 op_type::CLOSE_RECURRENCE_PAUSE,
1037 ChangePayload::workspace(workspace)
1038 .set("interval_id", &interval_id)
1039 .set("paused_at", &paused_at)
1040 .set("resumed_at", &resumed_at),
1041 )
1042 .await?;
1043 sqlx::query(
1044 "UPDATE recurrence_pause_intervals SET resumed_at = ?, resolved_by_change_id = ?
1045 WHERE workspace_id = ? AND id = ? AND resumed_at = ''",
1046 )
1047 .bind(&resumed_at)
1048 .bind(&close_change_id)
1049 .bind(&workspace.id)
1050 .bind(&interval_id)
1051 .execute(&mut *tx)
1052 .await?;
1053 set_series_state(
1054 &mut tx,
1055 workspace,
1056 series_id,
1057 RecurrenceSeriesState::Active,
1058 None,
1059 &resumed_at,
1060 op_type::SET_RECURRENCE_STATE,
1061 )
1062 .await?;
1063
1064 let schedule = series.schedule();
1065 let mut occurrence = load_projected_occurrence(&mut tx, &workspace.id, series_id).await?;
1066 let suspended_still_live = if let Some(suspended_slot) = suspended_slot {
1067 occurrence
1068 .as_ref()
1069 .is_some_and(|value| value.slot_on == suspended_slot)
1070 && at < slot_cutoff(&schedule, suspended_slot)?
1071 } else {
1072 false
1073 };
1074 if occurrence.is_some() && !suspended_still_live {
1075 let slot = occurrence.as_ref().expect("checked occurrence").slot_on;
1076 sqlx::query(
1077 "UPDATE recurrence_occurrences
1078 SET projection_state = 'archived', archived_at = ?
1079 WHERE workspace_id = ? AND series_id = ? AND slot_on = ?
1080 AND projection_state = 'projected'",
1081 )
1082 .bind(&resumed_at)
1083 .bind(&workspace.id)
1084 .bind(series_id)
1085 .bind(slot.format("%Y-%m-%d").to_string())
1086 .execute(&mut *tx)
1087 .await?;
1088 occurrence = None;
1089 }
1090 if occurrence.is_none() {
1091 let mut target = projection_slot_at(&schedule, at)?;
1092 let boundary = slot_values(&schedule, target)?.boundary_at;
1093 if boundary >= paused_at && boundary < resumed_at {
1094 target = next_slot_after(&series.rule, series.start_on, target)
1095 .context("recurrence schedule has no representable resumed slot")?;
1096 }
1097 let labels = load_series_labels(&mut tx, &workspace.id, series_id).await?;
1098 occurrence =
1099 Some(materialize_occurrence(&mut tx, workspace, &series, &labels, target).await?);
1100 }
1101 let series = load_series(&mut tx, &workspace.id, series_id).await?;
1102 tx.commit().await?;
1103 Ok(RecurrenceStateOutcome { series, occurrence })
1104}
1105
1106pub async fn stop_recurrence_series(
1107 conn: &mut SqliteConnection,
1108 workspace: &Workspace,
1109 series_id: &RecurrenceSeriesId,
1110 skip_current: bool,
1111 stopped_at: &str,
1112) -> Result<RecurrenceStateOutcome> {
1113 let mut stopped_at = stopped_at.to_string();
1114 let mut tx = begin_immediate(conn).await?;
1115 let series = load_series(&mut tx, &workspace.id, series_id).await?;
1116 ensure!(
1117 !matches!(series.state, RecurrenceSeriesState::Stopped),
1118 CoreError::validation("error recurrence-already-stopped")
1119 );
1120 ensure_no_series_conflict(&mut tx, &workspace.id, series_id, "state").await?;
1121 let open_pause = if matches!(series.state, RecurrenceSeriesState::Paused) {
1122 let interval = sqlx::query(
1123 "SELECT id, paused_at FROM recurrence_pause_intervals
1124 WHERE workspace_id = ? AND series_id = ? AND resumed_at = ''",
1125 )
1126 .bind(&workspace.id)
1127 .bind(series_id)
1128 .fetch_optional(&mut *tx)
1129 .await?
1130 .ok_or_else(|| CoreError::validation("error recurrence-open-pause-missing"))?;
1131 let interval_id: String = interval.get("id");
1132 let paused_at: String = interval.get("paused_at");
1133 stopped_at = timestamp_strictly_after(&paused_at, &stopped_at)?;
1134 Some((interval_id, paused_at))
1135 } else {
1136 None
1137 };
1138 let stopped_at_utc = DateTime::parse_from_rfc3339(&stopped_at)
1139 .context("invalid recurrence stop time")?
1140 .with_timezone(&Utc);
1141 reconcile_recurrence_series_in_transaction(&mut tx, workspace, series_id, stopped_at_utc)
1142 .await?;
1143 if let Some((interval_id, paused_at)) = open_pause {
1144 let close_change_id = append_change(
1145 &mut tx,
1146 ChangeEntity::RecurrenceSeries,
1147 series_id.as_str(),
1148 Some("pause"),
1149 op_type::CLOSE_RECURRENCE_PAUSE,
1150 ChangePayload::workspace(workspace)
1151 .set("interval_id", &interval_id)
1152 .set("paused_at", &paused_at)
1153 .set("resumed_at", &stopped_at),
1154 )
1155 .await?;
1156 sqlx::query(
1157 "UPDATE recurrence_pause_intervals SET resumed_at = ?, resolved_by_change_id = ?
1158 WHERE workspace_id = ? AND id = ? AND resumed_at = ''",
1159 )
1160 .bind(&stopped_at)
1161 .bind(&close_change_id)
1162 .bind(&workspace.id)
1163 .bind(&interval_id)
1164 .execute(&mut *tx)
1165 .await?;
1166 }
1167 set_series_state(
1168 &mut tx,
1169 workspace,
1170 series_id,
1171 RecurrenceSeriesState::Stopped,
1172 Some(&stopped_at),
1173 &stopped_at,
1174 op_type::STOP_RECURRENCE_SERIES,
1175 )
1176 .await?;
1177 let projected = load_projected_occurrence(&mut tx, &workspace.id, series_id).await?;
1178 let occurrence = if skip_current {
1179 let projected = projected
1180 .ok_or_else(|| CoreError::validation("error recurrence-current-occurrence-missing"))?;
1181 let task_id = projected
1182 .task_id
1183 .as_ref()
1184 .expect("projected occurrence has a task");
1185 Some(
1186 resolve_recurrence_occurrence_in_transaction(
1187 &mut tx,
1188 workspace,
1189 task_id,
1190 RecurrenceOutcome::Skipped,
1191 &stopped_at,
1192 )
1193 .await?
1194 .resolved,
1195 )
1196 } else {
1197 projected
1198 };
1199 let series = load_series(&mut tx, &workspace.id, series_id).await?;
1200 tx.commit().await?;
1201 Ok(RecurrenceStateOutcome { series, occurrence })
1202}
1203
1204pub(crate) async fn route_recurrence_task_field(
1205 conn: &mut SqliteConnection,
1206 workspace: &Workspace,
1207 task_id: &TaskId,
1208 field: &str,
1209 value: &str,
1210) -> Result<Option<bool>> {
1211 let Some(mut occurrence) = load_occurrence_for_task(conn, &workspace.id, task_id).await? else {
1212 return Ok(None);
1213 };
1214 if matches!(
1215 occurrence.projection_state,
1216 RecurrenceProjectionState::Projected
1217 ) {
1218 reconcile_recurrence_series_in_transaction(
1219 conn,
1220 workspace,
1221 &occurrence.series_id,
1222 Utc::now(),
1223 )
1224 .await?;
1225 occurrence = load_occurrence_for_task(conn, &workspace.id, task_id)
1226 .await?
1227 .ok_or_else(|| CoreError::not_found("error recurrence-occurrence-not-found"))?;
1228 }
1229 if matches!(
1230 occurrence.projection_state,
1231 RecurrenceProjectionState::Archived
1232 ) {
1233 return Err(CoreError::validation(format!(
1234 "error recurrence-occurrence-archived task_id={task_id}"
1235 ))
1236 .into());
1237 }
1238 if field == "status" {
1239 let task = get_task_in_workspace(conn, workspace, task_id).await?;
1240 let target = TaskStatus::parse(value)?;
1241 if task.status == target {
1242 return Ok(Some(false));
1243 }
1244 if task.status.is_terminal() {
1245 if target.is_open() {
1246 return Err(CoreError::validation(format!(
1247 "error recurrence-terminal-reopen task_id={task_id} hint=\"use immediate undo\""
1248 ))
1249 .into());
1250 }
1251 return Err(CoreError::validation(format!(
1252 "error recurrence-outcome-final task_id={task_id} hint=\"use immediate undo\""
1253 ))
1254 .into());
1255 }
1256 if target.is_terminal() {
1257 ensure!(
1258 matches!(
1259 occurrence.projection_state,
1260 RecurrenceProjectionState::Projected
1261 ),
1262 "error recurrence-occurrence-not-current task_id={task_id}"
1263 );
1264 let outcome = if target == TaskStatus::Done {
1265 RecurrenceOutcome::Completed
1266 } else {
1267 RecurrenceOutcome::Skipped
1268 };
1269 resolve_recurrence_occurrence_in_transaction(conn, workspace, task_id, outcome, &now())
1270 .await?;
1271 return Ok(Some(true));
1272 }
1273 return Ok(None);
1274 }
1275 if field == "deleted"
1276 && matches!(
1277 occurrence.projection_state,
1278 RecurrenceProjectionState::Projected
1279 )
1280 && value == "1"
1281 {
1282 let series = load_series(conn, &workspace.id, &occurrence.series_id).await?;
1283 let guidance = match series.state {
1284 RecurrenceSeriesState::Active => "skip, pause, or stop the series",
1285 RecurrenceSeriesState::Paused => "skip the occurrence or stop the series",
1286 RecurrenceSeriesState::Stopped => "complete or cancel the final occurrence",
1287 };
1288 return Err(CoreError::validation(format!(
1289 "error recurrence-current-delete task_id={task_id} hint=\"{guidance}\""
1290 ))
1291 .into());
1292 }
1293 Ok(None)
1294}
1295
1296pub(crate) async fn undo_recurrence_resolution(
1297 conn: &mut SqliteConnection,
1298 workspace_id: &WorkspaceId,
1299 task_id: &TaskId,
1300 before_status: &str,
1301 after_status: &str,
1302) -> Result<bool> {
1303 let Some(occurrence) = load_occurrence_for_task(conn, workspace_id, task_id).await? else {
1304 return Ok(false);
1305 };
1306 if !matches!(
1307 occurrence.projection_state,
1308 RecurrenceProjectionState::Resolved
1309 ) || occurrence.outcome.is_none()
1310 {
1311 return Ok(false);
1312 }
1313 let task_status: String =
1314 sqlx::query_scalar("SELECT status FROM tasks WHERE workspace_id = ? AND id = ?")
1315 .bind(workspace_id)
1316 .bind(task_id)
1317 .fetch_one(&mut *conn)
1318 .await?;
1319 ensure!(
1320 task_status == after_status,
1321 "error undo-state-changed task_id={task_id} field=status"
1322 );
1323 let series = load_series(conn, workspace_id, &occurrence.series_id).await?;
1324 let outcome_change_id = occurrence
1325 .outcome_change_id
1326 .as_deref()
1327 .context("error recurrence-undo-missing-outcome-change")?;
1328 ensure_change_is_latest_series_transition(conn, workspace_id, &series.id, outcome_change_id)
1329 .await?;
1330 ensure_change_unsynced(conn, outcome_change_id).await?;
1331 let status_change_id = entity_field_version(
1332 conn,
1333 workspace_id,
1334 MutableEntityType::Task,
1335 task_id.as_str(),
1336 "status",
1337 )
1338 .await?
1339 .context("error recurrence-undo-missing-status-change")?;
1340 ensure_change_unsynced(conn, &status_change_id).await?;
1341 let prior_status_version: Option<String> =
1342 sqlx::query_scalar("SELECT base_version FROM changes WHERE change_id = ?")
1343 .bind(&status_change_id)
1344 .fetch_one(&mut *conn)
1345 .await?;
1346
1347 let successor = load_projected_occurrence(conn, workspace_id, &series.id).await?;
1348 match series.state {
1349 RecurrenceSeriesState::Active => {
1350 let successor = successor.context("error recurrence-undo-successor-missing")?;
1351 let expected_slot = next_slot_after(&series.rule, series.start_on, occurrence.slot_on)
1352 .context("recurrence schedule has no representable successor")?;
1353 ensure!(
1354 successor.slot_on == expected_slot,
1355 "error recurrence-undo-successor-changed"
1356 );
1357 ensure_successor_untouched(conn, workspace_id, &series, &successor).await?;
1358 remove_materialized_occurrence(conn, workspace_id, &series, &successor).await?;
1359 }
1360 RecurrenceSeriesState::Paused | RecurrenceSeriesState::Stopped => {
1361 ensure!(
1362 successor.is_none(),
1363 "error recurrence-undo-successor-exists"
1364 );
1365 }
1366 }
1367
1368 apply_field_value_in_workspace(conn, workspace_id, task_id, "status", before_status).await?;
1369 if let Some(version) = prior_status_version {
1370 set_entity_field_version(
1371 conn,
1372 workspace_id,
1373 MutableEntityType::Task,
1374 task_id.as_str(),
1375 "status",
1376 &version,
1377 )
1378 .await?;
1379 } else {
1380 sqlx::query(
1381 "DELETE FROM field_versions
1382 WHERE workspace_id = ? AND entity_type = 'task' AND entity_id = ? AND field = 'status'",
1383 )
1384 .bind(workspace_id)
1385 .bind(task_id)
1386 .execute(&mut *conn)
1387 .await?;
1388 }
1389 sqlx::query(
1390 "UPDATE recurrence_occurrences
1391 SET outcome = '', resolved_at = '', outcome_change_id = '', projection_state = 'projected'
1392 WHERE workspace_id = ? AND series_id = ? AND slot_on = ?",
1393 )
1394 .bind(workspace_id)
1395 .bind(&series.id)
1396 .bind(occurrence.slot_on.format("%Y-%m-%d").to_string())
1397 .execute(&mut *conn)
1398 .await?;
1399 sqlx::query("DELETE FROM changes WHERE change_id IN (?, ?)")
1400 .bind(&status_change_id)
1401 .bind(outcome_change_id)
1402 .execute(&mut *conn)
1403 .await?;
1404 Ok(true)
1405}
1406
1407async fn materialize_occurrence(
1408 conn: &mut SqliteConnection,
1409 workspace: &Workspace,
1410 series: &RecurrenceSeries,
1411 labels: &[String],
1412 slot_on: NaiveDate,
1413) -> Result<RecurrenceOccurrence> {
1414 let schedule = series.schedule();
1415 let slot = slot_values(&schedule, slot_on)?;
1416 let identity = derive_occurrence_identity(&workspace.id, &series.id, &schedule, slot_on)?;
1417 if let Some(existing) = load_occurrence(conn, &workspace.id, &series.id, slot_on).await? {
1418 verify_materialized_occurrence(
1419 conn, workspace, series, labels, &slot, &identity, &existing,
1420 )
1421 .await?;
1422 return Ok(existing);
1423 }
1424
1425 sqlx::query(
1426 "INSERT INTO tasks(
1427 workspace_id, id, title, description, project_id, status, priority,
1428 created_at, updated_at, queue_activity_at, available_at, due_on, deleted, is_epic
1429 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0)",
1430 )
1431 .bind(&workspace.id)
1432 .bind(&identity.task_id)
1433 .bind(&series.title)
1434 .bind(&series.description)
1435 .bind(&series.project_id)
1436 .bind(series.initial_status.as_str())
1437 .bind(series.priority.as_str())
1438 .bind(&identity.created_at)
1439 .bind(&identity.updated_at)
1440 .bind(&identity.created_at)
1441 .bind(&slot.available_at)
1442 .bind(slot.due_on.as_deref().unwrap_or(""))
1443 .execute(&mut *conn)
1444 .await?;
1445 for label in labels {
1446 sqlx::query("INSERT INTO task_labels(workspace_id, task_id, label) VALUES (?, ?, ?)")
1447 .bind(&workspace.id)
1448 .bind(&identity.task_id)
1449 .bind(label)
1450 .execute(&mut *conn)
1451 .await?;
1452 }
1453 sqlx::query(
1454 "INSERT INTO recurrence_occurrences(
1455 workspace_id, series_id, slot_on, task_id, outcome, resolved_at,
1456 outcome_change_id, projection_state, archived_at
1457 ) VALUES (?, ?, ?, ?, '', '', '', 'projected', '')",
1458 )
1459 .bind(&workspace.id)
1460 .bind(&series.id)
1461 .bind(slot_on.format("%Y-%m-%d").to_string())
1462 .bind(&identity.task_id)
1463 .execute(&mut *conn)
1464 .await?;
1465
1466 let payload = deterministic_task_payload(workspace, series, labels, &slot, &identity);
1467 insert_change_with_identity(
1468 conn,
1469 IdentifiedChange {
1470 change_id: &identity.task_change_id,
1471 entity_type: "task",
1472 entity_id: identity.task_id.as_str(),
1473 field: None,
1474 op_type: op_type::CREATE_TASK,
1475 payload,
1476 base_version: None,
1477 created_at: &identity.created_at,
1478 },
1479 )
1480 .await?;
1481 let occurrence_payload = ChangePayload::workspace(workspace)
1482 .set("series_id", series.id.as_str())
1483 .set("slot_on", slot_on.format("%Y-%m-%d").to_string())
1484 .set("task_id", identity.task_id.as_str())
1485 .set("projected_at", &identity.occurrence_link.projected_at)
1486 .set("task_change_id", &identity.task_change_id)
1487 .set("occurrence_change_id", &identity.occurrence_change_id)
1488 .set(
1489 "task_field_version_seed",
1490 &identity.field_version_seeds.task,
1491 )
1492 .set(
1493 "occurrence_field_version_seed",
1494 &identity.field_version_seeds.occurrence,
1495 )
1496 .set("frequency", series.rule.frequency().as_str())
1497 .set("interval", series.rule.interval())
1498 .set("weekdays", series.rule.weekdays_set().to_string())
1499 .set("timezone", series.timezone.as_str())
1500 .set("start_on", series.start_on.format("%Y-%m-%d").to_string())
1501 .set(
1502 "available_local_time",
1503 format_local_time(series.available_local_time),
1504 )
1505 .set("due_policy", series.due_policy.as_str())
1506 .into_value();
1507 insert_change_with_identity(
1508 conn,
1509 IdentifiedChange {
1510 change_id: &identity.occurrence_change_id,
1511 entity_type: "recurrence_series",
1512 entity_id: series.id.as_str(),
1513 field: Some("projection"),
1514 op_type: op_type::PROJECT_RECURRENCE_OCCURRENCE,
1515 payload: occurrence_payload,
1516 base_version: None,
1517 created_at: &identity.occurrence_link.projected_at,
1518 },
1519 )
1520 .await?;
1521 for field in TaskField::VERSIONED {
1522 set_entity_field_version(
1523 conn,
1524 &workspace.id,
1525 MutableEntityType::Task,
1526 identity.task_id.as_str(),
1527 field.as_str(),
1528 &identity.field_version_seeds.task,
1529 )
1530 .await?;
1531 }
1532 load_occurrence(conn, &workspace.id, &series.id, slot_on)
1533 .await?
1534 .context("materialized recurrence occurrence missing")
1535}
1536
1537async fn verify_materialized_occurrence(
1538 conn: &mut SqliteConnection,
1539 workspace: &Workspace,
1540 series: &RecurrenceSeries,
1541 labels: &[String],
1542 slot: &crate::recurrence::RecurrenceSlot,
1543 identity: &crate::recurrence::RecurrenceOccurrenceIdentity,
1544 occurrence: &RecurrenceOccurrence,
1545) -> Result<()> {
1546 ensure!(
1547 occurrence.task_id.as_ref() == Some(&identity.task_id),
1548 CoreError::generation_conflict(format!(
1549 "error recurrence-generation-conflict slot={} field=task_id",
1550 occurrence.slot_on
1551 ))
1552 );
1553 let task = get_task_in_workspace(conn, workspace, &identity.task_id).await?;
1554 ensure!(
1555 task.title == series.title
1556 && task.description == series.description
1557 && task.project_id == series.project_id
1558 && task.status == series.initial_status
1559 && task.priority == series.priority
1560 && task.created_at == identity.created_at
1561 && task.updated_at == identity.updated_at
1562 && task.queue_activity_at == identity.created_at
1563 && task.available_at.as_deref() == Some(slot.available_at.as_str())
1564 && task.due_on == slot.due_on
1565 && !task.deleted
1566 && !task.is_epic,
1567 CoreError::generation_conflict(format!(
1568 "error recurrence-generation-conflict slot={} field=task",
1569 occurrence.slot_on
1570 ))
1571 );
1572 let stored_labels: Vec<String> = sqlx::query_scalar(
1573 "SELECT label FROM task_labels WHERE workspace_id = ? AND task_id = ? ORDER BY label",
1574 )
1575 .bind(&workspace.id)
1576 .bind(&identity.task_id)
1577 .fetch_all(&mut *conn)
1578 .await?;
1579 ensure!(
1580 stored_labels == labels,
1581 CoreError::generation_conflict(format!(
1582 "error recurrence-generation-conflict slot={} field=labels",
1583 occurrence.slot_on
1584 ))
1585 );
1586 for field in TaskField::VERSIONED {
1587 let version = entity_field_version(
1588 conn,
1589 &workspace.id,
1590 MutableEntityType::Task,
1591 identity.task_id.as_str(),
1592 field.as_str(),
1593 )
1594 .await?;
1595 ensure!(
1596 version.as_deref() == Some(identity.field_version_seeds.task.as_str()),
1597 CoreError::generation_conflict(format!(
1598 "error recurrence-generation-conflict slot={} field=field_versions",
1599 occurrence.slot_on
1600 ))
1601 );
1602 }
1603 let payload = deterministic_task_payload(workspace, series, labels, slot, identity);
1604 insert_change_with_identity(
1605 conn,
1606 IdentifiedChange {
1607 change_id: &identity.task_change_id,
1608 entity_type: "task",
1609 entity_id: identity.task_id.as_str(),
1610 field: None,
1611 op_type: op_type::CREATE_TASK,
1612 payload,
1613 base_version: None,
1614 created_at: &identity.created_at,
1615 },
1616 )
1617 .await?;
1618 Ok(())
1619}
1620
1621fn deterministic_task_payload(
1622 workspace: &Workspace,
1623 series: &RecurrenceSeries,
1624 labels: &[String],
1625 slot: &crate::recurrence::RecurrenceSlot,
1626 identity: &crate::recurrence::RecurrenceOccurrenceIdentity,
1627) -> Value {
1628 ChangePayload::workspace(workspace)
1629 .set("task_id", identity.task_id.as_str())
1630 .set("series_id", series.id.as_str())
1631 .set("slot_on", slot.scheduled_on.format("%Y-%m-%d").to_string())
1632 .set("title", &series.title)
1633 .set("description", &series.description)
1634 .set("project_id", series.project_id.as_str())
1635 .set("status", series.initial_status.as_str())
1636 .set("priority", series.priority.as_str())
1637 .set("available_at", &slot.available_at)
1638 .set("due_on", slot.due_on.as_deref().unwrap_or(""))
1639 .set("is_epic", "0")
1640 .set("labels", labels)
1641 .set("created_at", &identity.created_at)
1642 .set("updated_at", &identity.updated_at)
1643 .set("task_change_id", &identity.task_change_id)
1644 .set("occurrence_change_id", &identity.occurrence_change_id)
1645 .set(
1646 "task_field_version_seed",
1647 &identity.field_version_seeds.task,
1648 )
1649 .set(
1650 "occurrence_field_version_seed",
1651 &identity.field_version_seeds.occurrence,
1652 )
1653 .set("frequency", series.rule.frequency().as_str())
1654 .set("interval", series.rule.interval())
1655 .set("weekdays", series.rule.weekdays_set().to_string())
1656 .set("timezone", series.timezone.as_str())
1657 .set("start_on", series.start_on.format("%Y-%m-%d").to_string())
1658 .set(
1659 "available_local_time",
1660 format_local_time(series.available_local_time),
1661 )
1662 .set("due_policy", series.due_policy.as_str())
1663 .into_value()
1664}
1665
1666async fn write_task_status(
1667 conn: &mut SqliteConnection,
1668 workspace: &Workspace,
1669 task_id: &TaskId,
1670 before: TaskStatus,
1671 after: TaskStatus,
1672 resolved_at: &str,
1673) -> Result<String> {
1674 let field = "status";
1675 ensure!(
1676 !crate::db::conflict_exists(conn, &workspace.id, task_id, field).await?,
1677 "error conflicted-field ref={} field=status hint=\"use conflict resolve\"",
1678 task_id
1679 );
1680 let base = entity_field_version(
1681 conn,
1682 &workspace.id,
1683 MutableEntityType::Task,
1684 task_id.as_str(),
1685 field,
1686 )
1687 .await?;
1688 apply_field_value_in_workspace(conn, &workspace.id, task_id, field, after.as_str()).await?;
1689 let change_id = insert_change(
1690 conn,
1691 "task",
1692 task_id.as_str(),
1693 Some(field),
1694 op_type::SET_FIELD,
1695 TaskField::Status.scalar_payload(&workspace.id, &workspace.key, after.as_str())?,
1696 base.as_deref(),
1697 )
1698 .await?;
1699 set_field_version(conn, task_id, field, &change_id).await?;
1700 ensure!(
1701 before.is_open() && after.is_terminal(),
1702 "error recurrence-invalid-terminal-transition"
1703 );
1704 sqlx::query(
1705 "UPDATE tasks SET updated_at = ?, queue_activity_at = ?
1706 WHERE workspace_id = ? AND id = ?",
1707 )
1708 .bind(resolved_at)
1709 .bind(resolved_at)
1710 .bind(&workspace.id)
1711 .bind(task_id)
1712 .execute(&mut *conn)
1713 .await?;
1714 Ok(change_id)
1715}
1716
1717async fn set_series_state(
1718 conn: &mut SqliteConnection,
1719 workspace: &Workspace,
1720 series_id: &RecurrenceSeriesId,
1721 state: RecurrenceSeriesState,
1722 stopped_at: Option<&str>,
1723 changed_at: &str,
1724 operation: &'static str,
1725) -> Result<String> {
1726 let base = entity_field_version(
1727 conn,
1728 &workspace.id,
1729 MutableEntityType::RecurrenceSeries,
1730 series_id.as_str(),
1731 "state",
1732 )
1733 .await?;
1734 sqlx::query(
1735 "UPDATE recurrence_series SET state = ?, stopped_at = ?, updated_at = ?
1736 WHERE workspace_id = ? AND id = ?",
1737 )
1738 .bind(state.as_str())
1739 .bind(stopped_at.unwrap_or(""))
1740 .bind(changed_at)
1741 .bind(&workspace.id)
1742 .bind(series_id)
1743 .execute(&mut *conn)
1744 .await?;
1745 let change_id = insert_change(
1746 conn,
1747 "recurrence_series",
1748 series_id.as_str(),
1749 Some("state"),
1750 operation,
1751 ChangePayload::workspace(workspace)
1752 .set("state", state.as_str())
1753 .set("stopped_at", stopped_at.unwrap_or(""))
1754 .set("changed_at", changed_at)
1755 .into_value(),
1756 base.as_deref(),
1757 )
1758 .await?;
1759 set_entity_field_version(
1760 conn,
1761 &workspace.id,
1762 MutableEntityType::RecurrenceSeries,
1763 series_id.as_str(),
1764 "state",
1765 &change_id,
1766 )
1767 .await?;
1768 if matches!(state, RecurrenceSeriesState::Stopped) {
1769 set_entity_field_version(
1770 conn,
1771 &workspace.id,
1772 MutableEntityType::RecurrenceSeries,
1773 series_id.as_str(),
1774 "stopped_at",
1775 &change_id,
1776 )
1777 .await?;
1778 }
1779 Ok(change_id)
1780}
1781
1782async fn update_series_template_scalar(
1783 conn: &mut SqliteConnection,
1784 workspace_id: &WorkspaceId,
1785 series_id: &RecurrenceSeriesId,
1786 field: &str,
1787 value: &str,
1788 updated_at: &str,
1789) -> Result<()> {
1790 let query = match field {
1791 "title" => {
1792 "UPDATE recurrence_series SET title = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
1793 }
1794 "description" => {
1795 "UPDATE recurrence_series SET description = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
1796 }
1797 "project" => {
1798 "UPDATE recurrence_series SET project_id = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
1799 }
1800 "priority" => {
1801 "UPDATE recurrence_series SET priority = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
1802 }
1803 "initial_status" => {
1804 "UPDATE recurrence_series SET initial_status = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
1805 }
1806 "available_local_time" => {
1807 "UPDATE recurrence_series SET available_local_time = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
1808 }
1809 "due_policy" => {
1810 "UPDATE recurrence_series SET due_policy = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
1811 }
1812 _ => bail!("invalid recurrence template field: {field}"),
1813 };
1814 sqlx::query(query)
1815 .bind(value)
1816 .bind(updated_at)
1817 .bind(workspace_id)
1818 .bind(series_id)
1819 .execute(&mut *conn)
1820 .await?;
1821 Ok(())
1822}
1823
1824async fn load_series(
1825 conn: &mut SqliteConnection,
1826 workspace_id: &WorkspaceId,
1827 series_id: &RecurrenceSeriesId,
1828) -> Result<RecurrenceSeries> {
1829 let row = sqlx::query(
1830 "SELECT workspace_id, id, title, description, project_id, priority, initial_status,
1831 frequency, interval, weekdays, timezone, start_on, available_local_time,
1832 due_policy, state, stopped_at, created_at, updated_at, deleted
1833 FROM recurrence_series WHERE workspace_id = ? AND id = ?",
1834 )
1835 .bind(workspace_id)
1836 .bind(series_id)
1837 .fetch_optional(&mut *conn)
1838 .await?
1839 .ok_or_else(|| {
1840 CoreError::not_found(format!(
1841 "error recurrence-series-not-found series_id={series_id}"
1842 ))
1843 })?;
1844 recurrence_series_from_row(&row)
1845}
1846
1847async fn load_series_labels(
1848 conn: &mut SqliteConnection,
1849 workspace_id: &WorkspaceId,
1850 series_id: &RecurrenceSeriesId,
1851) -> Result<Vec<String>> {
1852 Ok(sqlx::query_scalar(
1853 "SELECT label FROM recurrence_series_labels
1854 WHERE workspace_id = ? AND series_id = ? ORDER BY label",
1855 )
1856 .bind(workspace_id)
1857 .bind(series_id)
1858 .fetch_all(&mut *conn)
1859 .await?)
1860}
1861
1862async fn load_occurrence(
1863 conn: &mut SqliteConnection,
1864 workspace_id: &WorkspaceId,
1865 series_id: &RecurrenceSeriesId,
1866 slot_on: NaiveDate,
1867) -> Result<Option<RecurrenceOccurrence>> {
1868 let row = sqlx::query(
1869 "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at,
1870 outcome_change_id, projection_state, archived_at
1871 FROM recurrence_occurrences
1872 WHERE workspace_id = ? AND series_id = ? AND slot_on = ?",
1873 )
1874 .bind(workspace_id)
1875 .bind(series_id)
1876 .bind(slot_on.format("%Y-%m-%d").to_string())
1877 .fetch_optional(&mut *conn)
1878 .await?;
1879 row.as_ref().map(recurrence_occurrence_from_row).transpose()
1880}
1881
1882async fn load_projected_occurrence(
1883 conn: &mut SqliteConnection,
1884 workspace_id: &WorkspaceId,
1885 series_id: &RecurrenceSeriesId,
1886) -> Result<Option<RecurrenceOccurrence>> {
1887 let row = sqlx::query(
1888 "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at,
1889 outcome_change_id, projection_state, archived_at
1890 FROM recurrence_occurrences
1891 WHERE workspace_id = ? AND series_id = ? AND projection_state = 'projected'",
1892 )
1893 .bind(workspace_id)
1894 .bind(series_id)
1895 .fetch_optional(&mut *conn)
1896 .await?;
1897 row.as_ref().map(recurrence_occurrence_from_row).transpose()
1898}
1899
1900async fn load_occurrence_for_task(
1901 conn: &mut SqliteConnection,
1902 workspace_id: &WorkspaceId,
1903 task_id: &TaskId,
1904) -> Result<Option<RecurrenceOccurrence>> {
1905 let row = sqlx::query(
1906 "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at,
1907 outcome_change_id, projection_state, archived_at
1908 FROM recurrence_occurrences WHERE workspace_id = ? AND task_id = ?",
1909 )
1910 .bind(workspace_id)
1911 .bind(task_id)
1912 .fetch_optional(&mut *conn)
1913 .await?;
1914 row.as_ref().map(recurrence_occurrence_from_row).transpose()
1915}
1916
1917async fn lifecycle_conflict_exists(
1918 conn: &mut SqliteConnection,
1919 workspace_id: &WorkspaceId,
1920 series_id: &RecurrenceSeriesId,
1921) -> Result<bool> {
1922 for field in ["state", "stopped_at"] {
1923 if entity_conflict_exists(
1924 conn,
1925 workspace_id,
1926 MutableEntityType::RecurrenceSeries,
1927 series_id.as_str(),
1928 field,
1929 )
1930 .await?
1931 {
1932 return Ok(true);
1933 }
1934 }
1935 Ok(false)
1936}
1937
1938async fn ensure_no_series_conflict(
1939 conn: &mut SqliteConnection,
1940 workspace_id: &WorkspaceId,
1941 series_id: &RecurrenceSeriesId,
1942 field: &str,
1943) -> Result<()> {
1944 ensure!(
1945 !entity_conflict_exists(
1946 conn,
1947 workspace_id,
1948 MutableEntityType::RecurrenceSeries,
1949 series_id.as_str(),
1950 field,
1951 )
1952 .await?,
1953 CoreError::open_conflict(format!(
1954 "error recurrence-conflicted-field series_id={series_id} field={field}"
1955 ))
1956 );
1957 Ok(())
1958}
1959
1960pub async fn resolve_recurrence_ref(
1961 conn: &mut SqliteConnection,
1962 workspace: &Workspace,
1963 input: &str,
1964) -> Result<RecurrenceSeries> {
1965 let (hint, suffix) = split_ref(input);
1966 if suffix.len() < 3 {
1967 return Err(CoreError::validation(format!(
1968 "error recurrence-ref-too-short input={input} minimum=3"
1969 ))
1970 .into());
1971 }
1972 let series_rows = sqlx::query(
1973 "SELECT workspace_id, id, title, description, project_id, priority, initial_status,
1974 frequency, interval, weekdays, timezone, start_on, available_local_time,
1975 due_policy, state, stopped_at, created_at, updated_at, deleted
1976 FROM recurrence_series WHERE workspace_id = ? AND id LIKE ? || '%' ORDER BY id",
1977 )
1978 .bind(&workspace.id)
1979 .bind(&suffix)
1980 .fetch_all(&mut *conn)
1981 .await?;
1982 let mut series = series_rows
1983 .iter()
1984 .map(recurrence_series_from_row)
1985 .collect::<Result<Vec<_>>>()?;
1986 if hint
1987 .as_deref()
1988 .is_some_and(|value| value != SERIES_REF_PREFIX)
1989 {
1990 series.clear();
1991 }
1992 if series.len() == 1 {
1993 return Ok(series.remove(0));
1994 }
1995 if series.len() > 1 {
1996 return Err(
1997 CoreError::validation(format!("error ambiguous-recurrence-ref input={input}")).into(),
1998 );
1999 }
2000 let task = crate::refs::resolve_task_ref_in_workspace(conn, workspace, input).await?;
2001 let occurrence = load_occurrence_for_task(conn, &workspace.id, &task.id)
2002 .await?
2003 .context("error task-is-not-recurring")?;
2004 load_series(conn, &workspace.id, &occurrence.series_id).await
2005}
2006
2007async fn recurrence_series_ref(
2008 conn: &mut SqliteConnection,
2009 workspace_id: &WorkspaceId,
2010 series_id: &RecurrenceSeriesId,
2011) -> Result<String> {
2012 let ids = sqlx::query_scalar::<_, RecurrenceSeriesId>(
2013 "SELECT id FROM recurrence_series WHERE workspace_id = ? ORDER BY id",
2014 )
2015 .bind(workspace_id)
2016 .fetch_all(&mut *conn)
2017 .await?;
2018 ensure!(
2019 ids.iter().any(|candidate| candidate == series_id),
2020 CoreError::not_found(format!(
2021 "error recurrence-series-not-found series_id={series_id}"
2022 ))
2023 );
2024 Ok(recurrence_series_display_ref(series_id, &ids))
2025}
2026
2027fn split_ref(input: &str) -> (Option<String>, String) {
2028 let (hint, suffix) = input
2029 .split_once('-')
2030 .map_or((None, input), |(hint, suffix)| (Some(hint), suffix));
2031 let suffix = suffix
2032 .chars()
2033 .filter(|value| value.is_ascii_alphanumeric())
2034 .map(|value| match value.to_ascii_uppercase() {
2035 'O' => '0',
2036 'I' | 'L' => '1',
2037 value => value,
2038 })
2039 .collect();
2040 (hint.map(|value| value.to_ascii_uppercase()), suffix)
2041}
2042
2043async fn ensure_change_is_latest_series_transition(
2044 conn: &mut SqliteConnection,
2045 workspace_id: &WorkspaceId,
2046 series_id: &RecurrenceSeriesId,
2047 change_id: &str,
2048) -> Result<()> {
2049 let latest: Option<String> = sqlx::query_scalar(
2050 "SELECT change_id FROM changes
2051 WHERE entity_type = 'recurrence_series' AND entity_id = ?
2052 AND json_extract(payload, '$.workspace_id') = ?
2053 AND op_type != 'project_recurrence_occurrence'
2054 ORDER BY local_seq DESC LIMIT 1",
2055 )
2056 .bind(series_id)
2057 .bind(workspace_id.as_str())
2058 .fetch_optional(&mut *conn)
2059 .await?;
2060 ensure!(
2061 latest.as_deref() == Some(change_id),
2062 "error recurrence-undo-later-operation"
2063 );
2064 Ok(())
2065}
2066
2067async fn ensure_change_unsynced(conn: &mut SqliteConnection, change_id: &str) -> Result<()> {
2068 let server_seq: Option<i64> =
2069 sqlx::query_scalar("SELECT server_seq FROM changes WHERE change_id = ?")
2070 .bind(change_id)
2071 .fetch_one(&mut *conn)
2072 .await?;
2073 ensure!(server_seq.is_none(), "error recurrence-undo-already-synced");
2074 Ok(())
2075}
2076
2077async fn ensure_successor_untouched(
2078 conn: &mut SqliteConnection,
2079 workspace_id: &WorkspaceId,
2080 series: &RecurrenceSeries,
2081 occurrence: &RecurrenceOccurrence,
2082) -> Result<()> {
2083 let task_id = occurrence
2084 .task_id
2085 .as_ref()
2086 .context("error recurrence-undo-successor-missing-task")?;
2087 let identity = derive_occurrence_identity(
2088 workspace_id,
2089 &series.id,
2090 &series.schedule(),
2091 occurrence.slot_on,
2092 )?;
2093 ensure!(
2094 task_id == &identity.task_id,
2095 "error recurrence-undo-successor-changed"
2096 );
2097 let workspace = crate::workspaces::workspace_for_id(conn, workspace_id).await?;
2098 let labels = load_series_labels(conn, workspace_id, &series.id).await?;
2099 let slot = slot_values(&series.schedule(), occurrence.slot_on)?;
2100 verify_materialized_occurrence(
2101 conn, &workspace, series, &labels, &slot, &identity, occurrence,
2102 )
2103 .await
2104 .map_err(|_| anyhow::anyhow!("error recurrence-undo-successor-touched"))?;
2105 ensure_change_unsynced(conn, &identity.task_change_id).await?;
2106 ensure_change_unsynced(conn, &identity.occurrence_change_id).await?;
2107 let extra_changes: i64 = sqlx::query_scalar(
2108 "SELECT count(*) FROM changes
2109 WHERE entity_type = 'task' AND entity_id = ? AND change_id != ?",
2110 )
2111 .bind(task_id)
2112 .bind(&identity.task_change_id)
2113 .fetch_one(&mut *conn)
2114 .await?;
2115 let notes: i64 =
2116 sqlx::query_scalar("SELECT count(*) FROM notes WHERE workspace_id = ? AND task_id = ?")
2117 .bind(workspace_id)
2118 .bind(task_id)
2119 .fetch_one(&mut *conn)
2120 .await?;
2121 let attachments: i64 = sqlx::query_scalar(
2122 "SELECT count(*) FROM task_attachments WHERE workspace_id = ? AND task_id = ?",
2123 )
2124 .bind(workspace_id)
2125 .bind(task_id)
2126 .fetch_one(&mut *conn)
2127 .await?;
2128 let dependencies: i64 = sqlx::query_scalar(
2129 "SELECT count(*) FROM task_dependencies
2130 WHERE workspace_id = ? AND (task_id = ? OR depends_on_task_id = ?)",
2131 )
2132 .bind(workspace_id)
2133 .bind(task_id)
2134 .bind(task_id)
2135 .fetch_one(&mut *conn)
2136 .await?;
2137 let epics: i64 = sqlx::query_scalar(
2138 "SELECT count(*) FROM task_epic_links
2139 WHERE workspace_id = ? AND (child_task_id = ? OR epic_task_id = ?)",
2140 )
2141 .bind(workspace_id)
2142 .bind(task_id)
2143 .bind(task_id)
2144 .fetch_one(&mut *conn)
2145 .await?;
2146 ensure!(
2147 extra_changes == 0 && notes == 0 && attachments == 0 && dependencies == 0 && epics == 0,
2148 "error recurrence-undo-successor-touched"
2149 );
2150 Ok(())
2151}
2152
2153async fn remove_materialized_occurrence(
2154 conn: &mut SqliteConnection,
2155 workspace_id: &WorkspaceId,
2156 series: &RecurrenceSeries,
2157 occurrence: &RecurrenceOccurrence,
2158) -> Result<()> {
2159 let task_id = occurrence
2160 .task_id
2161 .as_ref()
2162 .expect("verified successor has task");
2163 let identity = derive_occurrence_identity(
2164 workspace_id,
2165 &series.id,
2166 &series.schedule(),
2167 occurrence.slot_on,
2168 )?;
2169 sqlx::query(
2170 "DELETE FROM recurrence_occurrences
2171 WHERE workspace_id = ? AND series_id = ? AND slot_on = ?",
2172 )
2173 .bind(workspace_id)
2174 .bind(&series.id)
2175 .bind(occurrence.slot_on.format("%Y-%m-%d").to_string())
2176 .execute(&mut *conn)
2177 .await?;
2178 sqlx::query("DELETE FROM task_labels WHERE workspace_id = ? AND task_id = ?")
2179 .bind(workspace_id)
2180 .bind(task_id)
2181 .execute(&mut *conn)
2182 .await?;
2183 sqlx::query(
2184 "DELETE FROM field_versions
2185 WHERE workspace_id = ? AND entity_type = 'task' AND entity_id = ?",
2186 )
2187 .bind(workspace_id)
2188 .bind(task_id)
2189 .execute(&mut *conn)
2190 .await?;
2191 sqlx::query("DELETE FROM tasks WHERE workspace_id = ? AND id = ?")
2192 .bind(workspace_id)
2193 .bind(task_id)
2194 .execute(&mut *conn)
2195 .await?;
2196 sqlx::query("DELETE FROM changes WHERE change_id IN (?, ?)")
2197 .bind(&identity.task_change_id)
2198 .bind(&identity.occurrence_change_id)
2199 .execute(&mut *conn)
2200 .await?;
2201 Ok(())
2202}
2203
2204fn retryable_reconcile_error(error: &anyhow::Error) -> bool {
2205 error
2206 .downcast_ref::<sqlx::Error>()
2207 .and_then(sqlx::Error::as_database_error)
2208 .and_then(|error| error.code())
2209 .is_some_and(|code| matches!(code.as_ref(), "5" | "6" | "19"))
2210}
2211
2212pub(crate) fn timestamp_strictly_after(earlier: &str, candidate: &str) -> Result<String> {
2213 if candidate > earlier {
2214 return Ok(candidate.to_string());
2215 }
2216 let earlier = DateTime::parse_from_rfc3339(earlier)?.with_timezone(&Utc);
2217 let adjusted = earlier
2218 .checked_add_signed(chrono::Duration::seconds(1))
2219 .context("recurrence lifecycle timestamp is out of range")?;
2220 Ok(format_utc(adjusted))
2221}
2222
2223fn format_local_time(value: Option<NaiveTime>) -> String {
2224 value
2225 .map(|value| value.format("%H:%M:%S").to_string())
2226 .unwrap_or_default()
2227}
2228
2229fn format_utc(value: DateTime<Utc>) -> String {
2230 value.format("%Y-%m-%dT%H:%M:%SZ").to_string()
2231}
2232
2233fn utc_now() -> Result<DateTime<Utc>> {
2234 Ok(DateTime::parse_from_rfc3339(&now())?.with_timezone(&Utc))
2235}