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