From bd10e15c650179fd56da69b0074ffaecb3a67db1 Mon Sep 17 00:00:00 2001 From: rustmailer Date: Wed, 1 Apr 2026 04:49:18 +0800 Subject: [PATCH] feat: use fjall to store detached emails and attachments --- Cargo.lock | 782 +++++----------------- Cargo.toml | 20 +- src/main.rs | 14 +- src/modules/account/migration.rs | 8 +- src/modules/{indexer => blob}/envelope.rs | 0 src/modules/{indexer => blob}/manager.rs | 53 +- src/modules/{indexer => blob}/mod.rs | 12 +- src/modules/blob/storage.rs | 184 +++++ src/modules/cache/imap/sync/flow.rs | 2 +- src/modules/cache/imap/sync/rebuild.rs | 15 +- src/modules/cache/imap/task.rs | 51 +- src/modules/common/periodic.rs | 25 +- src/modules/common/rustls.rs | 6 +- src/modules/context/mod.rs | 6 +- src/modules/dashboard/mod.rs | 6 +- src/modules/duckdb/build.rs | 2 +- src/modules/duckdb/init.rs | 19 +- src/modules/envelope/extractor.rs | 58 +- src/modules/imap/executor.rs | 4 +- src/modules/import/mod.rs | 1 - src/modules/indexer/attachment.rs | 299 --------- src/modules/indexer/eml.rs | 331 --------- src/modules/indexer/fields.rs | 27 - src/modules/indexer/schema.rs | 49 -- src/modules/mailbox/delete.rs | 8 +- src/modules/message/attachment.rs | 42 +- src/modules/message/content.rs | 2 +- src/modules/message/delete.rs | 8 +- src/modules/message/list.rs | 2 +- src/modules/message/search.rs | 2 +- src/modules/mod.rs | 2 +- src/modules/oauth2/refresh/mod.rs | 11 +- src/modules/oauth2/task.rs | 11 +- src/modules/rest/api/message.rs | 14 +- src/modules/settings/dir.rs | 13 +- src/modules/tasks/mod.rs | 22 +- 36 files changed, 569 insertions(+), 1542 deletions(-) rename src/modules/{indexer => blob}/envelope.rs (100%) rename src/modules/{indexer => blob}/manager.rs (89%) rename src/modules/{indexer => blob}/mod.rs (82%) create mode 100644 src/modules/blob/storage.rs delete mode 100644 src/modules/indexer/attachment.rs delete mode 100644 src/modules/indexer/eml.rs delete mode 100644 src/modules/indexer/fields.rs delete mode 100644 src/modules/indexer/schema.rs diff --git a/Cargo.lock b/Cargo.lock index e0fec3a..41ef030 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -193,15 +193,6 @@ version = "1.0.100" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" -[[package]] -name = "ar_archive_writer" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7eb93bbb63b9c227414f6eb3a0adfddca591a8ce1e9b60661bb08969b87e340b" -dependencies = [ - "object", -] - [[package]] name = "arbitrary" version = "1.4.2" @@ -211,12 +202,6 @@ dependencies = [ "derive_arbitrary", ] -[[package]] -name = "arc-swap" -version = "1.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457" - [[package]] name = "arrayref" version = "0.3.9" @@ -231,9 +216,9 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" [[package]] name = "arrow" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e4754a624e5ae42081f464514be454b39711daae0458906dacde5f4c632f33a8" +checksum = "d441fdda254b65f3e9025910eb2c2066b6295d9c8ed409522b8d2ace1ff8574c" dependencies = [ "arrow-arith", "arrow-array", @@ -252,9 +237,9 @@ dependencies = [ [[package]] name = "arrow-arith" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7b3141e0ec5145a22d8694ea8b6d6f69305971c4fa1c1a13ef0195aef2d678b" +checksum = "ced5406f8b720cc0bc3aa9cf5758f93e8593cda5490677aa194e4b4b383f9a59" dependencies = [ "arrow-array", "arrow-buffer", @@ -266,9 +251,9 @@ dependencies = [ [[package]] name = "arrow-array" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c8955af33b25f3b175ee10af580577280b4bd01f7e823d94c7cdef7cf8c9aef" +checksum = "772bd34cacdda8baec9418d80d23d0fb4d50ef0735685bd45158b83dfeb6e62d" dependencies = [ "ahash 0.8.12", "arrow-buffer", @@ -284,9 +269,9 @@ dependencies = [ [[package]] name = "arrow-buffer" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c697ddca96183182f35b3a18e50b9110b11e916d7b7799cbfd4d34662f2c56c2" +checksum = "898f4cf1e9598fdb77f356fdf2134feedfd0ee8d5a4e0a5f573e7d0aec16baa4" dependencies = [ "bytes 1.11.1", "half", @@ -296,9 +281,9 @@ dependencies = [ [[package]] name = "arrow-cast" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "646bbb821e86fd57189c10b4fcdaa941deaf4181924917b0daa92735baa6ada5" +checksum = "b0127816c96533d20fc938729f48c52d3e48f99717e7a0b5ade77d742510736d" dependencies = [ "arrow-array", "arrow-buffer", @@ -318,9 +303,9 @@ dependencies = [ [[package]] name = "arrow-csv" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8da746f4180004e3ce7b83c977daf6394d768332349d3d913998b10a120b790a" +checksum = "ca025bd0f38eeecb57c2153c0123b960494138e6a957bbda10da2b25415209fe" dependencies = [ "arrow-array", "arrow-cast", @@ -333,9 +318,9 @@ dependencies = [ [[package]] name = "arrow-data" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fdd994a9d28e6365aa78e15da3f3950c0fdcea6b963a12fa1c391afb637b304" +checksum = "42d10beeab2b1c3bb0b53a00f7c944a178b622173a5c7bcabc3cb45d90238df4" dependencies = [ "arrow-buffer", "arrow-schema", @@ -346,9 +331,9 @@ dependencies = [ [[package]] name = "arrow-ipc" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "abf7df950701ab528bf7c0cf7eeadc0445d03ef5d6ffc151eaae6b38a58feff1" +checksum = "609a441080e338147a84e8e6904b6da482cefb957c5cdc0f3398872f69a315d0" dependencies = [ "arrow-array", "arrow-buffer", @@ -360,9 +345,9 @@ dependencies = [ [[package]] name = "arrow-json" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ff8357658bedc49792b13e2e862b80df908171275f8e6e075c460da5ee4bf86" +checksum = "6ead0914e4861a531be48fe05858265cf854a4880b9ed12618b1d08cba9bebc8" dependencies = [ "arrow-array", "arrow-buffer", @@ -384,9 +369,9 @@ dependencies = [ [[package]] name = "arrow-ord" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7d8f1870e03d4cbed632959498bcc84083b5a24bded52905ae1695bd29da45b" +checksum = "763a7ba279b20b52dad300e68cfc37c17efa65e68623169076855b3a9e941ca5" dependencies = [ "arrow-array", "arrow-buffer", @@ -397,9 +382,9 @@ dependencies = [ [[package]] name = "arrow-row" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "18228633bad92bff92a95746bbeb16e5fc318e8382b75619dec26db79e4de4c0" +checksum = "e14fe367802f16d7668163ff647830258e6e0aeea9a4d79aaedf273af3bdcd3e" dependencies = [ "arrow-array", "arrow-buffer", @@ -410,18 +395,18 @@ dependencies = [ [[package]] name = "arrow-schema" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8c872d36b7bf2a6a6a2b40de9156265f0242910791db366a2c17476ba8330d68" +checksum = "c30a1365d7a7dc50cc847e54154e6af49e4c4b0fddc9f607b687f29212082743" dependencies = [ "bitflags", ] [[package]] name = "arrow-select" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68bf3e3efbd1278f770d67e5dc410257300b161b93baedb3aae836144edcaf4b" +checksum = "78694888660a9e8ac949853db393af2a8b8fc82c19ce333132dfa2e72cc1a7fe" dependencies = [ "ahash 0.8.12", "arrow-array", @@ -433,9 +418,9 @@ dependencies = [ [[package]] name = "arrow-string" -version = "57.3.0" +version = "58.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85e968097061b3c0e9fe3079cf2e703e487890700546b5b0647f60fca1b5a8d8" +checksum = "61e04a01f8bb73ce54437514c5fd3ee2aa3e8abe4c777ee5cc55853b1652f79e" dependencies = [ "arrow-array", "arrow-buffer", @@ -718,12 +703,6 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8" -[[package]] -name = "base64" -version = "0.21.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" - [[package]] name = "base64" version = "0.22.1" @@ -741,7 +720,6 @@ dependencies = [ "base64 0.22.1", "blake3", "bytes 1.11.1", - "cacache", "chrono", "clap", "codepage-strings", @@ -752,6 +730,7 @@ dependencies = [ "duckdb", "email_address", "encoding_rs", + "fjall", "futures", "gethostname", "governor", @@ -762,10 +741,10 @@ dependencies = [ "itertools", "itoa", "lettre", - "lru 0.16.3", + "lru", "mail-parser", "mail-send", - "memmap2 0.9.10", + "memmap2", "mimalloc", "mime_guess", "murmur3", @@ -795,7 +774,6 @@ dependencies = [ "serde_json", "snafu", "sysinfo", - "tantivy", "tempfile", "time 0.3.47", "timeago", @@ -829,15 +807,6 @@ version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" -[[package]] -name = "bitpacking" -version = "0.9.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c1d3e2bfd8d06048a179f7b17afc3188effa10385e7b00dc65af6aae732ea92" -dependencies = [ - "crunchy", -] - [[package]] name = "bitvec" version = "1.0.1" @@ -852,16 +821,16 @@ dependencies = [ [[package]] name = "blake3" -version = "1.8.3" +version = "1.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2468ef7d57b3fb7e16b576e8377cdbde2320c60e1491e961d11da40fc4f02a2d" +checksum = "4d2d5991425dfd0785aed03aedcf0b321d61975c9b5b3689c774a2610ae0b51e" dependencies = [ "arrayref", "arrayvec", "cc", "cfg-if", "constant_time_eq 0.4.2", - "cpufeatures 0.2.17", + "cpufeatures 0.3.0", ] [[package]] @@ -895,31 +864,6 @@ dependencies = [ "piper", ] -[[package]] -name = "bon" -version = "3.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ebeb9aaf9329dff6ceb65c689ca3db33dbf15f324909c60e4e5eef5701ce31b1" -dependencies = [ - "bon-macros", - "rustversion", -] - -[[package]] -name = "bon-macros" -version = "3.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77e9d642a7e3a318e37c2c9427b5a6a48aa1ad55dcd986f3034ab2239045a645" -dependencies = [ - "darling 0.21.3", - "ident_case", - "prettyplease", - "proc-macro2", - "quote", - "rustversion", - "syn 2.0.117", -] - [[package]] name = "borsh" version = "1.6.0" @@ -1004,6 +948,12 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" +[[package]] +name = "byteorder-lite" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" + [[package]] name = "bytes" version = "0.5.6" @@ -1016,6 +966,12 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "byteview" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c53ba0f290bfc610084c05582d9c5d421662128fc69f4bf236707af6fd321b9" + [[package]] name = "bzip2" version = "0.6.1" @@ -1025,33 +981,6 @@ dependencies = [ "libbz2-rs-sys", ] -[[package]] -name = "cacache" -version = "13.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c5063741c7b2e260bbede781cf4679632dd90e2718e99f7715e46824b65670b" -dependencies = [ - "digest 0.10.7", - "either", - "futures", - "hex", - "libc", - "memmap2 0.5.10", - "miette", - "reflink-copy", - "serde", - "serde_derive", - "serde_json", - "sha1 0.10.6", - "sha2 0.10.9", - "ssri", - "tempfile", - "thiserror 1.0.69", - "tokio", - "tokio-stream", - "walkdir", -] - [[package]] name = "camino" version = "1.2.1" @@ -1101,12 +1030,6 @@ dependencies = [ "shlex", ] -[[package]] -name = "census" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f4c707c6a209cbe82d10abd08e1ea8995e9ea937d2550646e02798948992be0" - [[package]] name = "cfg-if" version = "1.0.4" @@ -1144,16 +1067,6 @@ dependencies = [ "windows-link", ] -[[package]] -name = "chumsky" -version = "0.9.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8eebd66744a15ded14960ab4ccdbfb51ad3b81f51f3f04a80adac98c985396c9" -dependencies = [ - "hashbrown 0.14.5", - "stacker", -] - [[package]] name = "cipher" version = "0.2.5" @@ -1256,14 +1169,20 @@ checksum = "e0d05af1e006a2407bedef5af410552494ce5be9090444dbbcb57258c1af3d56" dependencies = [ "strum 0.26.3", "strum_macros 0.26.4", - "unicode-width 0.2.2", + "unicode-width", ] [[package]] -name = "compressed-rtf" -version = "1.0.0" +name = "compare" +version = "0.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50bd502b5779b9f999a2d69115f7e2fb1662883aa5a0e50e33cff684dae101d9" +checksum = "ea0095f6103c2a8b44acd6fd15960c801dafebf02e21940360833e0673f48ba7" + +[[package]] +name = "compressed-rtf" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51b7f2aaf6791c6517669240b82ef12c1283b39772f0cad3bc5c73f19f6e7f16" dependencies = [ "byteorder", "thiserror 2.0.17", @@ -1306,7 +1225,7 @@ checksum = "d64e8af5551369d19cf50138de61f1c42074ab970f74e99be916646777f8fc87" dependencies = [ "encode_unicode", "libc", - "unicode-width 0.2.2", + "unicode-width", "windows-sys 0.61.2", ] @@ -1444,16 +1363,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "crossbeam-deque" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" -dependencies = [ - "crossbeam-epoch", - "crossbeam-utils", -] - [[package]] name = "crossbeam-epoch" version = "0.9.18" @@ -1463,6 +1372,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-skiplist" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df29de440c58ca2cc6e587ec3d22347551a32435fbde9d2bff64e78a9ffa151b" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -1562,18 +1481,8 @@ version = "0.20.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc7f46116c46ff9ab3eb1597a45688b6715c6e628b5c133e288e709a29bcb4ee" dependencies = [ - "darling_core 0.20.11", - "darling_macro 0.20.11", -] - -[[package]] -name = "darling" -version = "0.21.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9cdf337090841a411e2a7f3deb9187445851f91b309c0c0a29e05f74a00a48c0" -dependencies = [ - "darling_core 0.21.3", - "darling_macro 0.21.3", + "darling_core", + "darling_macro", ] [[package]] @@ -1590,38 +1499,13 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "darling_core" -version = "0.21.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1247195ecd7e3c85f83c8d2a366e4210d588e802133e1e355180a9870b517ea4" -dependencies = [ - "fnv", - "ident_case", - "proc-macro2", - "quote", - "strsim", - "syn 2.0.117", -] - [[package]] name = "darling_macro" version = "0.20.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ - "darling_core 0.20.11", - "quote", - "syn 2.0.117", -] - -[[package]] -name = "darling_macro" -version = "0.21.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d38308df82d1080de0afee5d069fa14b0326a88c14f15c5ccda35b4a6c414c81" -dependencies = [ - "darling_core 0.21.3", + "darling_core", "quote", "syn 2.0.117", ] @@ -1673,7 +1557,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ececcb659e7ba858fb4f10388c250a7252eb0a27373f1a72b8748afdd248e587" dependencies = [ "powerfmt", - "serde_core", ] [[package]] @@ -1757,17 +1640,11 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "downcast-rs" -version = "2.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "117240f60069e65410b3ae1bb213295bd828f707b5bec6596a1afc8793ce0cbc" - [[package]] name = "duckdb" -version = "1.10500.0" +version = "1.10501.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a2048e4963a37ec458d9e93444192e0e949ac0a188d201ea53472f2c42db822" +checksum = "f13bc6d6487032fc2825a62ef8b4924b2378a2eb3166e132e5f3141ae9dd633f" dependencies = [ "arrow", "cast", @@ -1853,6 +1730,18 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "enum_dispatch" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd" +dependencies = [ + "once_cell", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -1917,12 +1806,6 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" -[[package]] -name = "fastdivide" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9afc2bd4d5a73106dd53d10d73d3401c2f32730ba2c0b93ddb888a8983680471" - [[package]] name = "fastrand" version = "1.9.0" @@ -1955,6 +1838,23 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a3076410a55c90011c298b04d0cfa770b00fa04e1e3c97d3f6c9de105a03844" +[[package]] +name = "fjall" +version = "3.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a9530ff159bc3ad3a15da746da0f6e95375c2ac64708cbb85ec1ebd26761a84" +dependencies = [ + "byteorder-lite", + "byteview", + "dashmap", + "flume 0.12.0", + "log", + "lsm-tree", + "lz4_flex", + "tempfile", + "xxhash-rust", +] + [[package]] name = "flatbuffers" version = "25.12.19" @@ -1987,6 +1887,15 @@ dependencies = [ "spinning_top 0.2.5", ] +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -2029,16 +1938,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "fs4" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8640e34b88f7652208ce9e88b1a37a2ae95227d84abec377ccd3c5cfeb141ed4" -dependencies = [ - "rustix", - "windows-sys 0.59.0", -] - [[package]] name = "fs_extra" version = "1.3.0" @@ -2341,10 +2240,6 @@ name = "hashbrown" version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" -dependencies = [ - "ahash 0.8.12", - "allocator-api2", -] [[package]] name = "hashbrown" @@ -2352,8 +2247,6 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "allocator-api2", - "equivalent", "foldhash 0.1.5", ] @@ -2539,7 +2432,7 @@ dependencies = [ "html5ever", "tendril", "thiserror 2.0.17", - "unicode-width 0.2.2", + "unicode-width", ] [[package]] @@ -2552,12 +2445,6 @@ dependencies = [ "markup5ever", ] -[[package]] -name = "htmlescape" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e9025058dae765dee5070ec375f591e2ba14638c63feff74f13805a72e523163" - [[package]] name = "http" version = "0.2.12" @@ -2714,15 +2601,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "hyperloglogplus" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "621debdf94dcac33e50475fdd76d34d5ea9c0362a834b9db08c3024696c1fbe3" -dependencies = [ - "serde", -] - [[package]] name = "iana-time-zone" version = "0.1.64" @@ -2917,6 +2795,15 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "interval-heap" +version = "0.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11274e5e8e89b8607cfedc2910b6626e998779b48a019151c7604d0adcb86ac6" +dependencies = [ + "compare", +] + [[package]] name = "ipconfig" version = "0.3.2" @@ -2961,7 +2848,7 @@ dependencies = [ "crossbeam-utils", "curl", "curl-sys", - "flume", + "flume 0.9.2", "futures-lite 1.13.0", "http 0.2.12", "log", @@ -2994,9 +2881,9 @@ dependencies = [ [[package]] name = "itoa" -version = "1.0.17" +version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "jobserver" @@ -3041,12 +2928,11 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" [[package]] name = "lettre" -version = "0.11.19" +version = "0.11.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e13e10e8818f8b2a60f52cb127041d388b89f3a96a62be9ceaffa22262fef7f" +checksum = "471816f3e24b85e820dee02cde962379ea1a669e5242f19c61bcbcffedf4c4fb" dependencies = [ "base64 0.22.1", - "chumsky", "email-encoding", "email_address", "fastrand 2.3.0", @@ -3064,12 +2950,6 @@ dependencies = [ "url", ] -[[package]] -name = "levenshtein_automata" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c2cdeb66e45e9f36bfad5bbdb4d2384e70936afbee843c6f6543f0c551ebb25" - [[package]] name = "lexical-core" version = "1.0.6" @@ -3141,9 +3021,9 @@ checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112" [[package]] name = "libduckdb-sys" -version = "1.10500.0" +version = "1.10501.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6df9db064ff17120305ec6b7aee048f766cdf2a3ce9ffbb6953aaa3634605030" +checksum = "12096c1694924782b3fe21e790630b77bacb4fcb7ad9d7ee0fec626f985bf248" dependencies = [ "cc", "flate2", @@ -3250,15 +3130,6 @@ dependencies = [ "value-bag", ] -[[package]] -name = "lru" -version = "0.12.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" -dependencies = [ - "hashbrown 0.15.5", -] - [[package]] name = "lru" version = "0.16.3" @@ -3284,10 +3155,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] -name = "lz4_flex" -version = "0.11.5" +name = "lsm-tree" +version = "3.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08ab2867e3eeeca90e844d1940eab391c9dc5228783db2ed999acbc0a9ed375a" +checksum = "9d67f95fd716870329c30aaeedf87f23d426564e6ce46efa045a91444faf2a19" +dependencies = [ + "byteorder-lite", + "bytes 1.11.1", + "byteview", + "crossbeam-skiplist", + "enum_dispatch", + "interval-heap", + "log", + "lz4_flex", + "quick_cache", + "rustc-hash", + "self_cell", + "sfa", + "tempfile", + "varint-rs", + "xxhash-rust", +] + +[[package]] +name = "lz4_flex" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db9a0d582c2874f68138a16ce1867e0ffde6c0bb0a0df85e1f36d04146db488a" +dependencies = [ + "twox-hash", +] [[package]] name = "lzma-rust2" @@ -3392,30 +3289,12 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae960838283323069879657ca3de837e9f7bbb4c7bf6ea7f1b290d5e9476d2e0" -[[package]] -name = "measure_time" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51c55d61e72fc3ab704396c5fa16f4c184db37978ae4e94ca8959693a235fc0e" -dependencies = [ - "log", -] - [[package]] name = "memchr" version = "2.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273" -[[package]] -name = "memmap2" -version = "0.5.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "83faa42c0a078c393f6b29d5db232d8be22776a891f8f56e5284faee4a20b327" -dependencies = [ - "libc", -] - [[package]] name = "memmap2" version = "0.9.10" @@ -3425,29 +3304,6 @@ dependencies = [ "libc", ] -[[package]] -name = "miette" -version = "5.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59bb584eaeeab6bd0226ccf3509a69d7936d148cf3d036ad350abe35e8c6856e" -dependencies = [ - "miette-derive", - "once_cell", - "thiserror 1.0.69", - "unicode-width 0.1.14", -] - -[[package]] -name = "miette-derive" -version = "5.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49e7bc1560b95a3c4a25d03de42fe76ca718ab92d1a22a55b9b4cf67b3ae635c" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.117", -] - [[package]] name = "mimalloc" version = "0.1.48" @@ -3541,12 +3397,6 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9252111cf132ba0929b6f8e030cac2a24b507f3a4d6db6fb2896f27b354c714b" -[[package]] -name = "murmurhash32" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2195bf6aa996a481483b29d62a7663eed3fe39600c460e323f8ff41e90bdd89b" - [[package]] name = "native-tls" version = "0.2.18" @@ -3816,15 +3666,6 @@ dependencies = [ "objc2-core-foundation", ] -[[package]] -name = "object" -version = "0.37.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe" -dependencies = [ - "memchr", -] - [[package]] name = "oem_cp" version = "1.3.0" @@ -3860,12 +3701,6 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" -[[package]] -name = "oneshot" -version = "0.1.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4ce411919553d3f9fa53a0880544cda985a112117a0444d5ff1e870a893d6ea" - [[package]] name = "opaque-debug" version = "0.3.1" @@ -3942,15 +3777,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "ownedbytes" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2fbd56f7631767e61784dc43f8580f403f4475bd4aaa4da003e6295e1bab4a7e" -dependencies = [ - "stable_deref_trait", -] - [[package]] name = "parking" version = "2.2.1" @@ -4201,7 +4027,7 @@ version = "5.1.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "41273b691a3d467a8c44d05506afba9f7b6bd56c9cdf80123de13fe52d7ec587" dependencies = [ - "darling 0.20.11", + "darling", "http 1.4.0", "indexmap", "mime", @@ -4314,16 +4140,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "psm" -version = "0.1.30" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3852766467df634d74f0b2d7819bf8dc483a0eb2e3b0f50f756f9cfe8b0d18d8" -dependencies = [ - "ar_archive_writer", - "cc", -] - [[package]] name = "ptr_meta" version = "0.1.4" @@ -4602,16 +4418,6 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c8d0fd677905edcbeedbf2edb6494d676f0e98d54d5cf9bda0b061cb8fb8aba" -[[package]] -name = "rand_distr" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32cb0b9bc82b0a0876c2dd994a7e7a2683d3e7390ca40e6886785ef0c7e3ee31" -dependencies = [ - "num-traits", - "rand 0.8.5", -] - [[package]] name = "rand_hc" version = "0.2.0" @@ -4630,26 +4436,6 @@ dependencies = [ "bitflags", ] -[[package]] -name = "rayon" -version = "1.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "368f01d005bf8fd9b1206fb6fa653e6c4a81ceb1466406b81792d87c5677a58f" -dependencies = [ - "either", - "rayon-core", -] - -[[package]] -name = "rayon-core" -version = "1.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" -dependencies = [ - "crossbeam-deque", - "crossbeam-utils", -] - [[package]] name = "rcgen" version = "0.14.7" @@ -4740,18 +4526,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "reflink-copy" -version = "0.1.28" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23bbed272e39c47a095a5242218a67412a220006842558b03fe2935e8f3d7b92" -dependencies = [ - "cfg-if", - "libc", - "rustix", - "windows", -] - [[package]] name = "regex" version = "1.12.3" @@ -4924,16 +4698,6 @@ dependencies = [ "walkdir", ] -[[package]] -name = "rust-stemmers" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e46a2036019fdb888131db7a4c847a1063a7493f971ed94ea82c67eada63ca54" -dependencies = [ - "serde", - "serde_derive", -] - [[package]] name = "rust_decimal" version = "1.40.0" @@ -5260,14 +5024,14 @@ dependencies = [ ] [[package]] -name = "sha-1" -version = "0.10.1" +name = "sfa" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f5058ada175748e33390e40e872bd0fe59a19f265d0158daa551c5a88a76009c" +checksum = "a1296838937cab56cd6c4eeeb8718ec777383700c33f060e2869867bd01d1175" dependencies = [ - "cfg-if", - "cpufeatures 0.2.17", - "digest 0.10.7", + "byteorder-lite", + "log", + "xxhash-rust", ] [[package]] @@ -5383,15 +5147,6 @@ dependencies = [ "walkdir", ] -[[package]] -name = "sketches-ddsketch" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c1e9a774a6c28142ac54bb25d25562e6bcf957493a184f15ad4eebccb23e410a" -dependencies = [ - "serde", -] - [[package]] name = "slab" version = "0.4.11" @@ -5467,6 +5222,9 @@ name = "spin" version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] [[package]] name = "spinning_top" @@ -5486,42 +5244,12 @@ dependencies = [ "lock_api", ] -[[package]] -name = "ssri" -version = "9.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da7a2b3c2bc9693bcb40870c4e9b5bf0d79f9cb46273321bf855ec513e919082" -dependencies = [ - "base64 0.21.7", - "digest 0.10.7", - "hex", - "miette", - "serde", - "sha-1", - "sha2 0.10.9", - "thiserror 1.0.69", - "xxhash-rust", -] - [[package]] name = "stable_deref_trait" version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" -[[package]] -name = "stacker" -version = "0.1.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08d74a23609d509411d10e2176dc2a4346e3b4aea2e7b1869f19fdedbc71c013" -dependencies = [ - "cc", - "cfg-if", - "libc", - "psm", - "windows-sys 0.59.0", -] - [[package]] name = "standback" version = "0.2.17" @@ -5753,156 +5481,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" -[[package]] -name = "tantivy" -version = "0.25.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "502915c7381c5cb2d2781503962610cb880ad8f1a0ca95df1bae645d5ebf2545" -dependencies = [ - "aho-corasick", - "arc-swap", - "base64 0.22.1", - "bitpacking", - "bon", - "byteorder", - "census", - "crc32fast", - "crossbeam-channel", - "downcast-rs", - "fastdivide", - "fnv", - "fs4", - "futures-channel", - "futures-util", - "htmlescape", - "hyperloglogplus", - "itertools", - "levenshtein_automata", - "log", - "lru 0.12.5", - "lz4_flex", - "measure_time", - "memmap2 0.9.10", - "once_cell", - "oneshot", - "rayon", - "regex", - "rust-stemmers", - "rustc-hash", - "serde", - "serde_json", - "sketches-ddsketch", - "smallvec", - "tantivy-bitpacker", - "tantivy-columnar", - "tantivy-common", - "tantivy-fst", - "tantivy-query-grammar", - "tantivy-sstable", - "tantivy-stacker", - "tantivy-tokenizer-api", - "tempfile", - "thiserror 2.0.17", - "time 0.3.47", - "uuid", - "winapi", - "zstd", -] - -[[package]] -name = "tantivy-bitpacker" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3b04eed5108d8283607da6710fe17a7663523440eaf7ea5a1a440d19a1448b6" -dependencies = [ - "bitpacking", -] - -[[package]] -name = "tantivy-columnar" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b628488ae936c83e92b5c4056833054ca56f76c0e616aee8339e24ac89119cd" -dependencies = [ - "downcast-rs", - "fastdivide", - "itertools", - "serde", - "tantivy-bitpacker", - "tantivy-common", - "tantivy-sstable", - "tantivy-stacker", -] - -[[package]] -name = "tantivy-common" -version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f880aa7cab0c063a47b62596d10991cdd0b6e0e0575d9c5eeb298b307a25de55" -dependencies = [ - "async-trait", - "byteorder", - "ownedbytes", - "serde", - "time 0.3.47", -] - -[[package]] -name = "tantivy-fst" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d60769b80ad7953d8a7b2c70cdfe722bbcdcac6bccc8ac934c40c034d866fc18" -dependencies = [ - "byteorder", - "regex-syntax", - "utf8-ranges", -] - -[[package]] -name = "tantivy-query-grammar" -version = "0.25.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "768fccdc84d60d86235d42d7e4c33acf43c418258ff5952abf07bd7837fcd26b" -dependencies = [ - "nom 7.1.3", - "serde", - "serde_json", -] - -[[package]] -name = "tantivy-sstable" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8292095d1a8a2c2b36380ec455f910ab52dde516af36321af332c93f20ab7d5" -dependencies = [ - "futures-util", - "itertools", - "tantivy-bitpacker", - "tantivy-common", - "tantivy-fst", - "zstd", -] - -[[package]] -name = "tantivy-stacker" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23d38a379411169f0b3002c9cba61cdfe315f757e9d4f239c00c282497a0749d" -dependencies = [ - "murmurhash32", - "rand_distr", - "tantivy-common", -] - -[[package]] -name = "tantivy-tokenizer-api" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23024f6aeb25ceb1a0e27740c84bdb0fae52626737b7e9a9de6ad5aa25c7b038" -dependencies = [ - "serde", -] - [[package]] name = "tap" version = "1.0.1" @@ -6435,6 +6013,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "twox-hash" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c" + [[package]] name = "typenum" version = "1.19.0" @@ -6477,12 +6061,6 @@ dependencies = [ "tinyvec", ] -[[package]] -name = "unicode-width" -version = "0.1.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" - [[package]] name = "unicode-width" version = "0.2.2" @@ -6553,12 +6131,6 @@ dependencies = [ "regex", ] -[[package]] -name = "utf8-ranges" -version = "1.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7fcfc827f90e53a02eaef5e535ee14266c1d569214c6aa70133a624d8a3164ba" - [[package]] name = "utf8_iter" version = "1.0.4" @@ -6573,9 +6145,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "uuid" -version = "1.22.0" +version = "1.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a68d3c8f01c0cfa54a75291d83601161799e4a89a39e0929f4b0354d88757a37" +checksum = "5ac8b6f42ead25368cf5b098aeb3dc8a1a2c05a3eee8a9a1a68c640edbfc79d9" dependencies = [ "getrandom 0.4.1", "js-sys", @@ -6595,6 +6167,12 @@ version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ba6f5989077681266825251a52748b8c1d8a4ad098cc37e440103d0ea717fc0" +[[package]] +name = "varint-rs" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f54a172d0620933a27a4360d3db3e2ae0dd6cceae9730751a036bbf182c4b23" + [[package]] name = "vcpkg" version = "0.2.15" diff --git a/Cargo.toml b/Cargo.toml index c3c1178..cd7ea22 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,10 +79,6 @@ oauth2 = { version = "5.0.0", features = ["reqwest-blocking"] } url = { version = "2.5.8", features = ["serde"] } sysinfo = "0.38.4" num_cpus = "1.17.0" -cacache = { version = "13.1.0", default-features = false, features = [ - "tokio-runtime", - "mmap", -] } rand = "0.10.0" encoding_rs = "0.8.35" async-imap = { version = "0.11.2", default-features = false, features = [ @@ -111,8 +107,7 @@ dashmap = "6.1.0" # Statically links OpenSSL by compiling from source, avoiding system library dependencies openssl-sys = { version = "0.9.112", optional = true, features = ["vendored"] } gethostname = "1.1.0" -tantivy = { version = "0.25.0", features = ["quickwit", "zstd-compression"] } -itoa = "1.0.17" +itoa = "1.0.18" html2text = "0.16.7" bytes = "1.11.1" dialoguer = "0.12.0" @@ -120,25 +115,26 @@ console = "0.16.3" toml = "0.9.8" memmap2 = "0.9.10" outlook-pst = { git = "https://github.com/rustmailer/outlook-pst-rs.git", branch = "main" } -compressed-rtf = "1.0.0" +compressed-rtf = "1.0.1" codepage-strings = "1.0.2" mail-send = "0.5.2" -duckdb = { version = "1.10500.0", features = [ +duckdb = { version = "1.10501.0", features = [ "chrono", "bundled", "r2d2", "appender-arrow", ] } -arrow = { version = "57", features = ["ffi"] } +arrow = { version = "58", features = ["ffi"] } r2d2 = { version = "0.8", default-features = false } refinery = { version = "0.9", default-features = false } refinery-core = { version = "0.9", default-features = false } rcgen = "0.14.7" rustls-pemfile = "2.2.0" -blake3 = "1.8.3" -uuid = { version = "1.22.0", features = ["v4", "serde"] } +blake3 = "1.8.4" +uuid = { version = "1.23.0", features = ["v4", "serde"] } +fjall = { version = "3.1.2", features = ["lz4", "metrics", "bytes_1"] } [dev-dependencies] #bincode = "1.3.3" #secret-lib = "1.0.0" tempfile = "3.27.0" -lettre = "0.11.19" +lettre = "0.11.20" diff --git a/src/main.rs b/src/main.rs index 7f0633c..08f3119 100644 --- a/src/main.rs +++ b/src/main.rs @@ -19,7 +19,9 @@ use bichon::{ bichon_version, modules::{ - common::rustls::RustMailerTls, + blob::{manager::ENVELOPE_INDEX_MANAGER, storage::BLOB_MANAGER}, + cache::imap::task::SYNC_TASKS, + common::rustls::BichonTls, context::{executors::BichonContext, Initialize}, duckdb::init::DuckDBManager, error::{code::ErrorCode, BichonResult}, @@ -64,6 +66,8 @@ async fn main() -> BichonResult<()> { return Err(error); } + let periodic_tasks = PeriodicTasks::setup(); + let mut smtp_service: Option = None; if SETTINGS.bichon_enable_smtp { info!("SMTP service is enabled, starting..."); @@ -82,6 +86,7 @@ async fn main() -> BichonResult<()> { } start_http_server().await?; + periodic_tasks.shutdown().await; if let Some(server) = smtp_service { info!("Shutting down SMTP server..."); @@ -89,19 +94,20 @@ async fn main() -> BichonResult<()> { info!("SMTP server stopped."); } + SYNC_TASKS.shutdown().await; + ENVELOPE_INDEX_MANAGER.shutdown().await; + BLOB_MANAGER.shutdown().await; info!("Bichon server stopped."); Ok(()) } /// Initialize the system by validating settings and starting necessary tasks. async fn initialize() -> BichonResult<()> { - // SETTINGS.validate()?; SignalManager::initialize().await?; DataDirManager::initialize().await?; DuckDBManager::initialize().await?; UserManager::initialize().await?; - RustMailerTls::initialize().await?; + BichonTls::initialize().await?; BichonContext::initialize().await?; - PeriodicTasks::start_background_tasks(); Ok(()) } diff --git a/src/modules/account/migration.rs b/src/modules/account/migration.rs index af0abef..59891c6 100644 --- a/src/modules/account/migration.rs +++ b/src/modules/account/migration.rs @@ -32,13 +32,10 @@ use crate::{ since::{DateSince, RelativeDate}, state::AccountRunningState, }, + blob::{manager::ENVELOPE_INDEX_MANAGER, storage::BLOB_MANAGER}, cache::imap::mailbox::MailBox, database::{list_all_impl, secondary_find_impl, with_transaction}, error::BichonResult, - indexer::{ - attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, - manager::ENVELOPE_INDEX_MANAGER, - }, users::{role::DEFAULT_ACCOUNT_MANAGER_ROLE_ID, UserModel, DEFAULT_ADMIN_USER_ID}, }, utc_now, @@ -365,8 +362,7 @@ impl AccountV4 { let content_hashes = ENVELOPE_INDEX_MANAGER .delete_account_envelopes(account.id) .await?; - EML_INDEX_MANAGER.delete(&content_hashes).await?; - ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + BLOB_MANAGER.delete(&content_hashes, &content_hashes)?; Self::delete_account(account.id).await?; info!("Sequential cleanup completed for account: {}", account.id); Ok(()) diff --git a/src/modules/indexer/envelope.rs b/src/modules/blob/envelope.rs similarity index 100% rename from src/modules/indexer/envelope.rs rename to src/modules/blob/envelope.rs diff --git a/src/modules/indexer/manager.rs b/src/modules/blob/manager.rs similarity index 89% rename from src/modules/indexer/manager.rs rename to src/modules/blob/manager.rs index b40066d..038d36e 100644 --- a/src/modules/indexer/manager.rs +++ b/src/modules/blob/manager.rs @@ -33,39 +33,46 @@ use crate::modules::{ }; use crate::{ modules::{ + blob::envelope::Envelope, common::signal::SIGNAL_MANAGER, dashboard::{DashboardStats, LargestEmail}, error::{code::ErrorCode, BichonResult}, - indexer::envelope::Envelope, message::search::SearchFilter, rest::response::DataPage, }, raise_error, }; -use tokio::{sync::mpsc, task}; +use tokio::{ + sync::{mpsc, Mutex}, + task::{self, JoinHandle}, +}; pub static ENVELOPE_INDEX_MANAGER: LazyLock = LazyLock::new(EnvelopeIndexManager::new); -pub const ENVELOPE_BATCH_SIZE: usize = 500; -pub const EML_BATCH_SIZE: usize = 100; +pub const ENVELOPE_BATCH_SIZE: usize = 100; const MAX_BUFFER_DURATION: Duration = Duration::from_secs(10); -pub enum MetadataOp { - Record((Envelope, Vec)), - Shutdown, -} - pub struct EnvelopeIndexManager { - sender: mpsc::Sender, + sender: mpsc::Sender<(Envelope, Vec)>, + handle: Mutex>>, } impl EnvelopeIndexManager { + pub async fn shutdown(&self) { + let mut guard = self.handle.lock().await; + if let Some(handle) = guard.take() { + tracing::info!("Waiting for EnvelopeIndexManager to sync all data..."); + let _ = handle.await; + tracing::info!("EnvelopeIndexManager synchronized and closed."); + } + } + pub fn new() -> Self { - let (sender, mut receiver) = mpsc::channel::(1000); - task::spawn(async move { + let (sender, mut receiver) = mpsc::channel::<(Envelope, Vec)>(1000); + let handle = task::spawn(async move { let mut buffer: Vec<(Envelope, Vec)> = Vec::with_capacity(ENVELOPE_BATCH_SIZE); let mut interval = tokio::time::interval(MAX_BUFFER_DURATION); @@ -74,17 +81,19 @@ impl EnvelopeIndexManager { tokio::select! { maybe_msg = receiver.recv() => { match maybe_msg { - Some(MetadataOp::Record(doc)) => { + Some(doc) => { buffer.push(doc); if buffer.len() >= ENVELOPE_BATCH_SIZE { ENVELOPE_INDEX_MANAGER.drain_and_commit(&mut buffer).await; } } - Some(MetadataOp::Shutdown) => { - ENVELOPE_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + None => { + if !buffer.is_empty() { + tracing::info!("Channel closed, flushing remaining {} items", buffer.len()); + ENVELOPE_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + } break; - } - None => break, + }, } } _ = interval.tick() => { @@ -93,16 +102,20 @@ impl EnvelopeIndexManager { } } _ = shutdown.recv() => { - let _ = ENVELOPE_INDEX_MANAGER.sender.send(MetadataOp::Shutdown).await; + ENVELOPE_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + break; } } } }); - Self { sender } + Self { + sender, + handle: Mutex::new(Some(handle)), + } } pub async fn add_document(&self, doc: (Envelope, Vec)) { - let _ = self.sender.send(MetadataOp::Record(doc)).await; + let _ = self.sender.send(doc).await; } async fn drain_and_commit(&self, buffer: &mut Vec<(Envelope, Vec)>) { diff --git a/src/modules/indexer/mod.rs b/src/modules/blob/mod.rs similarity index 82% rename from src/modules/indexer/mod.rs rename to src/modules/blob/mod.rs index 8cab6d1..0aab906 100644 --- a/src/modules/indexer/mod.rs +++ b/src/modules/blob/mod.rs @@ -16,16 +16,6 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use tantivy::TantivyDocument; - -pub mod attachment; -pub mod eml; pub mod envelope; -pub mod fields; pub mod manager; -pub mod schema; - -pub enum DocumentOp { - Document((String, TantivyDocument)), - Shutdown, -} +pub mod storage; diff --git a/src/modules/blob/storage.rs b/src/modules/blob/storage.rs new file mode 100644 index 0000000..64617a9 --- /dev/null +++ b/src/modules/blob/storage.rs @@ -0,0 +1,184 @@ +use crate::modules::{ + common::signal::SIGNAL_MANAGER, + envelope::extractor::reattach_eml_content, + error::{code::ErrorCode, BichonResult}, + settings::dir::DATA_DIR_MANAGER, +}; +use crate::raise_error; +use bytes::Bytes; +use fjall::{CompressionType, Database, Keyspace, KeyspaceCreateOptions, KvSeparationOptions}; + +use std::{io::Cursor, sync::LazyLock}; +use tokio::{ + sync::{mpsc, Mutex}, + task::{self, JoinHandle}, +}; + +pub static BLOB_MANAGER: LazyLock = LazyLock::new(BlobManager::new); + +pub struct DetachedEmail { + pub email: (String, Bytes), + pub attachments: Option>, +} + +pub struct BlobManager { + sender: mpsc::Sender, + db: Database, + email_keyspace: Keyspace, + attachments_keyspace: Keyspace, + handle: Mutex>>, +} + +impl BlobManager { + pub async fn shutdown(&self) { + let mut guard = self.handle.lock().await; + if let Some(handle) = guard.take() { + let _ = handle.await; + } + } + + pub fn new() -> Self { + let db = Database::builder(&DATA_DIR_MANAGER.eml_dir) + .open() + .expect("Failed to initialize Fjall database: Check if the directory exists and has write permissions."); + let email_keyspace = db + .keyspace("email", || { + KeyspaceCreateOptions::default() + .with_kv_separation(Some( + KvSeparationOptions::default() + .separation_threshold(0) + .compression(CompressionType::Lz4) + .file_target_size(128 * 1024 * 1024) + .staleness_threshold(0.5) + .age_cutoff(0.6), + )) + .max_memtable_size(64 * 1024 * 1024) + }) + .expect("Failed to open 'email' keyspace: The partition metadata might be corrupted or inaccessible."); + + let attachments_keyspace = db + .keyspace("attachments", || { + KeyspaceCreateOptions::default() + .with_kv_separation(Some( + KvSeparationOptions::default() + .separation_threshold(0) + .compression(CompressionType::Lz4) + .file_target_size(256 * 1024 * 1024) + .staleness_threshold(0.5) + .age_cutoff(0.6), + )) + .max_memtable_size(64 * 1024 * 1024) + }) + .expect("Failed to open 'attachments' keyspace: Check disk space for blob storage initialization."); + + let (sender, mut receiver) = mpsc::channel::(100); + + let store = db.clone(); + let email_keyspace_clone = email_keyspace.clone(); + let attachments_keyspace_clone = attachments_keyspace.clone(); + let handler = task::spawn(async move { + let mut shutdown = SIGNAL_MANAGER.subscribe(); + loop { + tokio::select! { + res = receiver.recv() => { + match res { + Some(email) => { + let mut batch = store.batch(); + batch.insert(&email_keyspace_clone, email.email.0, email.email.1); + if let Some(attachments) = email.attachments { + for a in attachments { + batch.insert(&attachments_keyspace_clone,a.0, a.1); + } + } + if let Err(e) = batch.commit() { + tracing::error!("Fjall Put Error {:?}", e); + } else { + tracing::info!("Fjall Put Success"); + } + } + None => { + tracing::info!("BlobManager: All senders dropped, closing storage."); + break; + } + } + } + _ = shutdown.recv() => { + receiver.close(); + let remaining = receiver.len(); + tracing::info!( + "BlobManager: Shutdown signal received. Processing {} remaining tasks...", + remaining + ); + + while let Some(email) = receiver.recv().await { + let mut batch = store.batch(); + batch.insert(&email_keyspace_clone, email.email.0, email.email.1); + if let Some(attachments) = email.attachments { + for a in attachments { + batch.insert(&attachments_keyspace_clone,a.0, a.1); + } + } + if let Err(e) = batch.commit() { + tracing::error!("Fjall Put Error {:?}", e); + } else { + tracing::info!("Fjall Put Success"); + } + } + + tracing::info!("BlobManager: All remaining tasks processed. Closing Fjall."); + break; + } + } + } + }); + + Self { + sender, + db, + email_keyspace, + attachments_keyspace, + handle: Mutex::new(Some(handler)), + } + } + + pub async fn queue(&self, email: DetachedEmail) { + let _ = self.sender.send(email).await; + } + + pub fn get_email(&self, content_hash: &str) -> BichonResult> { + self.email_keyspace + .get(content_hash) + .map(|user_value| user_value.map(|s| s.into())) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError)) + } + + pub fn get_attachment(&self, content_hash: &str) -> BichonResult> { + self.attachments_keyspace + .get(content_hash) + .map(|user_value| user_value.map(|s| s.into())) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError)) + } + + pub fn delete( + &self, + email_content_hashes: &[String], + attachment_content_hashes: &[String], + ) -> BichonResult<()> { + let mut batch = self.db.batch(); + for hash in email_content_hashes { + batch.remove(&self.email_keyspace, hash); + } + for hash in attachment_content_hashes { + batch.remove(&self.attachments_keyspace, hash); + } + batch + .commit() + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + Ok(()) + } +} + +pub async fn get_reader(account_id: u64, eid: String) -> BichonResult> { + let (_, data) = reattach_eml_content(account_id, eid).await?; + Ok(Cursor::new(data)) +} diff --git a/src/modules/cache/imap/sync/flow.rs b/src/modules/cache/imap/sync/flow.rs index e004933..5e71338 100644 --- a/src/modules/cache/imap/sync/flow.rs +++ b/src/modules/cache/imap/sync/flow.rs @@ -32,7 +32,7 @@ use crate::{ }, error::{code::ErrorCode, BichonError, BichonResult}, imap::executor::ImapExecutor, - indexer::manager::ENVELOPE_INDEX_MANAGER, + blob::manager::ENVELOPE_INDEX_MANAGER, }, raise_error, }; diff --git a/src/modules/cache/imap/sync/rebuild.rs b/src/modules/cache/imap/sync/rebuild.rs index d309adc..d1c5e97 100644 --- a/src/modules/cache/imap/sync/rebuild.rs +++ b/src/modules/cache/imap/sync/rebuild.rs @@ -19,6 +19,7 @@ use crate::{ modules::{ account::migration::AccountModel, + blob::{manager::ENVELOPE_INDEX_MANAGER, storage::BLOB_MANAGER}, cache::{ imap::{ mailbox::MailBox, @@ -27,10 +28,6 @@ use crate::{ SEMAPHORE, }, error::{code::ErrorCode, BichonError, BichonResult}, - indexer::{ - attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, - manager::ENVELOPE_INDEX_MANAGER, - }, }, raise_error, }; @@ -200,10 +197,9 @@ pub async fn rebuild_mailbox_cache( let content_hashes = ENVELOPE_INDEX_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox.id]) .await?; - + // todo Distinguish these hashes to avoid mixing them if !content_hashes.is_empty() { - EML_INDEX_MANAGER.delete(&content_hashes).await?; - ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + BLOB_MANAGER.delete(&content_hashes, &content_hashes)?; } if remote_mailbox.exists == 0 { @@ -234,10 +230,9 @@ pub async fn rebuild_mailbox_cache_by_date( let content_hashes = ENVELOPE_INDEX_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox_id]) .await?; - + if !content_hashes.is_empty() { - EML_INDEX_MANAGER.delete(&content_hashes).await?; - ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + BLOB_MANAGER.delete(&content_hashes, &content_hashes)?; } if remote.exists == 0 { diff --git a/src/modules/cache/imap/task.rs b/src/modules/cache/imap/task.rs index 8c37348..cb8da2b 100644 --- a/src/modules/cache/imap/task.rs +++ b/src/modules/cache/imap/task.rs @@ -16,7 +16,6 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - use crate::modules::account::entity::AuthType; use crate::modules::cache::imap::sync::execute_imap_sync; use crate::modules::common::periodic::{PeriodicTask, TaskHandle}; @@ -26,10 +25,11 @@ use crate::modules::{ error::BichonResult, }; use crate::utc_now; -use dashmap::DashMap; +use std::collections::HashMap; use std::sync::atomic::{AtomicI64, Ordering}; use std::{sync::LazyLock, time::Duration}; -use tracing::{error, warn}; +use tokio::sync::Mutex; +use tracing::{error, info, warn}; static _DESCRIPTION: &str = "This task periodically synchronizes mailbox data for a specified account, ensuring that all local data is up-to-date."; const TASK_INTERVAL: Duration = Duration::from_secs(10); @@ -38,13 +38,13 @@ static LAST_WARN_TIME: AtomicI64 = AtomicI64::new(0); const WARN_INTERVAL_MS: i64 = 600_000; pub struct AccountSyncTask { - tasks: DashMap, + tasks: Mutex>>, } impl AccountSyncTask { pub fn new() -> Self { Self { - tasks: DashMap::new(), + tasks: Mutex::new(Some(HashMap::new())), } } @@ -103,15 +103,48 @@ impl AccountSyncTask { }) }; let handler = periodic_task.start(task, Some(account_id), TASK_INTERVAL, true, true); - self.tasks.insert(account_id, handler); + self.add_task(account_id, handler).await; + } + + pub async fn add_task(&self, account_id: u64, handler: TaskHandle) { + let mut guard = self.tasks.lock().await; + if let Some(map) = guard.as_mut() { + map.insert(account_id, handler); + } else { + tracing::error!("Failed to add task: HashMap has been taken during shutdown."); + } } pub async fn stop(&self, account_id: u64) -> BichonResult<()> { - if let Some((_, handler)) = self.tasks.remove(&account_id) { - handler.cancel().await; + let mut guard = self.tasks.lock().await; + if let Some(map) = guard.as_mut() { + if let Some(handler) = map.remove(&account_id) { + drop(guard); + handler.cancel().await; + } else { + warn!("No sync task found for account: {}", account_id); + } } else { - warn!("No sync task found for account: {}", account_id); + warn!( + "Stop called after global shutdown for account: {}", + account_id + ); } + Ok(()) } + + pub async fn shutdown(&self) { + let mut guard = self.tasks.lock().await; + if let Some(map) = guard.take() { + drop(guard); + for (account_id, handler) in map { + info!("Shutdown: Waiting for account {} to sync...", account_id); + handler.stop().await; + } + info!("Shutdown: All sync tasks stopped successfully."); + } else { + warn!("Shutdown: Sync tasks were already shut down or never initialized."); + } + } } diff --git a/src/modules/common/periodic.rs b/src/modules/common/periodic.rs index a95a1b4..0009922 100644 --- a/src/modules/common/periodic.rs +++ b/src/modules/common/periodic.rs @@ -16,10 +16,9 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - use crate::modules::{common::signal::SIGNAL_MANAGER, error::BichonResult}; use std::{future::Future, time::Duration}; -use tokio::{sync::oneshot, time::MissedTickBehavior}; +use tokio::{sync::oneshot, task::JoinHandle, time::MissedTickBehavior}; use tracing::{info, warn}; pub struct PeriodicTask { @@ -28,7 +27,7 @@ pub struct PeriodicTask { pub struct TaskHandle { cancel_sender: Option>, - join_handle: tokio::task::JoinHandle<()>, + join_handle: JoinHandle<()>, } impl TaskHandle { @@ -38,6 +37,10 @@ impl TaskHandle { } let _ = self.join_handle.await; } + + pub async fn stop(self) { + let _ = self.join_handle.await; + } } impl PeriodicTask { @@ -82,6 +85,14 @@ impl PeriodicTask { let mut cancel_receiver = cancel_receiver_opt; loop { + let cancel_fut = async { + if let Some(ref mut rx) = cancel_receiver { + rx.await.ok(); + } else { + std::future::pending::<()>().await; + } + }; + tokio::select! { _ = interval.tick() => { match task(param).await { @@ -92,13 +103,7 @@ impl PeriodicTask { } } // only enabled if cancel_receiver is Some - _ = async { - if let Some(ref mut rx) = cancel_receiver { - rx.await.ok() - } else { - futures::future::pending().await - } - } => { + _ = cancel_fut => { info!("Task '{}' received cancellation signal", name_clone); break; } diff --git a/src/modules/common/rustls.rs b/src/modules/common/rustls.rs index 44c1653..7fd75a2 100644 --- a/src/modules/common/rustls.rs +++ b/src/modules/common/rustls.rs @@ -16,7 +16,6 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - use crate::{ modules::{ context::Initialize, @@ -25,9 +24,9 @@ use crate::{ raise_error, }; -pub struct RustMailerTls; +pub struct BichonTls; -impl Initialize for RustMailerTls { +impl Initialize for BichonTls { async fn initialize() -> BichonResult<()> { rustls::crypto::CryptoProvider::install_default(rustls::crypto::ring::default_provider()) .map_err(|_| { @@ -38,4 +37,3 @@ impl Initialize for RustMailerTls { }) } } - \ No newline at end of file diff --git a/src/modules/context/mod.rs b/src/modules/context/mod.rs index e65cf36..7a81019 100644 --- a/src/modules/context/mod.rs +++ b/src/modules/context/mod.rs @@ -16,7 +16,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use crate::modules::error::BichonResult; +use crate::modules::{common::periodic::TaskHandle, error::BichonResult}; pub mod controller; pub mod executors; @@ -27,6 +27,6 @@ pub trait Initialize { async fn initialize() -> BichonResult<()>; } -pub trait RustMailTask { - fn start(); +pub trait BichonTask { + fn start() -> TaskHandle; } diff --git a/src/modules/dashboard/mod.rs b/src/modules/dashboard/mod.rs index b7307d7..64e7a93 100644 --- a/src/modules/dashboard/mod.rs +++ b/src/modules/dashboard/mod.rs @@ -25,9 +25,9 @@ use crate::{ bichon_version, modules::{ account::migration::AccountModel, + blob::manager::ENVELOPE_INDEX_MANAGER, common::auth::ClientContext, error::{code::ErrorCode, BichonResult}, - indexer::manager::ENVELOPE_INDEX_MANAGER, settings::dir::DATA_DIR_MANAGER, utils::get_total_size, }, @@ -81,9 +81,7 @@ impl DashboardStats { if has_all_accounts { stat.storage_usage_bytes = get_total_size(&DATA_DIR_MANAGER.eml_dir) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))? - + get_total_size(&DATA_DIR_MANAGER.attachment_dir) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; stat.index_usage_bytes = get_total_size(&DATA_DIR_MANAGER.envelope_dir) .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; diff --git a/src/modules/duckdb/build.rs b/src/modules/duckdb/build.rs index eb97d78..a31a289 100644 --- a/src/modules/duckdb/build.rs +++ b/src/modules/duckdb/build.rs @@ -21,7 +21,7 @@ use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; use std::sync::Arc; -use crate::modules::indexer::envelope::Envelope; +use crate::modules::blob::envelope::Envelope; pub const DEFAULT_SHARD_ID: u64 = 0; diff --git a/src/modules/duckdb/init.rs b/src/modules/duckdb/init.rs index 12b0383..479bd1f 100644 --- a/src/modules/duckdb/init.rs +++ b/src/modules/duckdb/init.rs @@ -28,14 +28,11 @@ use std::{ use crate::{ modules::{ account::migration::AccountModel, + blob::{envelope::Envelope, manager::ENVELOPE_INDEX_MANAGER, storage::BLOB_MANAGER}, context::Initialize, dashboard::{DashboardStats, Group, LargestEmail, TimeBucket}, duckdb::{build::build_record_batch, refinery::DuckDBConnection}, error::{code::ErrorCode, BichonResult}, - indexer::{ - attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, envelope::Envelope, - manager::ENVELOPE_INDEX_MANAGER, - }, message::{ attachment::AttachmentMetadata, content::{AttachmentDetail, AttachmentInfo}, @@ -86,7 +83,7 @@ pub struct DuckDBManager { impl Initialize for DuckDBManager { async fn initialize() -> BichonResult<()> { - tracing::debug!("Initializing databases"); + tracing::debug!("Initializing duckdb"); if !&DATA_DIR_MANAGER.envelope_dir.exists() { std::fs::create_dir_all(&DATA_DIR_MANAGER.envelope_dir) @@ -1187,19 +1184,11 @@ impl DuckDBManager { } }; - if let Err(e) = EML_INDEX_MANAGER.delete(&content_hashes).await { + if let Err(e) = BLOB_MANAGER.delete(&content_hashes, &content_hashes) { tracing::error!( account_id = account_id, error = %e, - "failed to cleanup eml index" - ); - } - - if let Err(e) = ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await { - tracing::error!( - account_id = account_id, - error = %e, - "failed to cleanup attachment index" + "failed to cleanup eml" ); } }); diff --git a/src/modules/envelope/extractor.rs b/src/modules/envelope/extractor.rs index ef9a2a5..64aaf10 100644 --- a/src/modules/envelope/extractor.rs +++ b/src/modules/envelope/extractor.rs @@ -16,27 +16,25 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use crate::modules::blob::manager::ENVELOPE_INDEX_MANAGER; +use crate::modules::blob::storage::{DetachedEmail, BLOB_MANAGER}; use crate::modules::common::AddrVec; use crate::modules::envelope::utils::normalize_subject; use crate::modules::error::code::ErrorCode; use crate::modules::error::BichonResult; -use crate::modules::indexer::attachment::ATTACHMENT_INDEX_MANAGER; -use crate::modules::indexer::eml::EML_INDEX_MANAGER; -use crate::modules::indexer::manager::ENVELOPE_INDEX_MANAGER; -use crate::modules::indexer::schema::SchemaTools; use crate::modules::message::content::AttachmentInfo; use crate::modules::utils::html::extract_text; use crate::modules::utils::{compute_content_hash, hex_hash}; -use crate::{id, modules::indexer::envelope::Envelope}; +use crate::{id, modules::blob::envelope::Envelope}; use crate::{raise_error, utc_now}; use async_imap::types::Fetch; +use bytes::Bytes; use mail_parser::{Address, HeaderName, Message, MessageParser, MimeHeaders}; -use tantivy::doc; use tracing::error; use uuid::Uuid; pub async fn extract_envelope_and_store_it( - fetch: &Fetch, + fetch: Fetch, account_id: u64, mailbox_id: u64, ) -> BichonResult<()> { @@ -301,22 +299,14 @@ pub async fn detach_and_store_attachments( .collect(); ranges.sort_by(|a, b| b.0.cmp(&a.0)); - - let fields = SchemaTools::fields(); + let mut attachments = Vec::with_capacity(ranges.len()); for (raw_start, raw_end, att) in ranges { // Step 2: Extract raw bytes and store them as standalone documents let raw_bytes = &original_body[raw_start..raw_end]; let content_hash = compute_content_hash(raw_bytes); - ATTACHMENT_INDEX_MANAGER - .add_document( - content_hash.clone(), - doc!( - fields.f_id => content_hash.clone(), - fields.f_blob => raw_bytes - ), - ) - .await; + attachments.push((content_hash.clone(), Bytes::copy_from_slice(raw_bytes))); + // Step 3: Replace raw attachment content with a hash-based placeholder let placeholder = format!("<>", &content_hash); let p_bytes = placeholder.as_bytes(); @@ -347,14 +337,11 @@ pub async fn detach_and_store_attachments( attachment_infos.push(info); } // Step 4: Store the final stripped EML content - EML_INDEX_MANAGER - .add_document( - eml_content_hash.to_string(), - doc!( - fields.f_id => eml_content_hash.to_string(), - fields.f_blob => stripped_eml - ), - ) + BLOB_MANAGER + .queue(DetachedEmail { + email: (eml_content_hash.to_string(), Bytes::from(stripped_eml)), + attachments: Some(attachments), + }) .await; attachment_infos @@ -363,7 +350,7 @@ pub async fn detach_and_store_attachments( pub async fn reattach_eml_content( account_id: u64, envelope_id: String, -) -> BichonResult<(Envelope, Vec)> { +) -> BichonResult<(Envelope, Bytes)> { let envelope = ENVELOPE_INDEX_MANAGER .get_envelope_by_id(account_id, envelope_id.clone()) .await? @@ -377,15 +364,14 @@ pub async fn reattach_eml_content( ) })?; - let mut restored_eml = EML_INDEX_MANAGER - .get(&envelope.content_hash) - .await? + let restored_eml = BLOB_MANAGER + .get_email(&envelope.content_hash)? .ok_or_else(|| { raise_error!( format!( - "Original email content not found: account_id={} envelope_id={} content_hash={}", - account_id, &envelope_id, &envelope.content_hash - ), + "Original email content not found: account_id={} envelope_id={} content_hash={}", + account_id, &envelope_id, &envelope.content_hash + ), ErrorCode::ResourceNotFound ) })?; @@ -394,6 +380,8 @@ pub async fn reattach_eml_content( return Ok((envelope, restored_eml)); } + let mut restored_eml = restored_eml.to_vec(); + let account_detail = ENVELOPE_INDEX_MANAGER .get_attachments_by_envelope_id(account_id, envelope_id) .await?; @@ -432,7 +420,7 @@ pub async fn reattach_eml_content( tasks.sort_by(|a, b| b.0.cmp(&a.0)); for (start, end, hash) in tasks { - if let Some(original_data) = ATTACHMENT_INDEX_MANAGER.get(&hash).await? { + if let Some(original_data) = BLOB_MANAGER.get_attachment(&hash)? { let actual_hash = compute_content_hash(&original_data); if actual_hash != hash { error!( @@ -447,7 +435,7 @@ pub async fn reattach_eml_content( } } - Ok((envelope, restored_eml)) + Ok((envelope, Bytes::from(restored_eml))) } #[cfg(test)] diff --git a/src/modules/imap/executor.rs b/src/modules/imap/executor.rs index 8ada3e6..9fd9334 100644 --- a/src/modules/imap/executor.rs +++ b/src/modules/imap/executor.rs @@ -202,7 +202,7 @@ impl ImapExecutor { .await .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? { - extract_envelope_and_store_it(&fetch, account_id, mailbox_id).await?; + extract_envelope_and_store_it(fetch, account_id, mailbox_id).await?; count += 1; } Ok(count) @@ -229,7 +229,7 @@ impl ImapExecutor { .await .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? { - extract_envelope_and_store_it(&fetch, account_id, mailbox_id).await?; + extract_envelope_and_store_it(fetch, account_id, mailbox_id).await?; } Ok(()) } diff --git a/src/modules/import/mod.rs b/src/modules/import/mod.rs index 19b0d9c..99b5ff9 100644 --- a/src/modules/import/mod.rs +++ b/src/modules/import/mod.rs @@ -19,7 +19,6 @@ use poem_openapi::Object; use serde::{Deserialize, Serialize}; -use tantivy::doc; use crate::{ base64_decode_url_safe, diff --git a/src/modules/indexer/attachment.rs b/src/modules/indexer/attachment.rs deleted file mode 100644 index 79bd36d..0000000 --- a/src/modules/indexer/attachment.rs +++ /dev/null @@ -1,299 +0,0 @@ -// -// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) -// -// This file is part of the Bichon Email Archiving Project -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU Affero General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Affero General Public License for more details. -// -// You should have received a copy of the GNU Affero General Public License -// along with this program. If not, see . - -use std::{ - collections::HashMap, - path::PathBuf, - sync::{Arc, LazyLock}, - time::Duration, -}; - -use crate::modules::{indexer::DocumentOp, settings::cli::SETTINGS}; -use crate::{ - modules::{ - common::signal::SIGNAL_MANAGER, - error::{code::ErrorCode, BichonResult}, - indexer::schema::SchemaTools, - settings::dir::DATA_DIR_MANAGER, - }, - raise_error, -}; - -use tantivy::indexer::{NoMergePolicy, UserOperation}; -use tantivy::{ - collector::TopDocs, - query::TermQuery, - schema::{IndexRecordOption, Value}, - store::Compressor, - Index, IndexBuilder, IndexReader, IndexSettings, IndexWriter, TantivyDocument, Term, -}; -use tokio::{ - sync::{mpsc, Mutex}, - task, -}; -use tracing::info; - -pub const ATTACHMENT_BATCH_SIZE: usize = 10; -const MAX_BUFFER_DURATION: Duration = Duration::from_secs(10); - -pub static ATTACHMENT_INDEX_MANAGER: LazyLock = - LazyLock::new(AttachmentManager::new); - -pub struct AttachmentManager { - index_writer: Arc>, - sender: mpsc::Sender, - reader: IndexReader, -} - -impl AttachmentManager { - pub fn new() -> Self { - let index = Self::open_or_create_index(&DATA_DIR_MANAGER.attachment_dir); - - let writer: IndexWriter = index - .writer_with_num_threads( - SETTINGS.bichon_tantivy_threads as usize, - SETTINGS.bichon_tantivy_buffer_size, - ) - .unwrap_or_else(|e| { - panic!( - "Failed to create IndexWriter (threads: {}, buffer: {}B) for {:?}: {}", - SETTINGS.bichon_tantivy_threads, - SETTINGS.bichon_tantivy_buffer_size, - DATA_DIR_MANAGER.attachment_dir, - e - ) - }); - - writer.set_merge_policy(Box::new(NoMergePolicy)); - let index_writer = Arc::new(Mutex::new(writer)); - - let reader = index.reader().unwrap_or_else(|e| { - panic!( - "Failed to create IndexReader for {:?}: {}", - DATA_DIR_MANAGER.eml_dir, e - ) - }); - let (sender, mut receiver) = mpsc::channel::(100); - task::spawn(async move { - let mut buffer: HashMap = - HashMap::with_capacity(ATTACHMENT_BATCH_SIZE); - let mut interval = tokio::time::interval(MAX_BUFFER_DURATION); - let mut shutdown = SIGNAL_MANAGER.subscribe(); - loop { - tokio::select! { - maybe_msg = receiver.recv() => { - match maybe_msg { - Some(DocumentOp::Document((eid, doc))) => { - buffer.insert(eid, doc); - if buffer.len() >= ATTACHMENT_BATCH_SIZE { - ATTACHMENT_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - } - } - Some(DocumentOp::Shutdown) => { - ATTACHMENT_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - break; - } - None => break, - } - } - _ = interval.tick() => { - if !buffer.is_empty() { - ATTACHMENT_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - } - } - _ = shutdown.recv() => { - let _ = ATTACHMENT_INDEX_MANAGER.sender.send(DocumentOp::Shutdown).await; - } - } - } - }); - Self { - index_writer, - sender, - reader, - } - } - - pub async fn add_document(&self, content_hash: String, doc: TantivyDocument) { - let _ = self - .sender - .send(DocumentOp::Document((content_hash, doc))) - .await; - } - - fn open_or_create_index(index_dir: &PathBuf) -> Index { - let need_create = !index_dir.exists() - || index_dir - .read_dir() - .map(|mut d| d.next().is_none()) - .unwrap_or(true); - - if need_create { - info!( - "Attachment storage not found or empty, creating new attachment storage at {}", - index_dir.display() - ); - std::fs::create_dir_all(&index_dir).unwrap_or_else(|e| { - panic!("Failed to create index directory {:?}: {}", index_dir, e) - }); - IndexBuilder::new() - .schema(SchemaTools::schema()) - .settings(IndexSettings { - docstore_compression: Compressor::None, - docstore_compress_dedicated_thread: Default::default(), - docstore_blocksize: Default::default(), - }) - .create_in_dir(&index_dir) - .unwrap_or_else(|e| panic!("Failed to create index in {:?}: {}", index_dir, e)) - } else { - info!( - "Opening existing attachment data storage at {}", - index_dir.display() - ); - open(&index_dir) - } - } - - fn term(&self, content_hash: &str) -> Term { - Term::from_field_text(SchemaTools::fields().f_id, content_hash) - } - - pub async fn get(&self, content_hash: &str) -> BichonResult>> { - let searcher = self.reader.searcher(); - let term = Term::from_field_text(SchemaTools::fields().f_id, content_hash); - let query = TermQuery::new(term, IndexRecordOption::Basic); - let docs = searcher - .search(&query, &TopDocs::with_limit(1)) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - - if docs.is_empty() { - return Ok(None); - } - - let (_, doc_address) = docs.first().unwrap(); - let doc: TantivyDocument = searcher - .doc_async(*doc_address) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - let fields = SchemaTools::fields(); - let value = doc.get_first(fields.f_blob).ok_or_else(|| { - raise_error!( - format!("miss '{}' field in tantivy document", stringify!(field)), - ErrorCode::InternalError - ) - })?; - let bytes = value.as_bytes().ok_or_else(|| { - raise_error!( - format!("'{}' field is not a bytes", stringify!(field)), - ErrorCode::InternalError - ) - })?; - - Ok(Some(bytes.to_vec())) - } - - pub async fn delete( - &self, - content_hashes: &Vec, // HashMap - ) -> BichonResult<()> { - if content_hashes.is_empty() { - tracing::warn!("deletes is empty, nothing to delete"); - return Ok(()); - } - - let mut writer = self.index_writer.lock().await; - for hash in content_hashes { - let term = self.term(hash); - writer.delete_term(term); - } - - writer - .commit() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - - Ok(()) - } - - async fn drain_and_commit(&self, buffer: &mut HashMap) { - if buffer.is_empty() { - return; - } - let mut writer = self.index_writer.lock().await; - let mut operations = Vec::new(); - - for (content_hash, doc) in buffer.drain() { - let delete_term = Term::from_field_text(SchemaTools::fields().f_id, &content_hash); - operations.push(UserOperation::Delete(delete_term)); - operations.push(UserOperation::Add(doc)); - } - if let Err(e) = writer.run(operations) { - eprintln!("[FATAL] Tantivy run failed: {e:?}"); - std::process::exit(1); - } - - fatal_commit(&mut writer); - } -} - -fn fatal_commit(writer: &mut IndexWriter) { - const MAX_RETRIES: usize = 3; - const RETRY_DELAY_MS: u64 = 1000; - - for attempt in 0..=MAX_RETRIES { - match writer.commit() { - Ok(_) => { - if attempt > 0 { - eprintln!("[INFO] Commit succeeded on attempt {}", attempt + 1); - } - return; - } - Err(e) => match &e { - tantivy::TantivyError::IoError(io_error) => { - if attempt < MAX_RETRIES { - eprintln!( - "[WARN] Commit failed (attempt {}/{}): {:?}. Retrying in {}ms...", - attempt + 1, - MAX_RETRIES + 1, - io_error, - RETRY_DELAY_MS * (attempt as u64 + 1) - ); - std::thread::sleep(std::time::Duration::from_millis( - RETRY_DELAY_MS * (attempt as u64 + 1), - )); - } else { - eprintln!( - "[FATAL] Tantivy commit failed after {} attempts: {:?}", - MAX_RETRIES + 1, - io_error - ); - std::process::exit(1); - } - } - _ => { - eprintln!("[FATAL] Tantivy commit failed with non-IO error: {e:?}"); - std::process::exit(1); - } - }, - } - } -} - -fn open(index_dir: &PathBuf) -> Index { - Index::open_in_dir(index_dir) - .unwrap_or_else(|e| panic!("Failed to open index in {:?}: {}", index_dir, e)) -} diff --git a/src/modules/indexer/eml.rs b/src/modules/indexer/eml.rs deleted file mode 100644 index cf0ce22..0000000 --- a/src/modules/indexer/eml.rs +++ /dev/null @@ -1,331 +0,0 @@ -// -// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) -// -// This file is part of the Bichon Email Archiving Project -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU Affero General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Affero General Public License for more details. -// -// You should have received a copy of the GNU Affero General Public License -// along with this program. If not, see . - -use std::{ - collections::HashMap, - path::PathBuf, - sync::{Arc, LazyLock}, - time::Duration, -}; - -use crate::modules::{ - envelope::extractor::reattach_eml_content, indexer::DocumentOp, settings::cli::SETTINGS, -}; -use crate::{ - modules::{ - common::signal::SIGNAL_MANAGER, - error::{code::ErrorCode, BichonResult}, - indexer::schema::SchemaTools, - settings::dir::DATA_DIR_MANAGER, - }, - raise_error, -}; -use tantivy::indexer::{NoMergePolicy, UserOperation}; -use tantivy::{ - collector::TopDocs, - query::TermQuery, - schema::{IndexRecordOption, Value}, - store::{Compressor, ZstdCompressor}, - Index, IndexBuilder, IndexReader, IndexSettings, IndexWriter, TantivyDocument, Term, -}; -use tokio::{ - fs::File, - io::AsyncWriteExt, - sync::{mpsc, Mutex}, - task, -}; -use tracing::info; - -pub const EML_BATCH_SIZE: usize = 100; -const MAX_BUFFER_DURATION: Duration = Duration::from_secs(10); - -pub static EML_INDEX_MANAGER: LazyLock = LazyLock::new(EmlIndexManager::new); - -pub struct EmlIndexManager { - index_writer: Arc>, - sender: mpsc::Sender, - reader: IndexReader, -} - -impl EmlIndexManager { - pub fn new() -> Self { - let index = Self::open_or_create_index(&DATA_DIR_MANAGER.eml_dir); - - let writer: IndexWriter = index - .writer_with_num_threads( - SETTINGS.bichon_tantivy_threads as usize, - SETTINGS.bichon_tantivy_buffer_size, - ) - .unwrap_or_else(|e| { - panic!( - "Failed to create IndexWriter (threads: {}, buffer: {}B) for {:?}: {}", - SETTINGS.bichon_tantivy_threads, - SETTINGS.bichon_tantivy_buffer_size, - DATA_DIR_MANAGER.eml_dir, - e - ) - }); - - writer.set_merge_policy(Box::new(NoMergePolicy)); - let index_writer = Arc::new(Mutex::new(writer)); - - let reader = index.reader().unwrap_or_else(|e| { - panic!( - "Failed to create IndexReader for {:?}: {}", - DATA_DIR_MANAGER.eml_dir, e - ) - }); - let (sender, mut receiver) = mpsc::channel::(100); - task::spawn(async move { - let mut buffer: HashMap = - HashMap::with_capacity(EML_BATCH_SIZE); - let mut interval = tokio::time::interval(MAX_BUFFER_DURATION); - let mut shutdown = SIGNAL_MANAGER.subscribe(); - loop { - tokio::select! { - maybe_msg = receiver.recv() => { - match maybe_msg { - Some(DocumentOp::Document((eid, doc))) => { - buffer.insert(eid, doc); - if buffer.len() >= EML_BATCH_SIZE { - EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - } - } - Some(DocumentOp::Shutdown) => { - EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - break; - } - None => break, - } - } - _ = interval.tick() => { - if !buffer.is_empty() { - EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - } - } - _ = shutdown.recv() => { - let _ = EML_INDEX_MANAGER.sender.send(DocumentOp::Shutdown).await; - } - } - } - }); - Self { - index_writer, - sender, - reader, - } - } - - /// Adds a document to the indexer. - /// - /// # Parameters - /// - `eid`: A hash derived from **Account ID + Message ID**. - /// This acts as a unique identifier for the EML content itself. - /// - /// - `doc`: The `TantivyDocument` representing the mail body/content. - /// - /// # Logical Design - /// Unlike the `envelope_id` (which is a hash of Account + Folder + Message ID), - /// this `eid` ignores the folder context. This ensures that while metadata - /// (envelopes) can be duplicated across different folders, the physical - /// EML/document storage remains de-duplicated and unique. - pub async fn add_document(&self, content_hash: String, doc: TantivyDocument) { - let _ = self - .sender - .send(DocumentOp::Document((content_hash, doc))) - .await; - } - - fn open_or_create_index(index_dir: &PathBuf) -> Index { - let need_create = !index_dir.exists() - || index_dir - .read_dir() - .map(|mut d| d.next().is_none()) - .unwrap_or(true); - - if need_create { - info!( - "Email storage not found or empty, creating new mail storage at {}", - index_dir.display() - ); - std::fs::create_dir_all(&index_dir).unwrap_or_else(|e| { - panic!("Failed to create index directory {:?}: {}", index_dir, e) - }); - IndexBuilder::new() - .schema(SchemaTools::schema()) - .settings(IndexSettings { - docstore_compression: Compressor::Zstd(ZstdCompressor { - compression_level: Some(SETTINGS.bichon_eml_compression_level as i32), - }), - docstore_compress_dedicated_thread: true, - docstore_blocksize: SETTINGS.bichon_eml_blocksize, - }) - .create_in_dir(&index_dir) - .unwrap_or_else(|e| panic!("Failed to create index in {:?}: {}", index_dir, e)) - } else { - info!("Opening existing email storage at {}", index_dir.display()); - open(&index_dir) - } - } - - fn term(&self, content_hash: &str) -> Term { - Term::from_field_text(SchemaTools::fields().f_id, content_hash) - } - - pub async fn get(&self, content_hash: &str) -> BichonResult>> { - let searcher = self.reader.searcher(); - let term = Term::from_field_text(SchemaTools::fields().f_id, content_hash); - let query = TermQuery::new(term, IndexRecordOption::Basic); - let docs = searcher - .search(&query, &TopDocs::with_limit(1)) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - - if docs.is_empty() { - return Ok(None); - } - - let (_, doc_address) = docs.first().unwrap(); - let doc: TantivyDocument = searcher - .doc_async(*doc_address) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - let fields = SchemaTools::fields(); - let value = doc.get_first(fields.f_blob).ok_or_else(|| { - raise_error!( - format!("miss '{}' field in tantivy document", stringify!(field)), - ErrorCode::InternalError - ) - })?; - let bytes = value.as_bytes().ok_or_else(|| { - raise_error!( - format!("'{}' field is not a bytes", stringify!(field)), - ErrorCode::InternalError - ) - })?; - - Ok(Some(bytes.to_vec())) - } - - pub async fn get_reader(&self, account_id: u64, eid: String) -> BichonResult { - let (envelope, data) = reattach_eml_content(account_id, eid).await?; - let mut path = DATA_DIR_MANAGER.temp_dir.clone(); - path.push(format!("{}.eml", envelope.content_hash)); - { - let mut file = File::create(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - file.write_all(&data) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - let file = File::open(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(file) - } - - pub async fn delete( - &self, - content_hashes: &Vec, // HashMap - ) -> BichonResult<()> { - if content_hashes.is_empty() { - tracing::warn!("delete_email_multi_account: deletes is empty, nothing to delete"); - return Ok(()); - } - - let mut writer = self.index_writer.lock().await; - for hash in content_hashes { - let term = self.term(hash); - writer.delete_term(term); - } - writer - .commit() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - - Ok(()) - } - - // Deduplicate directly by content_hash, regardless of the account. - async fn drain_and_commit(&self, buffer: &mut HashMap) { - if buffer.is_empty() { - return; - } - let mut writer = self.index_writer.lock().await; - let mut operations = Vec::new(); - - for (content_hash, doc) in buffer.drain() { - let delete_term = Term::from_field_text(SchemaTools::fields().f_id, &content_hash); - operations.push(UserOperation::Delete(delete_term)); - operations.push(UserOperation::Add(doc)); - } - if let Err(e) = writer.run(operations) { - eprintln!("[FATAL] Tantivy run failed: {e:?}"); - std::process::exit(1); - } - - fatal_commit(&mut writer); - } -} - -fn fatal_commit(writer: &mut IndexWriter) { - const MAX_RETRIES: usize = 3; - const RETRY_DELAY_MS: u64 = 1000; - - for attempt in 0..=MAX_RETRIES { - match writer.commit() { - Ok(_) => { - if attempt > 0 { - eprintln!("[INFO] Commit succeeded on attempt {}", attempt + 1); - } - return; - } - Err(e) => match &e { - tantivy::TantivyError::IoError(io_error) => { - if attempt < MAX_RETRIES { - eprintln!( - "[WARN] Commit failed (attempt {}/{}): {:?}. Retrying in {}ms...", - attempt + 1, - MAX_RETRIES + 1, - io_error, - RETRY_DELAY_MS * (attempt as u64 + 1) - ); - std::thread::sleep(std::time::Duration::from_millis( - RETRY_DELAY_MS * (attempt as u64 + 1), - )); - } else { - eprintln!( - "[FATAL] Tantivy commit failed after {} attempts: {:?}", - MAX_RETRIES + 1, - io_error - ); - std::process::exit(1); - } - } - _ => { - eprintln!("[FATAL] Tantivy commit failed with non-IO error: {e:?}"); - std::process::exit(1); - } - }, - } - } -} - -fn open(index_dir: &PathBuf) -> Index { - Index::open_in_dir(index_dir) - .unwrap_or_else(|e| panic!("Failed to open index in {:?}: {}", index_dir, e)) -} diff --git a/src/modules/indexer/fields.rs b/src/modules/indexer/fields.rs deleted file mode 100644 index 9a102b0..0000000 --- a/src/modules/indexer/fields.rs +++ /dev/null @@ -1,27 +0,0 @@ -// -// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) -// -// This file is part of the Bichon Email Archiving Project -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU Affero General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Affero General Public License for more details. -// -// You should have received a copy of the GNU Affero General Public License -// along with this program. If not, see . - -use tantivy::schema::Field; - -pub const F_ID: &str = "id"; -pub const F_BLOB: &str = "blob"; - -pub struct BlobFields { - pub f_id: Field, - pub f_blob: Field, -} diff --git a/src/modules/indexer/schema.rs b/src/modules/indexer/schema.rs deleted file mode 100644 index 139e302..0000000 --- a/src/modules/indexer/schema.rs +++ /dev/null @@ -1,49 +0,0 @@ -// -// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) -// -// This file is part of the Bichon Email Archiving Project -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU Affero General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Affero General Public License for more details. -// -// You should have received a copy of the GNU Affero General Public License -// along with this program. If not, see . - -use std::sync::{Arc, LazyLock}; - -use crate::modules::indexer::fields::*; -use tantivy::schema::STRING; -use tantivy::schema::{Schema, FAST, STORED}; - -static BLOB_FIELDS: LazyLock> = LazyLock::new(|| { - let (_, fields) = SchemaTools::create_schema(); - Arc::new(fields) -}); - -pub struct SchemaTools; - -impl SchemaTools { - pub fn schema() -> Schema { - let (schema, _) = Self::create_schema(); - schema - } - - pub fn fields() -> &'static BlobFields { - &BLOB_FIELDS - } - - pub fn create_schema() -> (Schema, BlobFields) { - let mut builder = Schema::builder(); - let f_id = builder.add_text_field(F_ID, STRING | FAST); - let f_blob = builder.add_bytes_field(F_BLOB, STORED); - let fields = BlobFields { f_id, f_blob }; - (builder.build(), fields) - } -} diff --git a/src/modules/mailbox/delete.rs b/src/modules/mailbox/delete.rs index 822a7ac..5209b36 100644 --- a/src/modules/mailbox/delete.rs +++ b/src/modules/mailbox/delete.rs @@ -1,10 +1,7 @@ use crate::modules::{ + blob::{manager::ENVELOPE_INDEX_MANAGER, storage::BLOB_MANAGER}, cache::imap::mailbox::MailBox, error::BichonResult, - indexer::{ - attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, - manager::ENVELOPE_INDEX_MANAGER, - }, }; pub async fn delete_mailbox_impl(account_id: u64, mailbox_id: u64) -> BichonResult<()> { @@ -33,7 +30,6 @@ pub async fn delete_mailbox_impl(account_id: u64, mailbox_id: u64) -> BichonResu .delete_mailbox_envelopes(account_id, ids_to_delete.clone()) .await?; - EML_INDEX_MANAGER.delete(&content_hashes).await?; - ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + BLOB_MANAGER.delete(&content_hashes, &content_hashes)?; Ok(()) } diff --git a/src/modules/message/attachment.rs b/src/modules/message/attachment.rs index 0b43d25..6493f65 100644 --- a/src/modules/message/attachment.rs +++ b/src/modules/message/attachment.rs @@ -1,19 +1,17 @@ -use std::collections::HashSet; +use std::{collections::HashSet, io::Cursor}; use crate::{ modules::{ envelope::extractor::reattach_eml_content, error::{code::ErrorCode, BichonResult}, - settings::dir::DATA_DIR_MANAGER, utils::compute_content_hash, }, raise_error, }; +use bytes::Bytes; use mail_parser::MessageParser; use poem_openapi::Object; use serde::{Deserialize, Serialize}; -use tokio::fs::File; -use tokio::io::AsyncWriteExt; #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)] pub struct AttachmentMetadata { @@ -34,8 +32,8 @@ pub async fn retrieve_attachment_content( account_id: u64, envelope_id: String, content_hash: &str, -) -> BichonResult { - let (envelope, eml) = reattach_eml_content(account_id, envelope_id).await?; +) -> BichonResult> { + let (_, eml) = reattach_eml_content(account_id, envelope_id).await?; let message = MessageParser::default().parse(&eml).ok_or_else(|| { raise_error!( "Failed to parse parent EML".into(), @@ -53,20 +51,7 @@ pub async fn retrieve_attachment_content( ErrorCode::ResourceNotFound ) })?; - let mut path = DATA_DIR_MANAGER.temp_dir.clone(); - path.push(format!("{}.eml", envelope.content_hash)); - { - let mut file = File::create(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - file.write_all(attachment_content) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - let file = File::open(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(file) + Ok(Cursor::new(Bytes::copy_from_slice(attachment_content))) } pub async fn retrieve_nested_attachment_content( @@ -74,7 +59,7 @@ pub async fn retrieve_nested_attachment_content( envelope_id: String, content_hash: &str, nested_content_hash: &str, -) -> BichonResult { +) -> BichonResult> { let (_, eml) = reattach_eml_content(account_id, envelope_id).await?; let parent_message = MessageParser::default().parse(&eml).ok_or_else(|| { raise_error!( @@ -114,18 +99,5 @@ pub async fn retrieve_nested_attachment_content( ) })?; - let mut path = DATA_DIR_MANAGER.temp_dir.clone(); - path.push(format!("{}.eml", nested_content_hash)); - { - let mut file = File::create(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - file.write_all(attachment_content) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - let file = File::open(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(file) + Ok(Cursor::new(Bytes::copy_from_slice(attachment_content))) } diff --git a/src/modules/message/content.rs b/src/modules/message/content.rs index 3654670..71f4594 100644 --- a/src/modules/message/content.rs +++ b/src/modules/message/content.rs @@ -22,7 +22,7 @@ use crate::modules::envelope::extractor::{ extract_envelope_from_nested_message, reattach_eml_content, }; use crate::modules::error::code::ErrorCode; -use crate::modules::indexer::envelope::Envelope; +use crate::modules::blob::envelope::Envelope; use crate::modules::utils::compute_content_hash; use crate::{modules::error::BichonResult, raise_error}; use mail_parser::{MessageParser, MimeHeaders}; diff --git a/src/modules/message/delete.rs b/src/modules/message/delete.rs index bbd5c3d..a01a67b 100644 --- a/src/modules/message/delete.rs +++ b/src/modules/message/delete.rs @@ -16,10 +16,9 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use crate::modules::blob::manager::ENVELOPE_INDEX_MANAGER; +use crate::modules::blob::storage::BLOB_MANAGER; use crate::modules::error::BichonResult; -use crate::modules::indexer::attachment::ATTACHMENT_INDEX_MANAGER; -use crate::modules::indexer::eml::EML_INDEX_MANAGER; -use crate::modules::indexer::manager::ENVELOPE_INDEX_MANAGER; use std::collections::HashMap; pub async fn delete_messages_impl(request: HashMap>) -> BichonResult<()> { @@ -27,8 +26,7 @@ pub async fn delete_messages_impl(request: HashMap>) -> BichonR .get_orphan_hashes_in_memory(request.clone()) .await?; if !content_hashes.is_empty() { - EML_INDEX_MANAGER.delete(&content_hashes).await?; - ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + BLOB_MANAGER.delete(&content_hashes, &content_hashes)?; } ENVELOPE_INDEX_MANAGER .delete_envelopes_multi_account(request) diff --git a/src/modules/message/list.rs b/src/modules/message/list.rs index 93fbfac..d468854 100644 --- a/src/modules/message/list.rs +++ b/src/modules/message/list.rs @@ -20,7 +20,7 @@ use crate::{ modules::{ account::migration::AccountModel, error::{code::ErrorCode, BichonResult}, - indexer::{envelope::Envelope, manager::ENVELOPE_INDEX_MANAGER}, + blob::{envelope::Envelope, manager::ENVELOPE_INDEX_MANAGER}, rest::response::DataPage, }, raise_error, diff --git a/src/modules/message/search.rs b/src/modules/message/search.rs index 405ee35..fd4cb37 100644 --- a/src/modules/message/search.rs +++ b/src/modules/message/search.rs @@ -25,7 +25,7 @@ use crate::{ modules::{ duckdb::init::duckdb, error::{code::ErrorCode, BichonResult}, - indexer::{envelope::Envelope, manager::ENVELOPE_INDEX_MANAGER}, + blob::{envelope::Envelope, manager::ENVELOPE_INDEX_MANAGER}, rest::response::DataPage, }, raise_error, diff --git a/src/modules/mod.rs b/src/modules/mod.rs index 4229d01..6d249de 100644 --- a/src/modules/mod.rs +++ b/src/modules/mod.rs @@ -29,7 +29,7 @@ pub mod envelope; pub mod error; pub mod imap; pub mod import; -pub mod indexer; +pub mod blob; pub mod logger; pub mod mailbox; pub mod message; diff --git a/src/modules/oauth2/refresh/mod.rs b/src/modules/oauth2/refresh/mod.rs index d70ba15..70fa260 100644 --- a/src/modules/oauth2/refresh/mod.rs +++ b/src/modules/oauth2/refresh/mod.rs @@ -16,9 +16,8 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - -use crate::modules::common::periodic::PeriodicTask; -use crate::modules::context::RustMailTask; +use crate::modules::common::periodic::{PeriodicTask, TaskHandle}; +use crate::modules::context::BichonTask; use crate::modules::oauth2::token::EXTERNAL_OAUTH_APP_ID; use crate::modules::oauth2::{flow::OAuth2Flow, token::OAuth2AccessToken}; use crate::utc_now; @@ -30,8 +29,8 @@ const FIFTEEN_MINUTES: Duration = Duration::from_secs(45 * 60); ///This task cleans up expired OAuth2 pending authorizations that haven't been completed by users in a timely manner. pub struct OAuth2RefreshTask; -impl RustMailTask for OAuth2RefreshTask { - fn start() { +impl BichonTask for OAuth2RefreshTask { + fn start() -> TaskHandle { let periodic_task = PeriodicTask::new("oauth2-token-refresh-task"); let task = move |_: Option| { @@ -86,6 +85,6 @@ impl RustMailTask for OAuth2RefreshTask { }) }; - periodic_task.start(task, None, TASK_INTERVAL, false, true); + periodic_task.start(task, None, TASK_INTERVAL, false, true) } } diff --git a/src/modules/oauth2/task.rs b/src/modules/oauth2/task.rs index 4d87da8..6f81f3a 100644 --- a/src/modules/oauth2/task.rs +++ b/src/modules/oauth2/task.rs @@ -16,9 +16,10 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - use crate::modules::{ - common::periodic::PeriodicTask, context::RustMailTask, oauth2::pending::OAuth2PendingEntity, + common::periodic::{PeriodicTask, TaskHandle}, + context::BichonTask, + oauth2::pending::OAuth2PendingEntity, }; use std::time::Duration; @@ -27,8 +28,8 @@ const TASK_INTERVAL: Duration = Duration::from_secs(6 * 60 * 60); ///This task cleans up expired OAuth2 pending authorizations that haven't been completed by users in a timely manner. pub struct OAuth2CleanTask; -impl RustMailTask for OAuth2CleanTask { - fn start() { +impl BichonTask for OAuth2CleanTask { + fn start() -> TaskHandle { let periodic_task = PeriodicTask::new("oauth2-pending-task-cleaner"); let task = move |_: Option| { @@ -38,6 +39,6 @@ impl RustMailTask for OAuth2CleanTask { }) }; - periodic_task.start(task, None, TASK_INTERVAL, false, false); + periodic_task.start(task, None, TASK_INTERVAL, false, false) } } diff --git a/src/modules/rest/api/message.rs b/src/modules/rest/api/message.rs index 38889c1..1be717c 100644 --- a/src/modules/rest/api/message.rs +++ b/src/modules/rest/api/message.rs @@ -17,10 +17,10 @@ // along with this program. If not, see . use crate::modules::account::migration::AccountModel; +use crate::modules::blob::envelope::Envelope; +use crate::modules::blob::manager::ENVELOPE_INDEX_MANAGER; +use crate::modules::blob::storage::get_reader; use crate::modules::common::auth::ClientContext; -use crate::modules::indexer::eml::EML_INDEX_MANAGER; -use crate::modules::indexer::envelope::Envelope; -use crate::modules::indexer::manager::ENVELOPE_INDEX_MANAGER; use crate::modules::message::append::restore_emails; use crate::modules::message::append::RestoreMessagesRequest; use crate::modules::message::attachment::retrieve_attachment_content; @@ -252,9 +252,7 @@ impl MessageApi { .require_permission(Some(account_id), Permission::DATA_RAW_DOWNLOAD) .await?; let envelope_id = envelope_id.0; - let reader = EML_INDEX_MANAGER - .get_reader(account_id, envelope_id.clone()) - .await?; + let reader = get_reader(account_id, envelope_id.clone()).await?; let body = Body::from_async_read(reader); let attachment = Attachment::new(body) .attachment_type(AttachmentType::Attachment) @@ -391,9 +389,7 @@ impl MessageApi { .await?; } - ENVELOPE_INDEX_MANAGER - .update_envelope_tags(req) - .await?; + ENVELOPE_INDEX_MANAGER.update_envelope_tags(req).await?; Ok(()) } diff --git a/src/modules/settings/dir.rs b/src/modules/settings/dir.rs index 60c765c..7fbff74 100644 --- a/src/modules/settings/dir.rs +++ b/src/modules/settings/dir.rs @@ -28,8 +28,7 @@ use std::sync::LazyLock; pub const META_FILE: &str = "meta.db"; pub const MAILBOX_FILE: &str = "mailbox.db"; const ENVELOPE_DIR: &str = "envelope"; -const EML_DIR: &str = "eml"; -const ATTACHMENT_DIR: &str = "attachment"; +const EML_DIR: &str = "bichon-emls"; const TMP_DIR: &str = "tmp"; const LOG_DIR: &str = "logs"; const TLS_CERT: &str = "cert.pem"; @@ -48,7 +47,6 @@ pub struct DataDirManager { pub tls_key: PathBuf, pub envelope_dir: PathBuf, pub eml_dir: PathBuf, - pub attachment_dir: PathBuf, pub log_dir: PathBuf, } @@ -60,6 +58,8 @@ impl Initialize for DataDirManager { .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; std::fs::create_dir_all(&DATA_DIR_MANAGER.temp_dir) .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + std::fs::create_dir_all(&DATA_DIR_MANAGER.eml_dir) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; Ok(()) } } @@ -78,12 +78,6 @@ impl DataDirManager { root_dir.join(EML_DIR) }; - let attachment_dir = if let Some(ref data_dir) = SETTINGS.bichon_data_dir { - PathBuf::from(data_dir).join(ATTACHMENT_DIR) - } else { - root_dir.join(ATTACHMENT_DIR) - }; - Self { root_dir: root_dir.clone(), meta_db: root_dir.join(META_FILE), @@ -94,7 +88,6 @@ impl DataDirManager { envelope_dir, temp_dir: root_dir.join(TMP_DIR), eml_dir, - attachment_dir, } } } diff --git a/src/modules/tasks/mod.rs b/src/modules/tasks/mod.rs index ac3a774..aaa6136 100644 --- a/src/modules/tasks/mod.rs +++ b/src/modules/tasks/mod.rs @@ -16,14 +16,24 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . - -use crate::modules::context::RustMailTask; +use crate::modules::common::periodic::TaskHandle; +use crate::modules::context::BichonTask; use crate::modules::oauth2::{refresh::OAuth2RefreshTask, task::OAuth2CleanTask}; -pub struct PeriodicTasks; +pub struct PeriodicTasks { + tasks: Vec, +} impl PeriodicTasks { - pub fn start_background_tasks() { - OAuth2CleanTask::start(); - OAuth2RefreshTask::start(); + pub fn setup() -> Self { + let mut tasks = Vec::new(); + tasks.push(OAuth2CleanTask::start()); + tasks.push(OAuth2RefreshTask::start()); + Self { tasks } + } + + pub async fn shutdown(self) { + for handle in self.tasks { + handle.stop().await; + } } }