You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

654 lines
26KB

  1. /*
  2. * meli - imap module.
  3. *
  4. * Copyright 2019 Manos Pitsidianakis
  5. *
  6. * This file is part of meli.
  7. *
  8. * meli is free software: you can redistribute it and/or modify
  9. * it under the terms of the GNU General Public License as published by
  10. * the Free Software Foundation, either version 3 of the License, or
  11. * (at your option) any later version.
  12. *
  13. * meli is distributed in the hope that it will be useful,
  14. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. * GNU General Public License for more details.
  17. *
  18. * You should have received a copy of the GNU General Public License
  19. * along with meli. If not, see <http://www.gnu.org/licenses/>.
  20. */
  21. use super::*;
  22. use std::sync::{Arc, Mutex, RwLock};
  23. /// Arguments for IMAP watching functions
  24. pub struct ImapWatchKit {
  25. pub conn: ImapConnection,
  26. pub is_online: Arc<Mutex<bool>>,
  27. pub main_conn: Arc<Mutex<ImapConnection>>,
  28. pub uid_store: Arc<UIDStore>,
  29. pub folders: Arc<RwLock<FnvHashMap<FolderHash, ImapFolder>>>,
  30. pub sender: RefreshEventConsumer,
  31. pub work_context: WorkContext,
  32. }
  33. macro_rules! exit_on_error {
  34. ($sender:expr, $folder_hash:ident, $work_context:ident, $thread_id:ident, $($result:expr)+) => {
  35. $(if let Err(e) = $result {
  36. debug!("failure: {}", e.to_string());
  37. $work_context.set_status.send(($thread_id, e.to_string())).unwrap();
  38. $work_context.finished.send($thread_id).unwrap();
  39. $sender.send(RefreshEvent {
  40. hash: $folder_hash,
  41. kind: RefreshEventKind::Failure(e.clone()),
  42. });
  43. Err(e)
  44. } else { Ok(()) }?;)+
  45. };
  46. }
  47. pub fn poll_with_examine(kit: ImapWatchKit) -> Result<()> {
  48. debug!("poll with examine");
  49. let ImapWatchKit {
  50. is_online,
  51. mut conn,
  52. main_conn,
  53. uid_store,
  54. folders,
  55. sender,
  56. work_context,
  57. } = kit;
  58. loop {
  59. if *is_online.lock().unwrap() {
  60. break;
  61. }
  62. std::thread::sleep(std::time::Duration::from_millis(100));
  63. }
  64. let mut response = String::with_capacity(8 * 1024);
  65. let thread_id: std::thread::ThreadId = std::thread::current().id();
  66. loop {
  67. work_context
  68. .set_status
  69. .send((thread_id, "sleeping...".to_string()))
  70. .unwrap();
  71. std::thread::sleep(std::time::Duration::from_millis(5 * 60 * 1000));
  72. let folders = folders.read().unwrap();
  73. for folder in folders.values() {
  74. work_context
  75. .set_status
  76. .send((
  77. thread_id,
  78. format!("examining `{}` for updates...", folder.path()),
  79. ))
  80. .unwrap();
  81. examine_updates(folder, &sender, &mut conn, &uid_store, &work_context)?;
  82. }
  83. let mut main_conn = main_conn.lock().unwrap();
  84. main_conn.send_command(b"NOOP").unwrap();
  85. main_conn.read_response(&mut response).unwrap();
  86. }
  87. }
  88. pub fn idle(kit: ImapWatchKit) -> Result<()> {
  89. debug!("IDLE");
  90. /* IDLE only watches the connection's selected mailbox. We will IDLE on INBOX and every ~5
  91. * minutes wake up and poll the others */
  92. let ImapWatchKit {
  93. mut conn,
  94. is_online,
  95. main_conn,
  96. uid_store,
  97. folders,
  98. sender,
  99. work_context,
  100. } = kit;
  101. loop {
  102. if *is_online.lock().unwrap() {
  103. break;
  104. }
  105. std::thread::sleep(std::time::Duration::from_millis(100));
  106. }
  107. let thread_id: std::thread::ThreadId = std::thread::current().id();
  108. let folder: ImapFolder = match folders
  109. .read()
  110. .unwrap()
  111. .values()
  112. .find(|f| f.parent.is_none() && f.path().eq_ignore_ascii_case("INBOX"))
  113. .map(std::clone::Clone::clone)
  114. {
  115. Some(folder) => folder,
  116. None => {
  117. let err = MeliError::new("INBOX mailbox not found in local folder index. meli may have not parsed the IMAP folders correctly");
  118. debug!("failure: {}", err.to_string());
  119. work_context
  120. .set_status
  121. .send((thread_id, err.to_string()))
  122. .unwrap();
  123. sender.send(RefreshEvent {
  124. hash: 0,
  125. kind: RefreshEventKind::Failure(err.clone()),
  126. });
  127. return Err(err);
  128. }
  129. };
  130. let folder_hash = folder.hash();
  131. let mut response = String::with_capacity(8 * 1024);
  132. exit_on_error!(
  133. sender,
  134. folder_hash,
  135. work_context,
  136. thread_id,
  137. conn.send_command(format!("SELECT {}", folder.path()).as_bytes())
  138. conn.read_response(&mut response)
  139. );
  140. debug!("select response {}", &response);
  141. {
  142. let mut prev_exists = folder.exists.lock().unwrap();
  143. *prev_exists = match protocol_parser::select_response(&response) {
  144. Ok(ok) => {
  145. {
  146. let mut uidvalidities = uid_store.uidvalidity.lock().unwrap();
  147. if let Some(v) = uidvalidities.get_mut(&folder_hash) {
  148. if *v != ok.uidvalidity {
  149. sender.send(RefreshEvent {
  150. hash: folder_hash,
  151. kind: RefreshEventKind::Rescan,
  152. });
  153. uid_store.uid_index.lock().unwrap().clear();
  154. uid_store.hash_index.lock().unwrap().clear();
  155. uid_store.byte_cache.lock().unwrap().clear();
  156. *v = ok.uidvalidity;
  157. }
  158. } else {
  159. sender.send(RefreshEvent {
  160. hash: folder_hash,
  161. kind: RefreshEventKind::Rescan,
  162. });
  163. sender.send(RefreshEvent {
  164. hash: folder_hash,
  165. kind: RefreshEventKind::Failure(MeliError::new(format!(
  166. "Unknown mailbox: {} {}",
  167. folder.path(),
  168. folder_hash
  169. ))),
  170. });
  171. }
  172. }
  173. debug!(&ok);
  174. ok.exists
  175. }
  176. Err(e) => {
  177. debug!("{:?}", e);
  178. panic!("could not select mailbox");
  179. }
  180. };
  181. }
  182. exit_on_error!(
  183. sender,
  184. folder_hash,
  185. work_context,
  186. thread_id,
  187. conn.send_command(b"IDLE")
  188. );
  189. work_context
  190. .set_status
  191. .send((thread_id, "IDLEing".to_string()))
  192. .unwrap();
  193. let mut iter = ImapBlockingConnection::from(conn);
  194. let mut beat = std::time::Instant::now();
  195. let mut watch = std::time::Instant::now();
  196. /* duration interval to send heartbeat */
  197. let _26_mins = std::time::Duration::from_secs(26 * 60);
  198. /* duration interval to check other folders for changes */
  199. let _5_mins = std::time::Duration::from_secs(5 * 60);
  200. while let Some(line) = iter.next() {
  201. let now = std::time::Instant::now();
  202. if now.duration_since(beat) >= _26_mins {
  203. exit_on_error!(
  204. sender,
  205. folder_hash,
  206. work_context,
  207. thread_id,
  208. iter.conn.set_nonblocking(true)
  209. iter.conn.send_raw(b"DONE")
  210. iter.conn.read_response(&mut response)
  211. iter.conn.send_command(b"IDLE")
  212. iter.conn.set_nonblocking(false)
  213. main_conn.lock().unwrap().send_command(b"NOOP")
  214. main_conn.lock().unwrap().read_response(&mut response)
  215. );
  216. beat = now;
  217. }
  218. if now.duration_since(watch) >= _5_mins {
  219. /* Time to poll all inboxes */
  220. let mut conn = main_conn.lock().unwrap();
  221. for folder in folders.read().unwrap().values() {
  222. work_context
  223. .set_status
  224. .send((
  225. thread_id,
  226. format!("examining `{}` for updates...", folder.path()),
  227. ))
  228. .unwrap();
  229. exit_on_error!(
  230. sender,
  231. folder_hash,
  232. work_context,
  233. thread_id,
  234. examine_updates(folder, &sender, &mut conn, &uid_store, &work_context)
  235. );
  236. }
  237. work_context
  238. .set_status
  239. .send((thread_id, "done examining mailboxes.".to_string()))
  240. .unwrap();
  241. watch = now;
  242. }
  243. match protocol_parser::untagged_responses(line.as_slice())
  244. .to_full_result()
  245. .map_err(MeliError::from)
  246. {
  247. Ok(Some(Recent(r))) => {
  248. let mut conn = main_conn.lock().unwrap();
  249. work_context
  250. .set_status
  251. .send((thread_id, format!("got `{} RECENT` notification", r)))
  252. .unwrap();
  253. /* UID SEARCH RECENT */
  254. exit_on_error!(
  255. sender,
  256. folder_hash,
  257. work_context,
  258. thread_id,
  259. conn.send_command(b"EXAMINE INBOX")
  260. conn.read_response(&mut response)
  261. conn.send_command(b"UID SEARCH RECENT")
  262. conn.read_response(&mut response)
  263. );
  264. match protocol_parser::search_results_raw(response.as_bytes())
  265. .to_full_result()
  266. .map_err(MeliError::from)
  267. {
  268. Ok(&[]) => {
  269. debug!("UID SEARCH RECENT returned no results");
  270. }
  271. Ok(v) => {
  272. exit_on_error!(
  273. sender,
  274. folder_hash,
  275. work_context,
  276. thread_id,
  277. conn.send_command(
  278. &[b"UID FETCH", v, b"(FLAGS RFC822.HEADER)"]
  279. .join(&b' '),
  280. )
  281. conn.read_response(&mut response)
  282. );
  283. debug!(&response);
  284. match protocol_parser::uid_fetch_response(response.as_bytes())
  285. .to_full_result()
  286. .map_err(MeliError::from)
  287. {
  288. Ok(v) => {
  289. let len = v.len();
  290. let mut ctr = 0;
  291. for (uid, flags, b) in v {
  292. work_context
  293. .set_status
  294. .send((
  295. thread_id,
  296. format!("parsing {}/{} envelopes..", ctr, len),
  297. ))
  298. .unwrap();
  299. if let Ok(env) = Envelope::from_bytes(&b, flags) {
  300. ctr += 1;
  301. uid_store
  302. .hash_index
  303. .lock()
  304. .unwrap()
  305. .insert(env.hash(), (uid, folder_hash));
  306. uid_store.uid_index.lock().unwrap().insert(uid, env.hash());
  307. debug!(
  308. "Create event {} {} {}",
  309. env.hash(),
  310. env.subject(),
  311. folder.path(),
  312. );
  313. sender.send(RefreshEvent {
  314. hash: folder_hash,
  315. kind: Create(Box::new(env)),
  316. });
  317. }
  318. }
  319. work_context
  320. .set_status
  321. .send((thread_id, format!("parsed {}/{} envelopes.", ctr, len)))
  322. .unwrap();
  323. }
  324. Err(e) => {
  325. debug!(e);
  326. }
  327. }
  328. }
  329. Err(e) => {
  330. debug!(
  331. "UID SEARCH RECENT err: {}\nresp: {}",
  332. e.to_string(),
  333. &response
  334. );
  335. }
  336. }
  337. }
  338. Ok(Some(Expunge(n))) => {
  339. work_context
  340. .set_status
  341. .send((thread_id, format!("got `{} EXPUNGED` notification", n)))
  342. .unwrap();
  343. debug!("expunge {}", n);
  344. }
  345. Ok(Some(Exists(n))) => {
  346. let mut conn = main_conn.lock().unwrap();
  347. /* UID FETCH ALL UID, cross-ref, then FETCH difference headers
  348. * */
  349. let mut prev_exists = folder.exists.lock().unwrap();
  350. debug!("exists {}", n);
  351. work_context
  352. .set_status
  353. .send((
  354. thread_id,
  355. format!(
  356. "got `{} EXISTS` notification (EXISTS was previously {} for {}",
  357. n,
  358. *prev_exists,
  359. folder.path()
  360. ),
  361. ))
  362. .unwrap();
  363. if n > *prev_exists {
  364. exit_on_error!(
  365. sender,
  366. folder_hash,
  367. work_context,
  368. thread_id,
  369. conn.send_command(b"EXAMINE INBOX")
  370. conn.read_response(&mut response)
  371. conn.send_command(
  372. &[
  373. b"FETCH",
  374. format!("{}:{}", *prev_exists + 1, n).as_bytes(),
  375. b"(UID FLAGS RFC822.HEADER)",
  376. ]
  377. .join(&b' '),
  378. )
  379. conn.read_response(&mut response)
  380. );
  381. match protocol_parser::uid_fetch_response(response.as_bytes())
  382. .to_full_result()
  383. .map_err(MeliError::from)
  384. {
  385. Ok(v) => {
  386. let len = v.len();
  387. let mut ctr = 0;
  388. for (uid, flags, b) in v {
  389. work_context
  390. .set_status
  391. .send((
  392. thread_id,
  393. format!("parsing {}/{} envelopes..", ctr, len),
  394. ))
  395. .unwrap();
  396. if uid_store.uid_index.lock().unwrap().contains_key(&uid) {
  397. ctr += 1;
  398. continue;
  399. }
  400. if let Ok(env) = Envelope::from_bytes(&b, flags) {
  401. ctr += 1;
  402. uid_store
  403. .hash_index
  404. .lock()
  405. .unwrap()
  406. .insert(env.hash(), (uid, folder_hash));
  407. uid_store.uid_index.lock().unwrap().insert(uid, env.hash());
  408. debug!(
  409. "Create event {} {} {}",
  410. env.hash(),
  411. env.subject(),
  412. folder.path(),
  413. );
  414. sender.send(RefreshEvent {
  415. hash: folder_hash,
  416. kind: Create(Box::new(env)),
  417. });
  418. }
  419. }
  420. work_context
  421. .set_status
  422. .send((thread_id, format!("parsed {}/{} envelopes.", ctr, len)))
  423. .unwrap();
  424. }
  425. Err(e) => {
  426. debug!(e);
  427. }
  428. }
  429. *prev_exists = n;
  430. } else if n < *prev_exists {
  431. *prev_exists = n;
  432. }
  433. }
  434. Ok(None) | Err(_) => {}
  435. }
  436. work_context
  437. .set_status
  438. .send((thread_id, "IDLEing".to_string()))
  439. .unwrap();
  440. }
  441. debug!("IDLE connection dropped");
  442. let err: &str = iter.err().unwrap_or("Unknown reason.");
  443. work_context
  444. .set_status
  445. .send((thread_id, "IDLE connection dropped".to_string()))
  446. .unwrap();
  447. work_context.finished.send(thread_id).unwrap();
  448. sender.send(RefreshEvent {
  449. hash: folder_hash,
  450. kind: RefreshEventKind::Failure(MeliError::new(format!(
  451. "IDLE connection dropped: {}",
  452. &err
  453. ))),
  454. });
  455. Err(MeliError::new(format!("IDLE connection dropped: {}", err)))
  456. }
  457. fn examine_updates(
  458. folder: &ImapFolder,
  459. sender: &RefreshEventConsumer,
  460. conn: &mut ImapConnection,
  461. uid_store: &Arc<UIDStore>,
  462. work_context: &WorkContext,
  463. ) -> Result<()> {
  464. let thread_id: std::thread::ThreadId = std::thread::current().id();
  465. let folder_hash = folder.hash();
  466. debug!("examining folder {} {}", folder_hash, folder.path());
  467. let mut response = String::with_capacity(8 * 1024);
  468. exit_on_error!(
  469. sender,
  470. folder_hash,
  471. work_context,
  472. thread_id,
  473. conn.send_command(format!("EXAMINE {}", folder.path()).as_bytes())
  474. conn.read_response(&mut response)
  475. );
  476. match protocol_parser::select_response(&response) {
  477. Ok(ok) => {
  478. debug!(&ok);
  479. {
  480. let mut uidvalidities = uid_store.uidvalidity.lock().unwrap();
  481. if let Some(v) = uidvalidities.get_mut(&folder_hash) {
  482. if *v != ok.uidvalidity {
  483. sender.send(RefreshEvent {
  484. hash: folder_hash,
  485. kind: RefreshEventKind::Rescan,
  486. });
  487. uid_store.uid_index.lock().unwrap().clear();
  488. uid_store.hash_index.lock().unwrap().clear();
  489. uid_store.byte_cache.lock().unwrap().clear();
  490. *v = ok.uidvalidity;
  491. }
  492. } else {
  493. sender.send(RefreshEvent {
  494. hash: folder_hash,
  495. kind: RefreshEventKind::Rescan,
  496. });
  497. sender.send(RefreshEvent {
  498. hash: folder_hash,
  499. kind: RefreshEventKind::Failure(MeliError::new(format!(
  500. "Unknown mailbox: {} {}",
  501. folder.path(),
  502. folder_hash
  503. ))),
  504. });
  505. }
  506. }
  507. let mut prev_exists = folder.exists.lock().unwrap();
  508. let n = ok.exists;
  509. if ok.recent > 0 {
  510. {
  511. /* UID SEARCH RECENT */
  512. exit_on_error!(
  513. sender,
  514. folder_hash,
  515. work_context,
  516. thread_id,
  517. conn.send_command(b"UID SEARCH RECENT")
  518. conn.read_response(&mut response)
  519. );
  520. match protocol_parser::search_results_raw(response.as_bytes())
  521. .to_full_result()
  522. .map_err(MeliError::from)
  523. {
  524. Ok(&[]) => {
  525. debug!("UID SEARCH RECENT returned no results");
  526. }
  527. Ok(v) => {
  528. exit_on_error!(
  529. sender,
  530. folder_hash,
  531. work_context,
  532. thread_id,
  533. conn.send_command(
  534. &[b"UID FETCH", v, b"(FLAGS RFC822.HEADER)"]
  535. .join(&b' '),
  536. )
  537. conn.read_response(&mut response)
  538. );
  539. debug!(&response);
  540. match protocol_parser::uid_fetch_response(response.as_bytes())
  541. .to_full_result()
  542. .map_err(MeliError::from)
  543. {
  544. Ok(v) => {
  545. for (uid, flags, b) in v {
  546. if let Ok(env) = Envelope::from_bytes(&b, flags) {
  547. uid_store
  548. .hash_index
  549. .lock()
  550. .unwrap()
  551. .insert(env.hash(), (uid, folder_hash));
  552. uid_store
  553. .uid_index
  554. .lock()
  555. .unwrap()
  556. .insert(uid, env.hash());
  557. debug!(
  558. "Create event {} {} {}",
  559. env.hash(),
  560. env.subject(),
  561. folder.path(),
  562. );
  563. sender.send(RefreshEvent {
  564. hash: folder_hash,
  565. kind: Create(Box::new(env)),
  566. });
  567. }
  568. }
  569. }
  570. Err(e) => {
  571. debug!(e);
  572. }
  573. }
  574. }
  575. Err(e) => {
  576. debug!(
  577. "UID SEARCH RECENT err: {}\nresp: {}",
  578. e.to_string(),
  579. &response
  580. );
  581. }
  582. }
  583. }
  584. } else if n > *prev_exists {
  585. /* UID FETCH ALL UID, cross-ref, then FETCH difference headers
  586. * */
  587. debug!("exists {}", n);
  588. exit_on_error!(
  589. sender,
  590. folder_hash,
  591. work_context,
  592. thread_id,
  593. conn.send_command(
  594. &[
  595. b"FETCH",
  596. format!("{}:{}", *prev_exists + 1, n).as_bytes(),
  597. b"(UID FLAGS RFC822.HEADER)",
  598. ]
  599. .join(&b' '),
  600. )
  601. conn.read_response(&mut response)
  602. );
  603. match protocol_parser::uid_fetch_response(response.as_bytes())
  604. .to_full_result()
  605. .map_err(MeliError::from)
  606. {
  607. Ok(v) => {
  608. for (uid, flags, b) in v {
  609. if let Ok(env) = Envelope::from_bytes(&b, flags) {
  610. uid_store
  611. .hash_index
  612. .lock()
  613. .unwrap()
  614. .insert(env.hash(), (uid, folder_hash));
  615. uid_store.uid_index.lock().unwrap().insert(uid, env.hash());
  616. debug!(
  617. "Create event {} {} {}",
  618. env.hash(),
  619. env.subject(),
  620. folder.path(),
  621. );
  622. sender.send(RefreshEvent {
  623. hash: folder_hash,
  624. kind: Create(Box::new(env)),
  625. });
  626. }
  627. }
  628. }
  629. Err(e) => {
  630. debug!(e);
  631. }
  632. }
  633. *prev_exists = n;
  634. } else if n < *prev_exists {
  635. *prev_exists = n;
  636. }
  637. }
  638. Err(e) => {
  639. debug!("{:?}", e);
  640. panic!("could not select mailbox");
  641. }
  642. };
  643. Ok(())
  644. }