1use std::cmp::max;
66use std::collections::BTreeSet;
67use std::fmt;
68use std::num::ParseIntError;
69use std::str::FromStr;
70use std::time::{Duration, UNIX_EPOCH};
71
72use anyhow::{Context as _, Result, ensure};
73use async_channel::Receiver;
74use serde::{Deserialize, Serialize};
75use tokio::time::timeout;
76
77use crate::chat::{ChatId, ChatIdBlocked, send_msg};
78use crate::constants::{DC_CHAT_ID_LAST_SPECIAL, DC_CHAT_ID_TRASH};
79use crate::contact::ContactId;
80use crate::context::Context;
81use crate::download::MIN_DELETE_SERVER_AFTER;
82use crate::events::EventType;
83use crate::location;
84use crate::log::{LogExt, error, info, warn};
85use crate::message::{Message, MessageState, MsgId, Viewtype};
86use crate::mimeparser::SystemMessage;
87use crate::stock_str;
88use crate::tools::{SystemTime, duration_to_str, time};
89
90#[derive(Debug, PartialEq, Eq, Copy, Clone, Serialize, Deserialize)]
92pub enum Timer {
93 Disabled,
95
96 Enabled {
98 duration: u32,
102 },
103}
104
105impl Timer {
106 pub fn to_u32(self) -> u32 {
110 match self {
111 Self::Disabled => 0,
112 Self::Enabled { duration } => duration,
113 }
114 }
115
116 pub fn from_u32(duration: u32) -> Self {
120 if duration == 0 {
121 Self::Disabled
122 } else {
123 Self::Enabled { duration }
124 }
125 }
126}
127
128impl Default for Timer {
129 fn default() -> Self {
130 Self::Disabled
131 }
132}
133
134impl fmt::Display for Timer {
135 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136 write!(f, "{}", self.to_u32())
137 }
138}
139
140impl FromStr for Timer {
141 type Err = ParseIntError;
142
143 fn from_str(input: &str) -> Result<Timer, ParseIntError> {
144 input.parse::<u32>().map(Self::from_u32)
145 }
146}
147
148impl rusqlite::types::ToSql for Timer {
149 fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
150 let val = rusqlite::types::Value::Integer(match self {
151 Self::Disabled => 0,
152 Self::Enabled { duration } => i64::from(*duration),
153 });
154 let out = rusqlite::types::ToSqlOutput::Owned(val);
155 Ok(out)
156 }
157}
158
159impl rusqlite::types::FromSql for Timer {
160 fn column_result(value: rusqlite::types::ValueRef) -> rusqlite::types::FromSqlResult<Self> {
161 i64::column_result(value).and_then(|value| {
162 if value == 0 {
163 Ok(Self::Disabled)
164 } else if let Ok(duration) = u32::try_from(value) {
165 Ok(Self::Enabled { duration })
166 } else {
167 Err(rusqlite::types::FromSqlError::OutOfRange(value))
168 }
169 })
170 }
171}
172
173impl ChatId {
174 pub async fn get_ephemeral_timer(self, context: &Context) -> Result<Timer> {
176 let timer = context
177 .sql
178 .query_get_value(
179 "SELECT IFNULL(ephemeral_timer, 0) FROM chats WHERE id=?",
180 (self,),
181 )
182 .await?
183 .with_context(|| format!("Chat {self} not found"))?;
184 Ok(timer)
185 }
186
187 pub(crate) async fn inner_set_ephemeral_timer(
192 self,
193 context: &Context,
194 timer: Timer,
195 ) -> Result<()> {
196 ensure!(!self.is_special(), "Invalid chat ID");
197
198 context
199 .sql
200 .execute(
201 "UPDATE chats
202 SET ephemeral_timer=?
203 WHERE id=?;",
204 (timer, self),
205 )
206 .await?;
207
208 context.emit_event(EventType::ChatEphemeralTimerModified {
209 chat_id: self,
210 timer,
211 });
212 Ok(())
213 }
214
215 pub async fn set_ephemeral_timer(self, context: &Context, timer: Timer) -> Result<()> {
219 if timer == self.get_ephemeral_timer(context).await? {
220 return Ok(());
221 }
222 self.inner_set_ephemeral_timer(context, timer).await?;
223
224 if self.is_promoted(context).await? {
225 let mut msg = Message::new_text(
226 stock_ephemeral_timer_changed(context, timer, ContactId::SELF).await,
227 );
228 msg.param.set_cmd(SystemMessage::EphemeralTimerChanged);
229 if let Err(err) = send_msg(context, self, &mut msg).await {
230 error!(
231 context,
232 "Failed to send a message about ephemeral message timer change: {:?}", err
233 );
234 }
235 }
236 Ok(())
237 }
238}
239
240pub(crate) async fn stock_ephemeral_timer_changed(
242 context: &Context,
243 timer: Timer,
244 from_id: ContactId,
245) -> String {
246 match timer {
247 Timer::Disabled => stock_str::msg_ephemeral_timer_disabled(context, from_id).await,
248 Timer::Enabled { duration } => match duration {
249 0..=59 => {
250 stock_str::msg_ephemeral_timer_enabled(context, &timer.to_string(), from_id).await
251 }
252 60 => stock_str::msg_ephemeral_timer_minute(context, from_id).await,
253 61..=3599 => {
254 stock_str::msg_ephemeral_timer_minutes(
255 context,
256 &format!("{}", (f64::from(duration) / 6.0).round() / 10.0),
257 from_id,
258 )
259 .await
260 }
261 3600 => stock_str::msg_ephemeral_timer_hour(context, from_id).await,
262 3601..=86399 => {
263 stock_str::msg_ephemeral_timer_hours(
264 context,
265 &format!("{}", (f64::from(duration) / 360.0).round() / 10.0),
266 from_id,
267 )
268 .await
269 }
270 86400 => stock_str::msg_ephemeral_timer_day(context, from_id).await,
271 86401..=604_799 => {
272 stock_str::msg_ephemeral_timer_days(
273 context,
274 &format!("{}", (f64::from(duration) / 8640.0).round() / 10.0),
275 from_id,
276 )
277 .await
278 }
279 604_800 => stock_str::msg_ephemeral_timer_week(context, from_id).await,
280 _ => {
281 stock_str::msg_ephemeral_timer_weeks(
282 context,
283 &format!("{}", (f64::from(duration) / 60480.0).round() / 10.0),
284 from_id,
285 )
286 .await
287 }
288 },
289 }
290}
291
292impl MsgId {
293 pub(crate) async fn ephemeral_timer(self, context: &Context) -> Result<Timer> {
295 let res = match context
296 .sql
297 .query_get_value("SELECT ephemeral_timer FROM msgs WHERE id=?", (self,))
298 .await?
299 {
300 None | Some(0) => Timer::Disabled,
301 Some(duration) => Timer::Enabled { duration },
302 };
303 Ok(res)
304 }
305
306 pub(crate) async fn start_ephemeral_timer(self, context: &Context) -> Result<()> {
308 if let Timer::Enabled { duration } = self.ephemeral_timer(context).await? {
309 let ephemeral_timestamp = time().saturating_add(duration.into());
310
311 context
312 .sql
313 .execute(
314 "UPDATE msgs SET ephemeral_timestamp = ? \
315 WHERE (ephemeral_timestamp == 0 OR ephemeral_timestamp > ?) \
316 AND id = ?",
317 (ephemeral_timestamp, ephemeral_timestamp, self),
318 )
319 .await?;
320 context.scheduler.interrupt_ephemeral_task().await;
321 }
322 Ok(())
323 }
324}
325
326pub(crate) async fn start_ephemeral_timers_msgids(
327 context: &Context,
328 msg_ids: &[MsgId],
329) -> Result<()> {
330 let now = time();
331 let should_interrupt =
332 context
333 .sql
334 .transaction(move |transaction| {
335 let mut should_interrupt = false;
336 let mut stmt =
337 transaction.prepare(
338 "UPDATE msgs SET ephemeral_timestamp = ?1 + ephemeral_timer
339 WHERE (ephemeral_timestamp == 0 OR ephemeral_timestamp > ?1 + ephemeral_timer) AND ephemeral_timer > 0
340 AND id=?2")?;
341 for msg_id in msg_ids {
342 should_interrupt |= stmt.execute((now, msg_id))? > 0;
343 }
344 Ok(should_interrupt)
345 }).await?;
346 if should_interrupt {
347 context.scheduler.interrupt_ephemeral_task().await;
348 }
349 Ok(())
350}
351
352pub(crate) async fn start_chat_ephemeral_timers(context: &Context, chat_id: ChatId) -> Result<()> {
356 let now = time();
357 let should_interrupt = context
358 .sql
359 .execute(
360 "UPDATE msgs SET ephemeral_timestamp = ?1 + ephemeral_timer
361 WHERE chat_id = ?2
362 AND ephemeral_timer > 0
363 AND (ephemeral_timestamp == 0 OR ephemeral_timestamp > ?1 + ephemeral_timer)",
364 (now, chat_id),
365 )
366 .await?
367 > 0;
368 if should_interrupt {
369 context.scheduler.interrupt_ephemeral_task().await;
370 }
371 Ok(())
372}
373
374async fn select_expired_messages(
379 context: &Context,
380 now: i64,
381) -> Result<Vec<(MsgId, ChatId, Viewtype, u32)>> {
382 let mut rows = context
383 .sql
384 .query_map(
385 r#"
386SELECT id, chat_id, type, location_id
387FROM msgs
388WHERE
389 ephemeral_timestamp != 0
390 AND ephemeral_timestamp <= ?
391 AND chat_id != ?
392"#,
393 (now, DC_CHAT_ID_TRASH),
394 |row| {
395 let id: MsgId = row.get("id")?;
396 let chat_id: ChatId = row.get("chat_id")?;
397 let viewtype: Viewtype = row.get("type")?;
398 let location_id: u32 = row.get("location_id")?;
399 Ok((id, chat_id, viewtype, location_id))
400 },
401 |rows| rows.collect::<Result<Vec<_>, _>>().map_err(Into::into),
402 )
403 .await?;
404
405 if let Some(delete_device_after) = context.get_config_delete_device_after().await? {
406 let self_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::SELF)
407 .await?
408 .map(|c| c.id)
409 .unwrap_or_default();
410 let device_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::DEVICE)
411 .await?
412 .map(|c| c.id)
413 .unwrap_or_default();
414
415 let threshold_timestamp = now.saturating_sub(delete_device_after);
416
417 let rows_expired = context
418 .sql
419 .query_map(
420 r#"
421SELECT id, chat_id, type, location_id
422FROM msgs
423WHERE
424 timestamp < ?1
425 AND timestamp_rcvd < ?1
426 AND chat_id > ?
427 AND chat_id != ?
428 AND chat_id != ?
429"#,
430 (
431 threshold_timestamp,
432 DC_CHAT_ID_LAST_SPECIAL,
433 self_chat_id,
434 device_chat_id,
435 ),
436 |row| {
437 let id: MsgId = row.get("id")?;
438 let chat_id: ChatId = row.get("chat_id")?;
439 let viewtype: Viewtype = row.get("type")?;
440 let location_id: u32 = row.get("location_id")?;
441 Ok((id, chat_id, viewtype, location_id))
442 },
443 |rows| rows.collect::<Result<Vec<_>, _>>().map_err(Into::into),
444 )
445 .await?;
446
447 rows.extend(rows_expired);
448 }
449
450 Ok(rows)
451}
452
453pub(crate) async fn delete_expired_messages(context: &Context, now: i64) -> Result<()> {
459 let rows = select_expired_messages(context, now).await?;
460
461 if !rows.is_empty() {
462 info!(context, "Attempting to delete {} messages.", rows.len());
463
464 let (msgs_changed, webxdc_deleted) = context
465 .sql
466 .transaction(|transaction| {
467 let mut msgs_changed = Vec::with_capacity(rows.len());
468 let mut webxdc_deleted = Vec::new();
469 let mut del_msg_stmt = transaction.prepare(
472 "INSERT OR REPLACE INTO msgs (id, rfc724_mid, timestamp, chat_id)
473 SELECT ?1, rfc724_mid, timestamp, ? FROM msgs WHERE id=?1",
474 )?;
475 let mut del_location_stmt =
476 transaction.prepare("DELETE FROM locations WHERE independent=1 AND id=?")?;
477 for (msg_id, chat_id, viewtype, location_id) in rows {
478 del_msg_stmt.execute((msg_id, DC_CHAT_ID_TRASH))?;
479 if location_id > 0 {
480 del_location_stmt.execute((location_id,))?;
481 }
482
483 msgs_changed.push((chat_id, msg_id));
484 if viewtype == Viewtype::Webxdc {
485 webxdc_deleted.push(msg_id)
486 }
487 }
488 Ok((msgs_changed, webxdc_deleted))
489 })
490 .await?;
491
492 let mut modified_chat_ids = BTreeSet::new();
493
494 for (chat_id, msg_id) in msgs_changed {
495 context.emit_event(EventType::MsgDeleted { chat_id, msg_id });
496 modified_chat_ids.insert(chat_id);
497 }
498
499 for modified_chat_id in modified_chat_ids {
500 context.emit_msgs_changed_without_msg_id(modified_chat_id);
501 }
502
503 for msg_id in webxdc_deleted {
504 context.emit_event(EventType::WebxdcInstanceDeleted { msg_id });
505 }
506 }
507
508 Ok(())
509}
510
511async fn next_delete_device_after_timestamp(context: &Context) -> Result<Option<i64>> {
514 if let Some(delete_device_after) = context.get_config_delete_device_after().await? {
515 let self_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::SELF)
516 .await?
517 .map(|c| c.id)
518 .unwrap_or_default();
519 let device_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::DEVICE)
520 .await?
521 .map(|c| c.id)
522 .unwrap_or_default();
523
524 let oldest_message_timestamp: Option<i64> = context
525 .sql
526 .query_get_value(
527 r#"
528 SELECT min(max(timestamp, timestamp_rcvd))
529 FROM msgs
530 WHERE chat_id > ?
531 AND chat_id != ?
532 AND chat_id != ?
533 HAVING count(*) > 0
534 "#,
535 (DC_CHAT_ID_TRASH, self_chat_id, device_chat_id),
536 )
537 .await?;
538
539 Ok(oldest_message_timestamp.map(|x| x.saturating_add(delete_device_after)))
540 } else {
541 Ok(None)
542 }
543}
544
545async fn next_expiration_timestamp(context: &Context) -> Option<i64> {
550 let ephemeral_timestamp: Option<i64> = match context
551 .sql
552 .query_get_value(
553 r#"
554 SELECT min(ephemeral_timestamp)
555 FROM msgs
556 WHERE ephemeral_timestamp != 0
557 AND chat_id != ?
558 HAVING count(*) > 0
559 "#,
560 (DC_CHAT_ID_TRASH,), )
562 .await
563 {
564 Err(err) => {
565 warn!(context, "Can't calculate next ephemeral timeout: {}", err);
566 None
567 }
568 Ok(ephemeral_timestamp) => ephemeral_timestamp,
569 };
570
571 let delete_device_after_timestamp: Option<i64> =
572 match next_delete_device_after_timestamp(context).await {
573 Err(err) => {
574 warn!(
575 context,
576 "Can't calculate timestamp of the next message expiration: {}", err
577 );
578 None
579 }
580 Ok(timestamp) => timestamp,
581 };
582
583 ephemeral_timestamp
584 .into_iter()
585 .chain(delete_device_after_timestamp)
586 .min()
587}
588
589pub(crate) async fn ephemeral_loop(context: &Context, interrupt_receiver: Receiver<()>) {
590 loop {
591 let ephemeral_timestamp = next_expiration_timestamp(context).await;
592
593 let now = SystemTime::now();
594 let until = if let Some(ephemeral_timestamp) = ephemeral_timestamp {
595 UNIX_EPOCH
596 + Duration::from_secs(ephemeral_timestamp.try_into().unwrap_or(u64::MAX))
597 + Duration::from_secs(1)
598 } else {
599 now + Duration::from_secs(86400)
601 };
602
603 if let Ok(duration) = until.duration_since(now) {
604 info!(
605 context,
606 "Ephemeral loop waiting for deletion in {} or interrupt",
607 duration_to_str(duration)
608 );
609 match timeout(duration, interrupt_receiver.recv()).await {
610 Ok(Ok(())) => {
611 continue;
613 }
614 Ok(Err(err)) => {
615 warn!(
616 context,
617 "Interrupt channel closed, ephemeral loop exits now: {err:#}."
618 );
619 return;
620 }
621 Err(_err) => {
622 }
624 }
625 }
626
627 delete_expired_messages(context, time())
628 .await
629 .log_err(context)
630 .ok();
631
632 location::delete_expired(context, time())
633 .await
634 .log_err(context)
635 .ok();
636 }
637}
638
639pub(crate) async fn delete_expired_imap_messages(context: &Context) -> Result<()> {
641 let now = time();
642
643 let (threshold_timestamp, threshold_timestamp_extended) =
644 match context.get_config_delete_server_after().await? {
645 None => (0, 0),
646 Some(delete_server_after) => (
647 match delete_server_after {
648 0 => i64::MAX,
650 _ => now - delete_server_after,
651 },
652 now - max(delete_server_after, MIN_DELETE_SERVER_AFTER),
653 ),
654 };
655 let target = context.get_delete_msgs_target().await?;
656
657 context
658 .sql
659 .execute(
660 "UPDATE imap
661 SET target=?
662 WHERE rfc724_mid IN (
663 SELECT rfc724_mid FROM msgs
664 WHERE ((download_state = 0 AND timestamp < ?) OR
665 (download_state != 0 AND timestamp < ?) OR
666 (ephemeral_timestamp != 0 AND ephemeral_timestamp <= ?))
667 )",
668 (
669 &target,
670 threshold_timestamp,
671 threshold_timestamp_extended,
672 now,
673 ),
674 )
675 .await?;
676
677 Ok(())
678}
679
680pub(crate) async fn start_ephemeral_timers(context: &Context) -> Result<()> {
690 context
691 .sql
692 .execute(
693 "UPDATE msgs \
694 SET ephemeral_timestamp = ? + ephemeral_timer \
695 WHERE ephemeral_timer > 0 \
696 AND ephemeral_timestamp = 0 \
697 AND state NOT IN (?, ?, ?)",
698 (
699 time(),
700 MessageState::InFresh,
701 MessageState::InNoticed,
702 MessageState::OutDraft,
703 ),
704 )
705 .await?;
706
707 Ok(())
708}
709
710#[cfg(test)]
711mod ephemeral_tests;