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 31_536_000..=31_708_800 => stock_str::msg_ephemeral_timer_year(context, from_id).await,
281 _ => {
282 stock_str::msg_ephemeral_timer_weeks(
283 context,
284 &format!("{}", (f64::from(duration) / 60480.0).round() / 10.0),
285 from_id,
286 )
287 .await
288 }
289 },
290 }
291}
292
293impl MsgId {
294 pub(crate) async fn ephemeral_timer(self, context: &Context) -> Result<Timer> {
296 let res = match context
297 .sql
298 .query_get_value("SELECT ephemeral_timer FROM msgs WHERE id=?", (self,))
299 .await?
300 {
301 None | Some(0) => Timer::Disabled,
302 Some(duration) => Timer::Enabled { duration },
303 };
304 Ok(res)
305 }
306
307 pub(crate) async fn start_ephemeral_timer(self, context: &Context) -> Result<()> {
309 if let Timer::Enabled { duration } = self.ephemeral_timer(context).await? {
310 let ephemeral_timestamp = time().saturating_add(duration.into());
311
312 context
313 .sql
314 .execute(
315 "UPDATE msgs SET ephemeral_timestamp = ? \
316 WHERE (ephemeral_timestamp == 0 OR ephemeral_timestamp > ?) \
317 AND id = ?",
318 (ephemeral_timestamp, ephemeral_timestamp, self),
319 )
320 .await?;
321 context.scheduler.interrupt_ephemeral_task().await;
322 }
323 Ok(())
324 }
325}
326
327pub(crate) async fn start_ephemeral_timers_msgids(
328 context: &Context,
329 msg_ids: &[MsgId],
330) -> Result<()> {
331 let now = time();
332 let should_interrupt =
333 context
334 .sql
335 .transaction(move |transaction| {
336 let mut should_interrupt = false;
337 let mut stmt =
338 transaction.prepare(
339 "UPDATE msgs SET ephemeral_timestamp = ?1 + ephemeral_timer
340 WHERE (ephemeral_timestamp == 0 OR ephemeral_timestamp > ?1 + ephemeral_timer) AND ephemeral_timer > 0
341 AND id=?2")?;
342 for msg_id in msg_ids {
343 should_interrupt |= stmt.execute((now, msg_id))? > 0;
344 }
345 Ok(should_interrupt)
346 }).await?;
347 if should_interrupt {
348 context.scheduler.interrupt_ephemeral_task().await;
349 }
350 Ok(())
351}
352
353pub(crate) async fn start_chat_ephemeral_timers(context: &Context, chat_id: ChatId) -> Result<()> {
357 let now = time();
358 let should_interrupt = context
359 .sql
360 .execute(
361 "UPDATE msgs SET ephemeral_timestamp = ?1 + ephemeral_timer
362 WHERE chat_id = ?2
363 AND ephemeral_timer > 0
364 AND (ephemeral_timestamp == 0 OR ephemeral_timestamp > ?1 + ephemeral_timer)",
365 (now, chat_id),
366 )
367 .await?
368 > 0;
369 if should_interrupt {
370 context.scheduler.interrupt_ephemeral_task().await;
371 }
372 Ok(())
373}
374
375async fn select_expired_messages(
380 context: &Context,
381 now: i64,
382) -> Result<Vec<(MsgId, ChatId, Viewtype, u32)>> {
383 let mut rows = context
384 .sql
385 .query_map(
386 r#"
387SELECT id, chat_id, type, location_id
388FROM msgs
389WHERE
390 ephemeral_timestamp != 0
391 AND ephemeral_timestamp <= ?
392 AND chat_id != ?
393"#,
394 (now, DC_CHAT_ID_TRASH),
395 |row| {
396 let id: MsgId = row.get("id")?;
397 let chat_id: ChatId = row.get("chat_id")?;
398 let viewtype: Viewtype = row.get("type")?;
399 let location_id: u32 = row.get("location_id")?;
400 Ok((id, chat_id, viewtype, location_id))
401 },
402 |rows| rows.collect::<Result<Vec<_>, _>>().map_err(Into::into),
403 )
404 .await?;
405
406 if let Some(delete_device_after) = context.get_config_delete_device_after().await? {
407 let self_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::SELF)
408 .await?
409 .map(|c| c.id)
410 .unwrap_or_default();
411 let device_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::DEVICE)
412 .await?
413 .map(|c| c.id)
414 .unwrap_or_default();
415
416 let threshold_timestamp = now.saturating_sub(delete_device_after);
417
418 let rows_expired = context
419 .sql
420 .query_map(
421 r#"
422SELECT id, chat_id, type, location_id
423FROM msgs
424WHERE
425 timestamp < ?1
426 AND timestamp_rcvd < ?1
427 AND chat_id > ?
428 AND chat_id != ?
429 AND chat_id != ?
430"#,
431 (
432 threshold_timestamp,
433 DC_CHAT_ID_LAST_SPECIAL,
434 self_chat_id,
435 device_chat_id,
436 ),
437 |row| {
438 let id: MsgId = row.get("id")?;
439 let chat_id: ChatId = row.get("chat_id")?;
440 let viewtype: Viewtype = row.get("type")?;
441 let location_id: u32 = row.get("location_id")?;
442 Ok((id, chat_id, viewtype, location_id))
443 },
444 |rows| rows.collect::<Result<Vec<_>, _>>().map_err(Into::into),
445 )
446 .await?;
447
448 rows.extend(rows_expired);
449 }
450
451 Ok(rows)
452}
453
454pub(crate) async fn delete_expired_messages(context: &Context, now: i64) -> Result<()> {
460 let rows = select_expired_messages(context, now).await?;
461
462 if !rows.is_empty() {
463 info!(context, "Attempting to delete {} messages.", rows.len());
464
465 let (msgs_changed, webxdc_deleted) = context
466 .sql
467 .transaction(|transaction| {
468 let mut msgs_changed = Vec::with_capacity(rows.len());
469 let mut webxdc_deleted = Vec::new();
470 let mut del_msg_stmt = transaction.prepare(
473 "INSERT OR REPLACE INTO msgs (id, rfc724_mid, timestamp, chat_id)
474 SELECT ?1, rfc724_mid, timestamp, ? FROM msgs WHERE id=?1",
475 )?;
476 let mut del_location_stmt =
477 transaction.prepare("DELETE FROM locations WHERE independent=1 AND id=?")?;
478 for (msg_id, chat_id, viewtype, location_id) in rows {
479 del_msg_stmt.execute((msg_id, DC_CHAT_ID_TRASH))?;
480 if location_id > 0 {
481 del_location_stmt.execute((location_id,))?;
482 }
483
484 msgs_changed.push((chat_id, msg_id));
485 if viewtype == Viewtype::Webxdc {
486 webxdc_deleted.push(msg_id)
487 }
488 }
489 Ok((msgs_changed, webxdc_deleted))
490 })
491 .await?;
492
493 let mut modified_chat_ids = BTreeSet::new();
494
495 for (chat_id, msg_id) in msgs_changed {
496 context.emit_event(EventType::MsgDeleted { chat_id, msg_id });
497 modified_chat_ids.insert(chat_id);
498 }
499
500 for modified_chat_id in modified_chat_ids {
501 context.emit_msgs_changed_without_msg_id(modified_chat_id);
502 }
503
504 for msg_id in webxdc_deleted {
505 context.emit_event(EventType::WebxdcInstanceDeleted { msg_id });
506 }
507 }
508
509 Ok(())
510}
511
512async fn next_delete_device_after_timestamp(context: &Context) -> Result<Option<i64>> {
515 if let Some(delete_device_after) = context.get_config_delete_device_after().await? {
516 let self_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::SELF)
517 .await?
518 .map(|c| c.id)
519 .unwrap_or_default();
520 let device_chat_id = ChatIdBlocked::lookup_by_contact(context, ContactId::DEVICE)
521 .await?
522 .map(|c| c.id)
523 .unwrap_or_default();
524
525 let oldest_message_timestamp: Option<i64> = context
526 .sql
527 .query_get_value(
528 r#"
529 SELECT min(max(timestamp, timestamp_rcvd))
530 FROM msgs
531 WHERE chat_id > ?
532 AND chat_id != ?
533 AND chat_id != ?
534 HAVING count(*) > 0
535 "#,
536 (DC_CHAT_ID_TRASH, self_chat_id, device_chat_id),
537 )
538 .await?;
539
540 Ok(oldest_message_timestamp.map(|x| x.saturating_add(delete_device_after)))
541 } else {
542 Ok(None)
543 }
544}
545
546async fn next_expiration_timestamp(context: &Context) -> Option<i64> {
551 let ephemeral_timestamp: Option<i64> = match context
552 .sql
553 .query_get_value(
554 r#"
555 SELECT min(ephemeral_timestamp)
556 FROM msgs
557 WHERE ephemeral_timestamp != 0
558 AND chat_id != ?
559 HAVING count(*) > 0
560 "#,
561 (DC_CHAT_ID_TRASH,), )
563 .await
564 {
565 Err(err) => {
566 warn!(context, "Can't calculate next ephemeral timeout: {}", err);
567 None
568 }
569 Ok(ephemeral_timestamp) => ephemeral_timestamp,
570 };
571
572 let delete_device_after_timestamp: Option<i64> =
573 match next_delete_device_after_timestamp(context).await {
574 Err(err) => {
575 warn!(
576 context,
577 "Can't calculate timestamp of the next message expiration: {}", err
578 );
579 None
580 }
581 Ok(timestamp) => timestamp,
582 };
583
584 ephemeral_timestamp
585 .into_iter()
586 .chain(delete_device_after_timestamp)
587 .min()
588}
589
590pub(crate) async fn ephemeral_loop(context: &Context, interrupt_receiver: Receiver<()>) {
591 loop {
592 let ephemeral_timestamp = next_expiration_timestamp(context).await;
593
594 let now = SystemTime::now();
595 let until = if let Some(ephemeral_timestamp) = ephemeral_timestamp {
596 UNIX_EPOCH
597 + Duration::from_secs(ephemeral_timestamp.try_into().unwrap_or(u64::MAX))
598 + Duration::from_secs(1)
599 } else {
600 now + Duration::from_secs(86400)
602 };
603
604 if let Ok(duration) = until.duration_since(now) {
605 info!(
606 context,
607 "Ephemeral loop waiting for deletion in {} or interrupt",
608 duration_to_str(duration)
609 );
610 match timeout(duration, interrupt_receiver.recv()).await {
611 Ok(Ok(())) => {
612 continue;
614 }
615 Ok(Err(err)) => {
616 warn!(
617 context,
618 "Interrupt channel closed, ephemeral loop exits now: {err:#}."
619 );
620 return;
621 }
622 Err(_err) => {
623 }
625 }
626 }
627
628 delete_expired_messages(context, time())
629 .await
630 .log_err(context)
631 .ok();
632
633 location::delete_expired(context, time())
634 .await
635 .log_err(context)
636 .ok();
637 }
638}
639
640pub(crate) async fn delete_expired_imap_messages(context: &Context) -> Result<()> {
642 let now = time();
643
644 let (threshold_timestamp, threshold_timestamp_extended) =
645 match context.get_config_delete_server_after().await? {
646 None => (0, 0),
647 Some(delete_server_after) => (
648 match delete_server_after {
649 0 => i64::MAX,
651 _ => now - delete_server_after,
652 },
653 now - max(delete_server_after, MIN_DELETE_SERVER_AFTER),
654 ),
655 };
656 let target = context.get_delete_msgs_target().await?;
657
658 context
659 .sql
660 .execute(
661 "UPDATE imap
662 SET target=?
663 WHERE rfc724_mid IN (
664 SELECT rfc724_mid FROM msgs
665 WHERE ((download_state = 0 AND timestamp < ?) OR
666 (download_state != 0 AND timestamp < ?) OR
667 (ephemeral_timestamp != 0 AND ephemeral_timestamp <= ?))
668 )",
669 (
670 &target,
671 threshold_timestamp,
672 threshold_timestamp_extended,
673 now,
674 ),
675 )
676 .await?;
677
678 Ok(())
679}
680
681pub(crate) async fn start_ephemeral_timers(context: &Context) -> Result<()> {
691 context
692 .sql
693 .execute(
694 "UPDATE msgs \
695 SET ephemeral_timestamp = ? + ephemeral_timer \
696 WHERE ephemeral_timer > 0 \
697 AND ephemeral_timestamp = 0 \
698 AND state NOT IN (?, ?, ?)",
699 (
700 time(),
701 MessageState::InFresh,
702 MessageState::InNoticed,
703 MessageState::OutDraft,
704 ),
705 )
706 .await?;
707
708 Ok(())
709}
710
711#[cfg(test)]
712mod ephemeral_tests;