Skip to main content

max / goingson

6.1 KB · 188 lines History Blame Raw
1 //! The compose and send state machine: drafts, the outbox queue, the due set,
2 //! and send-failure bookkeeping.
3
4 use chrono::{DateTime, Utc};
5 use goingson_core::{CoreError, Email, EmailAccountId, EmailId, Result, UserId};
6 use rusqlite::{Connection, params};
7
8 use crate::utils::{execute, format_datetime, format_datetime_now, format_datetime_opt, query_all};
9
10 use super::query;
11 use super::row::{EMAIL_LIST_COLUMNS, EMAIL_SELECT_COLUMNS, EmailRow};
12
13 /// Every draft, newest first.
14 pub(super) fn list_drafts(conn: &Connection, user_id: UserId) -> Result<Vec<Email>> {
15 let sql = format!(
16 "SELECT {EMAIL_SELECT_COLUMNS} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.is_draft = 1 ORDER BY e.received_at DESC"
17 );
18 let rows = query_all(
19 conn,
20 &sql,
21 params![user_id.to_string(), user_id.to_string()],
22 EmailRow::from_row,
23 )?;
24 rows.into_iter().map(Email::try_from).collect()
25 }
26
27 /// Upsert a draft by id and return it as stored.
28 #[allow(clippy::too_many_arguments)]
29 pub(super) fn save_draft(
30 conn: &Connection,
31 id: EmailId,
32 user_id: UserId,
33 from: &str,
34 to: &str,
35 cc: Option<&str>,
36 bcc: Option<&str>,
37 subject: &str,
38 body: &str,
39 account_id: Option<EmailAccountId>,
40 in_reply_to: Option<&str>,
41 thread_id: Option<&str>,
42 ) -> Result<Email> {
43 let now = format_datetime_now();
44 let account_id_str = account_id.map(|a: EmailAccountId| a.to_string());
45
46 // Upsert: update if exists, insert if not
47 execute(
48 conn,
49 r"
50 INSERT INTO emails (id, user_id, from_address, to_address, cc_address, bcc_address, subject, body,
51 is_read, is_archived, is_draft, is_outgoing, received_at, draft_account_id, in_reply_to, thread_id)
52 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, 0, 1, 1, ?, ?, ?, ?)
53 ON CONFLICT(id) DO UPDATE SET
54 from_address = excluded.from_address,
55 to_address = excluded.to_address,
56 cc_address = excluded.cc_address,
57 bcc_address = excluded.bcc_address,
58 subject = excluded.subject,
59 body = excluded.body,
60 received_at = excluded.received_at,
61 draft_account_id = excluded.draft_account_id,
62 in_reply_to = excluded.in_reply_to,
63 thread_id = excluded.thread_id
64 ",
65 params![
66 id.to_string(),
67 user_id.to_string(),
68 from,
69 to,
70 cc,
71 bcc,
72 subject,
73 body,
74 &now,
75 &account_id_str,
76 in_reply_to,
77 thread_id
78 ],
79 )?;
80
81 query::get_by_id(conn, id, user_id)?
82 .ok_or_else(|| CoreError::internal("Failed to retrieve saved draft"))
83 }
84
85 /// Queue a draft for sending, optionally not before `send_after`.
86 pub(super) fn queue_draft(
87 conn: &Connection,
88 id: EmailId,
89 user_id: UserId,
90 send_after: Option<DateTime<Utc>>,
91 ) -> Result<Option<Email>> {
92 // `is_draft = 1` in the predicate rather than checked first: queueing a
93 // received message is refused by the write not matching, so there is no
94 // window between the check and the update.
95 let changed = execute(
96 conn,
97 "UPDATE emails SET queued_at = ?, send_after = ?, send_attempts = 0, send_error = NULL
98 WHERE id = ? AND user_id = ? AND is_draft = 1",
99 params![
100 format_datetime_now(),
101 format_datetime_opt(send_after),
102 id.to_string(),
103 user_id.to_string(),
104 ],
105 )?;
106 if changed == 0 {
107 return Ok(None);
108 }
109 query::get_by_id(conn, id, user_id)
110 }
111
112 /// Take a queued draft back out of the outbox.
113 pub(super) fn unqueue_draft(
114 conn: &Connection,
115 id: EmailId,
116 user_id: UserId,
117 ) -> Result<Option<Email>> {
118 let changed = execute(
119 conn,
120 "UPDATE emails SET queued_at = NULL, send_after = NULL, send_attempts = 0, send_error = NULL
121 WHERE id = ? AND user_id = ? AND queued_at IS NOT NULL",
122 params![id.to_string(), user_id.to_string()],
123 )?;
124 if changed == 0 {
125 return Ok(None);
126 }
127 query::get_by_id(conn, id, user_id)
128 }
129
130 /// Queued drafts, oldest queued first. Bodies are blanked.
131 pub(super) fn list_outbox(conn: &Connection, user_id: UserId) -> Result<Vec<Email>> {
132 let sql = format!(
133 "SELECT {EMAIL_LIST_COLUMNS} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ?
134 WHERE e.user_id = ? AND e.is_draft = 1 AND e.queued_at IS NOT NULL
135 ORDER BY e.queued_at ASC"
136 );
137 let rows = query_all(
138 conn,
139 &sql,
140 params![user_id.to_string(), user_id.to_string()],
141 EmailRow::from_row,
142 )?;
143 rows.into_iter().map(Email::try_from).collect()
144 }
145
146 /// Queued drafts whose send time has arrived, bodies included.
147 pub(super) fn list_due(
148 conn: &Connection,
149 user_id: UserId,
150 now: DateTime<Utc>,
151 ) -> Result<Vec<Email>> {
152 // The body is wanted here, unlike the outbox list: this is the set that
153 // is about to be sent, and a blanked body would send an empty message.
154 let sql = format!(
155 "SELECT {EMAIL_SELECT_COLUMNS} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ?
156 WHERE e.user_id = ? AND e.is_draft = 1 AND e.queued_at IS NOT NULL
157 AND (e.send_after IS NULL OR e.send_after <= ?)
158 ORDER BY e.queued_at ASC"
159 );
160 let rows = query_all(
161 conn,
162 &sql,
163 params![
164 user_id.to_string(),
165 user_id.to_string(),
166 format_datetime(&now)
167 ],
168 EmailRow::from_row,
169 )?;
170 rows.into_iter().map(Email::try_from).collect()
171 }
172
173 /// Count a failed send attempt and record its error.
174 pub(super) fn record_send_failure(
175 conn: &Connection,
176 id: EmailId,
177 user_id: UserId,
178 error: &str,
179 ) -> Result<()> {
180 execute(
181 conn,
182 "UPDATE emails SET send_attempts = send_attempts + 1, send_error = ?
183 WHERE id = ? AND user_id = ?",
184 params![error, id.to_string(), user_id.to_string()],
185 )?;
186 Ok(())
187 }
188