diff --git a/Cargo.lock b/Cargo.lock index 00cfd3d6..2552c1ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -20,6 +20,15 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "adobe-cmap-parser" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae8abfa9a4688de8fc9f42b3f013b6fffec18ed8a554f5f113577e0b9b3212a3" +dependencies = [ + "pom", +] + [[package]] name = "aead" version = "0.5.2" @@ -130,6 +139,15 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "arc-swap" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207" +dependencies = [ + "rustversion", +] + [[package]] name = "argon2" version = "0.5.3" @@ -983,6 +1001,15 @@ dependencies = [ "serde_core", ] +[[package]] +name = "bitpacking" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96a7139abd3d9cebf8cd6f920a389cf3dc9576172e32f4563f188cae3c3eb019" +dependencies = [ + "crunchy", +] + [[package]] name = "bitvec" version = "1.0.1" @@ -1038,6 +1065,40 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "block-padding" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93" +dependencies = [ + "generic-array", +] + +[[package]] +name = "bon" +version = "3.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2f04f6fef12d70d42a77b1433c9e0f065238479a6cefc4f5bab105e9873a3c3" +dependencies = [ + "bon-macros", + "rustversion", +] + +[[package]] +name = "bon-macros" +version = "3.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d0bd4c2f75335ad98052a37efb54f428b492f64340257143b3429c8a508fa7b" +dependencies = [ + "darling", + "ident_case", + "prettyplease", + "proc-macro2", + "quote", + "rustversion", + "syn 2.0.117", +] + [[package]] name = "borsh" version = "1.6.1" @@ -1097,6 +1158,12 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "bytecount" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e" + [[package]] name = "bytemuck" version = "1.25.0" @@ -1140,6 +1207,15 @@ dependencies = [ "rustversion", ] +[[package]] +name = "cbc" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6" +dependencies = [ + "cipher", +] + [[package]] name = "cc" version = "1.2.61" @@ -1152,6 +1228,12 @@ dependencies = [ "shlex", ] +[[package]] +name = "census" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f4c707c6a209cbe82d10abd08e1ea8995e9ea937d2550646e02798948992be0" + [[package]] name = "cfb" version = "0.7.3" @@ -1163,6 +1245,12 @@ dependencies = [ "uuid", ] +[[package]] +name = "cff-parser" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31f5b6e9141c036f3ff4ce7b2f7e432b0f00dee416ddcd4f17741d189ddc2e9d" + [[package]] name = "cfg-if" version = "1.0.4" @@ -1436,6 +1524,12 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + [[package]] name = "crypto-bigint" version = "0.4.9" @@ -1557,6 +1651,40 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "darling" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +dependencies = [ + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn 2.0.117", +] + +[[package]] +name = "darling_macro" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +dependencies = [ + "darling_core", + "quote", + "syn 2.0.117", +] + [[package]] name = "dashmap" version = "5.5.3" @@ -1599,6 +1727,12 @@ dependencies = [ "matches", ] +[[package]] +name = "datasketches" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c286de4e81ea2590afc24d754e0f83810c566f50a1388fa75ebd57928c0d9745" + [[package]] name = "der" version = "0.6.1" @@ -1677,6 +1811,12 @@ version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1" +[[package]] +name = "downcast-rs" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "117240f60069e65410b3ae1bb213295bd828f707b5bec6596a1afc8793ce0cbc" + [[package]] name = "dragonbox_ecma" version = "0.1.12" @@ -1710,6 +1850,15 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "ecb" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a8bfa975b1aec2145850fcaa1c6fe269a16578c44705a532ae3edc92b8881c7" +dependencies = [ + "cipher", +] + [[package]] name = "ecdsa" version = "0.14.8" @@ -1853,6 +2002,17 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "erased-serde" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2add8a07dd6a8d93ff627029c51de145e12686fbc36ecb298ac22e74cf02dec" +dependencies = [ + "serde", + "serde_core", + "typeid", +] + [[package]] name = "errno" version = "0.3.14" @@ -1874,6 +2034,15 @@ dependencies = [ "windows-sys 0.48.0", ] +[[package]] +name = "euclid" +version = "0.20.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2bb7ef65b3777a325d1eeefefab5b6d4959da54747e33bd6258e789640f307ad" +dependencies = [ + "num-traits", +] + [[package]] name = "event-listener" version = "2.5.3" @@ -1907,6 +2076,12 @@ version = "4.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77af40d8a8dadb92dc178569a5f5edb5f3056e98255c2de48ab5d59a52892e0c" +[[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" @@ -1971,6 +2146,7 @@ checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", "miniz_oxide", + "zlib-rs", ] [[package]] @@ -2030,6 +2206,16 @@ dependencies = [ "winapi", ] +[[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" @@ -2420,6 +2606,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[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" @@ -2746,6 +2938,12 @@ dependencies = [ "flate2", ] +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + [[package]] name = "idna" version = "1.1.0" @@ -2828,6 +3026,7 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" dependencies = [ + "block-padding", "generic-array", ] @@ -2840,6 +3039,15 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "inventory" +version = "0.3.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4f0c30c76f2f4ccee3fe55a2435f691ca00c0e4bd87abe4f4a851b1d4dac39b" +dependencies = [ + "rustversion", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -2971,7 +3179,7 @@ dependencies = [ "httpdate", "idna", "mime", - "nom", + "nom 8.0.0", "percent-encoding", "quoted_printable", "rustls 0.23.40", @@ -2983,6 +3191,12 @@ dependencies = [ "webpki-roots 1.0.7", ] +[[package]] +name = "levenshtein_automata" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c2cdeb66e45e9f36bfad5bbdb4d2384e70936afbee843c6f6543f0c551ebb25" + [[package]] name = "libc" version = "0.2.186" @@ -3093,6 +3307,34 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "lopdf" +version = "0.38.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7184fdea2bc3cd272a1acec4030c321a8f9875e877b3f92a53f2f6033fdc289" +dependencies = [ + "aes", + "bitflags", + "cbc", + "ecb", + "encoding_rs", + "flate2", + "getrandom 0.3.4", + "indexmap", + "itoa", + "log", + "md-5 0.10.6", + "nom 8.0.0", + "nom_locate", + "rand 0.9.4", + "rangemap", + "sha2 0.10.9", + "stringprep", + "thiserror 2.0.18", + "ttf-parser", + "weezl", +] + [[package]] name = "lru" version = "0.16.4" @@ -3108,6 +3350,12 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" +[[package]] +name = "lz4_flex" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ef0d4ed8669f8f8826eb00dc878084aa8f253506c4fd5e8f58f5bce72ddb97e" + [[package]] name = "matchers" version = "0.2.0" @@ -3149,6 +3397,15 @@ dependencies = [ "digest 0.11.3", ] +[[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.8.0" @@ -3189,6 +3446,12 @@ dependencies = [ "unicase", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -3292,12 +3555,28 @@ dependencies = [ "version_check", ] +[[package]] +name = "murmurhash32" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2195bf6aa996a481483b29d62a7663eed3fe39600c460e323f8ff41e90bdd89b" + [[package]] name = "mutate_once" version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13d2233c9842d08cfe13f9eac96e207ca6a2ea10b80259ebe8ad0268be27d2af" +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "nom" version = "8.0.0" @@ -3307,6 +3586,17 @@ dependencies = [ "memchr", ] +[[package]] +name = "nom_locate" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b577e2d69827c4740cba2b52efaad1c4cc7c73042860b199710b3575c68438d" +dependencies = [ + "bytecount", + "memchr", + "nom 8.0.0", +] + [[package]] name = "nonmax" version = "0.5.5" @@ -3390,6 +3680,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "oneshot" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "269bca4c2591a28585d6bf10d9ed0332b7d76900a1b02bec41bdc3a2cdcda107" + [[package]] name = "opaque-debug" version = "0.3.1" @@ -3402,6 +3698,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "ordered-float" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e" +dependencies = [ + "num-traits", +] + [[package]] name = "outref" version = "0.1.0" @@ -3414,6 +3719,15 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" +[[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 = "owo-colors" version = "4.3.0" @@ -3836,6 +4150,7 @@ dependencies = [ "oxc_parser", "oxc_semantic", "oxc_span", + "pdf-extract", "percent-encoding", "quick-xml 0.39.2", "rand_core 0.6.4", @@ -3847,6 +4162,7 @@ dependencies = [ "smol_str", "socket2 0.6.3", "sqlx", + "tantivy", "tempfile", "thiserror 2.0.18", "tokio", @@ -3860,6 +4176,7 @@ dependencies = [ "urlencoding", "utoipa", "uuid", + "zip", ] [[package]] @@ -3985,6 +4302,23 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" +[[package]] +name = "pdf-extract" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e28ba1758a3d3f361459645780e09570b573fc3c82637449e9963174c813a98" +dependencies = [ + "adobe-cmap-parser", + "cff-parser", + "encoding_rs", + "euclid", + "log", + "lopdf", + "postscript", + "type1-encoding-parser", + "unicode-normalization", +] + [[package]] name = "pem" version = "3.0.6" @@ -4205,6 +4539,12 @@ dependencies = [ "universal-hash", ] +[[package]] +name = "pom" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "60f6ce597ecdcc9a098e7fddacb1065093a3d66446fa16c675e7e71d1b5c28e6" + [[package]] name = "portable-atomic" version = "1.13.1" @@ -4223,6 +4563,12 @@ dependencies = [ "serde", ] +[[package]] +name = "postscript" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78451badbdaebaf17f053fd9152b3ffb33b516104eacb45e7864aaa9c712f306" + [[package]] name = "potential_utf" version = "0.1.5" @@ -4546,6 +4892,12 @@ dependencies = [ "rand_core 0.5.1", ] +[[package]] +name = "rangemap" +version = "1.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "973443cf09a9c8656b574a866ab68dfa19f0867d0340648c7d2f6a71b8a8ea68" + [[package]] name = "rayon" version = "1.12.0" @@ -4753,6 +5105,16 @@ dependencies = [ "zeroize", ] +[[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 = "rustc-hash" version = "2.1.2" @@ -5179,6 +5541,15 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" +[[package]] +name = "sketches-ddsketch" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05e40b6cf54d988dc1a2223531b969c9a9e30906ad90ef64890c27b4bfbb46ea" +dependencies = [ + "serde", +] + [[package]] name = "slab" version = "0.4.12" @@ -5486,6 +5857,12 @@ dependencies = [ "unicode-properties", ] +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "subtle" version = "2.6.1" @@ -5540,6 +5917,154 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" +[[package]] +name = "tantivy" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edde6a10743fff00a4e1a8c9ef020bf5f3cbad301b7d2d39f2b07f123c4eac07" +dependencies = [ + "aho-corasick", + "arc-swap", + "base64 0.22.1", + "bitpacking", + "bon", + "byteorder", + "census", + "crc32fast", + "crossbeam-channel", + "datasketches", + "downcast-rs", + "fastdivide", + "fnv", + "fs4", + "htmlescape", + "itertools 0.14.0", + "levenshtein_automata", + "log", + "lru", + "lz4_flex", + "measure_time", + "memmap2", + "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-stacker", + "tantivy-tokenizer-api", + "tempfile", + "thiserror 2.0.18", + "time", + "typetag", + "uuid", + "winapi", +] + +[[package]] +name = "tantivy-bitpacker" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fed3d674429bcd2de5d0a6d1aa5495fed8afd9c5ecce993019caf7615f53fa4" +dependencies = [ + "bitpacking", +] + +[[package]] +name = "tantivy-columnar" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c57166f5bcfd478f370ab8445afb4678dce44801fa5ce5c451aaf8595583c5dc" +dependencies = [ + "downcast-rs", + "fastdivide", + "itertools 0.14.0", + "serde", + "tantivy-bitpacker", + "tantivy-common", + "tantivy-sstable", + "tantivy-stacker", +] + +[[package]] +name = "tantivy-common" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbf10915aa75da3c3b0d58b58853d2e889efbaf32d4982a4c3715dde6bba23e5" +dependencies = [ + "async-trait", + "byteorder", + "ownedbytes", + "serde", + "time", +] + +[[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.26.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfadb8526b6da90704feb293b0701a6aae62ea14983143344be2dc5ce30f1d82" +dependencies = [ + "fnv", + "nom 7.1.3", + "ordered-float", + "serde", + "serde_json", +] + +[[package]] +name = "tantivy-sstable" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a2cfc3ac5164cbadc28965ffb145a8f47582a60ae5897859ad8d4316596c606" +dependencies = [ + "futures-util", + "itertools 0.14.0", + "tantivy-bitpacker", + "tantivy-common", + "tantivy-fst", + "zstd", +] + +[[package]] +name = "tantivy-stacker" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cbb051742da9d53ca9e8fff43a9b10e319338b24e2c0e15d0372df19ffeb951" +dependencies = [ + "murmurhash32", + "tantivy-common", +] + +[[package]] +name = "tantivy-tokenizer-api" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eac258c2c6390673f2685813afeeafcb8c4e0ee7de8dd3fc46838dcc37263f98" +dependencies = [ + "serde", +] + [[package]] name = "tap" version = "1.0.1" @@ -5880,12 +6405,63 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "ttf-parser" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2df906b07856748fa3f6e0ad0cbaa047052d4a7dd609e231c4f72cee8c36f31" + +[[package]] +name = "type1-encoding-parser" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa10c302f5a53b7ad27fd42a3996e23d096ba39b5b8dd6d9e683a05b01bee749" +dependencies = [ + "pom", +] + +[[package]] +name = "typed-path" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e28f89b80c87b8fb0cf04ab448d5dd0dd0ade2f8891bae878de66a75a28600e" + +[[package]] +name = "typeid" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c" + [[package]] name = "typenum" version = "1.20.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de" +[[package]] +name = "typetag" +version = "0.2.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5a897b12c6c1151ad0b138b8db50252dc301f93bc3b027db05eec82aeed298c" +dependencies = [ + "erased-serde", + "inventory", + "once_cell", + "serde", + "typetag-impl", +] + +[[package]] +name = "typetag-impl" +version = "0.2.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf808357c6ed7e13ba0f3277ec8d8f21b2d501274895104263985330c726c1c5" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "unicase" version = "2.9.0" @@ -5984,6 +6560,12 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" +[[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" @@ -6363,6 +6945,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-sys" version = "0.60.2" @@ -6794,12 +7385,72 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "zip" +version = "8.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d04a6b5381502aa6087c94c669499eb1602eb9c5e8198e534de571f7154809b" +dependencies = [ + "crc32fast", + "flate2", + "indexmap", + "memchr", + "typed-path", + "zopfli", +] + +[[package]] +name = "zlib-rs" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3be3d40e40a133f9c916ee3f9f4fa2d9d63435b5fbe1bfc6d9dae0aa0ada1513" + [[package]] name = "zmij" version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" +[[package]] +name = "zopfli" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f05cd8797d63865425ff89b5c4a48804f35ba0ce8d125800027ad6017d2b5249" +dependencies = [ + "bumpalo", + "crc32fast", + "log", + "simd-adler32", +] + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.0.16+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +dependencies = [ + "cc", + "pkg-config", +] + [[package]] name = "zune-core" version = "0.5.1" diff --git a/Cargo.toml b/Cargo.toml index 42fa985e..3a175cae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -75,6 +75,9 @@ idna = "1.1" smol_str = { version = "0.3.2", features = ["serde"] } accept-language = "3.1.0" askama = "0.16.0" +tantivy = "0.26.1" +zip = { version = "8.6.0", default-features = false, features = ["deflate"] } +pdf-extract = "0.10.0" [features] default = [] diff --git a/example.env b/example.env index c1dffd63..126803e0 100644 --- a/example.env +++ b/example.env @@ -204,6 +204,26 @@ DATABASE_URL=postgres://postgres:postgres@localhost:5432/oxicloud # Enable search functionality (default: true) #OXICLOUD_ENABLE_SEARCH=true +# Full-text content search (embedded Tantivy index over file names AND file +# content: PDF, Office, plain text/code). Indexing runs on the maintenance +# pool, off the request path. (default: true) +#OXICLOUD_ENABLE_CONTENT_SEARCH=true + +# Content index directory (default: {OXICLOUD_STORAGE_PATH}/.search-index) +#OXICLOUD_CONTENT_INDEX_DIR= + +# Index worker drain cadence in ms — upper bound on how long a new upload +# takes to become content-searchable (default: 1500) +#OXICLOUD_CONTENT_INDEX_FLUSH_MS=1500 + +# Files larger than this are indexed by name only, no text extraction +# (default: 33554432 = 32 MiB) +#OXICLOUD_CONTENT_INDEX_MAX_FILE_BYTES=33554432 + +# Cap on extracted text per unique blob fed to the index +# (default: 1048576 = 1 MiB) +#OXICLOUD_CONTENT_INDEX_MAX_TEXT_BYTES=1048576 + # Enable music playlists and audio metadata (default: true) #OXICLOUD_ENABLE_MUSIC=true diff --git a/migrations/20260701000000_content_search_index.sql b/migrations/20260701000000_content_search_index.sql new file mode 100644 index 00000000..198d4a5c --- /dev/null +++ b/migrations/20260701000000_content_search_index.sql @@ -0,0 +1,129 @@ +-- Content-search indexing pipeline: durable dirty-queue + per-blob text cache. +-- +-- OxiCloud's search gains an embedded Tantivy (BM25) index over file NAMES and +-- file CONTENT (extracted text from PDFs, Office documents, plain text/code). +-- The Tantivy index lives on local disk ({storage}/.search-index) and is a +-- DERIVED artifact: PostgreSQL remains the single source of truth, and the +-- index can always be rebuilt from it (the app reseeds this queue whenever the +-- on-disk index is missing or its schema version changed). +-- +-- This migration installs the PG side of the pipeline: +-- +-- * `storage.search_index_dirty` — append-only dirty queue, mirroring the +-- proven `storage.tree_etag_dirty` design: statement-level triggers on +-- `storage.files` only INSERT queue rows (plain heap append, zero shared +-- row locks, no unique constraints), and a single background worker in the +-- app (`ContentIndexWorker`, on the maintenance pool) drains them and +-- applies the Tantivy mutations. Trigger-based capture means EVERY write +-- surface (REST, WebDAV, NextCloud, WOPI, batch ops, trash) is covered, +-- including paths that never call the in-process lifecycle hooks, and the +-- queue survives crashes/restarts (an in-memory channel would not). +-- +-- * `storage.blob_extracted_text` — per-BLOB extraction cache. Text is +-- derived from CONTENT, and content is content-addressed (BLAKE3), so +-- extraction is keyed by `blob_hash`, not by file: N copies of the same +-- PDF cost ONE extraction, and rename/move/copy never re-extract. +-- No FK on blob_hash: a file's hash may resolve to either +-- `storage.blobs` (legacy whole blob) or `storage.chunk_manifests` +-- (CDC file hash); the worker garbage-collects orphans instead. +-- +-- Queue semantics: +-- * Duplicates are expected and harmless — upserts are idempotent +-- (Tantivy delete_term + add_document) and the worker dedups per drain +-- batch keeping the latest op per file. +-- * The worker deletes queue rows ONLY AFTER the Tantivy commit succeeds, +-- so a crash between drain and commit re-processes the batch (at-least- +-- once delivery; idempotency makes that safe). +-- * When content search is disabled the app still drains the queue +-- (discard-only janitor) so it cannot grow unboundedly; re-enabling +-- triggers a full reseed via the index-version check anyway. +-- +-- UPDATE value filter: only changes observable by the index enqueue — +-- a rename (name), a content swap (blob_hash) or a trash transition +-- (is_trashed). The EXIF `media_sort_date` sync and other metadata-only +-- updates never enqueue. Trashed files are removed from the index (they +-- must not appear in search) and re-added on restore. + +-- ── Dirty queue ────────────────────────────────────────────────────── +CREATE TABLE IF NOT EXISTS storage.search_index_dirty ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + file_id UUID NOT NULL, + op TEXT NOT NULL CHECK (op IN ('upsert', 'delete')) +); + +-- ── Per-blob extracted text cache ──────────────────────────────────── +-- `text` is NULL unless status = 'ok'. `status` is terminal per blob — +-- a failed/unsupported blob is never retried until an extractor-version +-- bump wipes the row (cheap: DELETE WHERE extractor <> current). +CREATE TABLE IF NOT EXISTS storage.blob_extracted_text ( + blob_hash VARCHAR(64) PRIMARY KEY, + text TEXT, + status TEXT NOT NULL CHECK (status IN ('ok', 'empty', 'failed', 'too_large')), + extractor TEXT NOT NULL, + extracted_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- ── File side: INSERT / DELETE ─────────────────────────────────────── +-- Both triggers alias their transition table to `changed_rows`; one body +-- serves both events. TG_OP picks the queue op. INSERTs of already-trashed +-- rows (trash restores go through UPDATE) are skipped. +CREATE OR REPLACE FUNCTION storage.enqueue_search_from_files_stmt() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'INSERT' THEN + INSERT INTO storage.search_index_dirty (file_id, op) + SELECT id, 'upsert' FROM changed_rows WHERE NOT is_trashed; + ELSE + INSERT INTO storage.search_index_dirty (file_id, op) + SELECT id, 'delete' FROM changed_rows; + END IF; + + RETURN NULL; +END; +$$; + +-- ── File side: UPDATE ──────────────────────────────────────────────── +-- Value filter: only rename / content swap / trash transitions enqueue. +-- (PostgreSQL forbids `AFTER UPDATE OF ` with transition tables, +-- so the filter lives here.) A row trashed by the update enqueues a +-- 'delete'; everything else enqueues an 'upsert' re-index. +CREATE OR REPLACE FUNCTION storage.enqueue_search_from_files_stmt_upd() +RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + INSERT INTO storage.search_index_dirty (file_id, op) + SELECT n.id, + CASE WHEN n.is_trashed THEN 'delete' ELSE 'upsert' END + FROM old_rows o + JOIN new_rows n USING (id) + WHERE (o.name, o.blob_hash, o.is_trashed) + IS DISTINCT FROM + (n.name, n.blob_hash, n.is_trashed); + + RETURN NULL; +END; +$$; + +-- PG 13 compatibility: DROP-then-CREATE (no CREATE OR REPLACE TRIGGER). +DROP TRIGGER IF EXISTS files_enqueue_search_ins ON storage.files; +CREATE TRIGGER files_enqueue_search_ins + AFTER INSERT ON storage.files + REFERENCING NEW TABLE AS changed_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.enqueue_search_from_files_stmt(); + +DROP TRIGGER IF EXISTS files_enqueue_search_del ON storage.files; +CREATE TRIGGER files_enqueue_search_del + AFTER DELETE ON storage.files + REFERENCING OLD TABLE AS changed_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.enqueue_search_from_files_stmt(); + +DROP TRIGGER IF EXISTS files_enqueue_search_upd ON storage.files; +CREATE TRIGGER files_enqueue_search_upd + AFTER UPDATE ON storage.files + REFERENCING OLD TABLE AS old_rows NEW TABLE AS new_rows + FOR EACH STATEMENT EXECUTE FUNCTION storage.enqueue_search_from_files_stmt_upd(); + +-- ── Backfill ───────────────────────────────────────────────────────── +-- Existing deployments: queue every live file once so the first worker +-- run indexes the historical corpus. Fresh databases enqueue nothing. +INSERT INTO storage.search_index_dirty (file_id, op) +SELECT id, 'upsert' FROM storage.files WHERE NOT is_trashed; diff --git a/src/application/dtos/search_dto.rs b/src/application/dtos/search_dto.rs index 5713f712..7236f1f6 100644 --- a/src/application/dtos/search_dto.rs +++ b/src/application/dtos/search_dto.rs @@ -134,6 +134,14 @@ pub struct SearchFileResultDto { /// that pre-date the column. #[serde(default)] pub blob_hash: String, + /// Plain-text fragment around the first content match, present only for + /// hits discovered through the full-text content index. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub snippet: Option, + /// Where the match came from: "name" (filename matched the query) or + /// "content" (discovered via the full-text content index). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub match_source: Option, } /// A folder search result enriched with server-computed metadata diff --git a/src/application/ports/content_index_ports.rs b/src/application/ports/content_index_ports.rs new file mode 100644 index 00000000..c407d0af --- /dev/null +++ b/src/application/ports/content_index_ports.rs @@ -0,0 +1,49 @@ +//! Content Index Port - Application layer abstraction for full-text search +//! over file names and extracted file content. +//! +//! The implementation (an embedded Tantivy BM25 index) lives in the +//! infrastructure layer; `SearchService` only sees this port. The index is a +//! DERIVED artifact fed asynchronously by a background worker — it never sits +//! on a request path, and PostgreSQL remains the source of truth (hits are +//! re-validated and hydrated through SQL before they reach the caller, so a +//! stale index can only ever produce a dropped candidate, never a leak). + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::common::errors::DomainError; + +/// One content-index hit: a candidate file id with its BM25 score and an +/// optional plain-text snippet around the first match. +/// +/// `file_id` is a CANDIDATE — callers must hydrate it through the metadata +/// repository (which re-applies user scoping, trash state and the active +/// search filters) before exposing it. +#[derive(Debug, Clone)] +pub struct ContentHitDto { + /// File UUID as string (matches `storage.files.id`). + pub file_id: String, + /// BM25 relevance score (positive, unbounded — normalize per result set). + pub score: f32, + /// Plain-text fragment around the first matched term, when available. + pub snippet: Option, +} + +/// Port for querying the full-text content index. +/// +/// `#[async_trait]` is used so the trait is dyn-compatible — `SearchService` +/// holds an `Option>` (the feature is toggleable). +#[async_trait] +pub trait ContentIndexPort: Send + Sync + 'static { + /// Search indexed file names + content for `query`, scoped to `user_id`. + /// + /// Returns up to `limit` hits sorted by BM25 score descending. Matching is + /// tokenized (not substring): exact terms, typo-tolerant fuzzy terms + /// (edit distance 1) and prefix expansion on the last query token. + async fn search_content( + &self, + user_id: Uuid, + query: &str, + limit: usize, + ) -> Result, DomainError>; +} diff --git a/src/application/ports/mod.rs b/src/application/ports/mod.rs index a6e756ea..163cb70c 100644 --- a/src/application/ports/mod.rs +++ b/src/application/ports/mod.rs @@ -7,6 +7,7 @@ pub mod calendar_ports; pub mod carddav_ports; pub mod chunked_upload_ports; pub mod compression_ports; +pub mod content_index_ports; pub mod dedup_ports; pub mod email_sender; pub mod favorites_ports; diff --git a/src/application/services/search_service.rs b/src/application/services/search_service.rs index 5ce252ab..c733514e 100644 --- a/src/application/services/search_service.rs +++ b/src/application/services/search_service.rs @@ -11,6 +11,7 @@ use crate::application::dtos::search_dto::{ SearchCriteriaDto, SearchFileResultDto, SearchFolderResultDto, SearchResultsDto, SearchSuggestionItem, SearchSuggestionsDto, }; +use crate::application::ports::content_index_ports::{ContentHitDto, ContentIndexPort}; use crate::application::ports::inbound::SearchUseCase; use crate::application::ports::storage_ports::FileReadPort; use crate::common::errors::Result; @@ -44,6 +45,11 @@ pub struct SearchService { /// Repository for folder operations folder_repository: Arc, + /// Optional full-text content index (embedded Tantivy). When present, + /// query-bearing searches additionally surface files whose CONTENT + /// matches; hits are hydrated and re-filtered through SQL before use. + content_index: Option>, + /// Lock-free concurrent cache with automatic TTL and LRU eviction (moka). /// Values are `Arc` so cache insert/hit is a single /// atomic ref-count increment (~1 ns) instead of cloning thousands of Strings. @@ -73,6 +79,35 @@ fn compute_relevance(name: &str, query_lower: &str) -> u32 { } } +/// Max content-index candidates fetched per search. Hydration re-filters +/// them in ONE SQL round-trip, so this bounds both index and DB work. +const CONTENT_HITS_LIMIT: usize = 200; + +/// Map a BM25 score into the 10–45 relevance band, normalized against the +/// best score of the result set. Deliberately below the weakest name match +/// (contains = 50): a filename hit is more specific than a body mention. +fn content_relevance(score: f32, max_score: f32) -> u32 { + if !score.is_finite() || max_score <= 0.0 { + return 10; + } + let ratio = (score / max_score).clamp(0.0, 1.0); + 10 + (ratio * 35.0).round() as u32 +} + +/// Re-sort the merged file list with the same semantics the folder list +/// uses. Only invoked when content hits were merged into a SQL-ordered page. +fn sort_enriched_files(files: &mut [SearchFileResultDto], sort_by: &str) { + match sort_by { + "name" => files.sort_by_cached_key(|f| f.name.to_lowercase()), + "name_desc" => files.sort_by_cached_key(|f| Reverse(f.name.to_lowercase())), + "date" => files.sort_by_key(|f| f.modified_at), + "date_desc" => files.sort_by_key(|f| Reverse(f.modified_at)), + "size" => files.sort_by_key(|f| f.size), + "size_desc" => files.sort_by_key(|f| Reverse(f.size)), + _ => files.sort_by_key(|f| Reverse(f.relevance_score)), + } +} + /// Format bytes into a human-readable string (e.g. "2.5 MB"). fn format_bytes(bytes: u64) -> String { const UNITS: &[&str] = &["B", "KB", "MB", "GB", "TB"]; @@ -115,6 +150,7 @@ impl SearchService { pub fn new( file_repository: Arc, folder_repository: Arc, + content_index: Option>, cache_ttl: u64, max_cache_size: usize, ) -> Self { @@ -126,6 +162,7 @@ impl SearchService { Self { file_repository, folder_repository, + content_index, search_cache, } } @@ -176,6 +213,8 @@ impl SearchService { // responses on the NC surface can emit the same ETag // (`File::compute_etag`) as PROPFIND/GET would. blob_hash: file.content_hash.clone(), + snippet: None, + match_source: (!query_lower.is_empty() && relevance > 0).then(|| "name".to_string()), } } @@ -201,6 +240,108 @@ impl SearchService { } } + /// Query the content index for files matching by CONTENT (when the index + /// is enabled). First page only — content hits have no stable + /// interleaving with SQL pagination beyond it, and page one is where + /// search UX lives. Index failures degrade to name-only results, never + /// to a failed search. + async fn lookup_content_hits( + &self, + criteria: &SearchCriteriaDto, + user_id: Uuid, + ) -> Vec { + let Some(index) = &self.content_index else { + return Vec::new(); + }; + if criteria.offset != 0 { + return Vec::new(); + } + let Some(query) = criteria + .name_contains + .as_deref() + .map(str::trim) + .filter(|q| q.len() >= 2) + else { + return Vec::new(); + }; + + match index + .search_content(user_id, query, CONTENT_HITS_LIMIT) + .await + { + Ok(hits) => hits, + Err(e) => { + tracing::warn!("Content-index lookup failed — returning name-only results: {e}"); + Vec::new() + } + } + } + + /// Merge content-index hits into the name-search result page: + /// * files the name search already found just gain their `snippet`; + /// * content-only candidates are hydrated through SQL in one round-trip + /// (re-applying user scope, trash state and every active filter — a + /// stale index id silently drops out), enriched, scored into the + /// content relevance band and appended; + /// * the merged page is re-sorted with the caller's `sort_by`. + /// + /// Returns how many files were added (callers bump their totals by it). + async fn merge_content_hits( + &self, + hits: Vec, + enriched_files: &mut Vec, + criteria: &SearchCriteriaDto, + user_id: Uuid, + ) -> Result { + if hits.is_empty() { + return Ok(0); + } + + let mut by_id: std::collections::HashMap<&str, &ContentHitDto> = + hits.iter().map(|h| (h.file_id.as_str(), h)).collect(); + for file in enriched_files.iter_mut() { + if let Some(hit) = by_id.remove(file.id.as_str()) { + file.snippet = hit.snippet.clone(); + } + } + if by_id.is_empty() { + return Ok(0); + } + + // Preserve the index's score order when collecting the leftovers. + let candidate_ids: Vec = hits + .iter() + .filter(|h| by_id.contains_key(h.file_id.as_str())) + .map(|h| h.file_id.clone()) + .collect(); + let files = self + .file_repository + .fetch_files_by_ids_filtered(&candidate_ids, criteria, user_id) + .await?; + if files.is_empty() { + return Ok(0); + } + + let max_score = hits.iter().map(|h| h.score).fold(0.0_f32, f32::max); + let mut added = 0usize; + for file in files { + let dto = FileDto::from(file); + let Some(hit) = by_id.get(dto.id.as_str()) else { + continue; + }; + let mut enriched = Self::enrich_file(&dto, ""); + enriched.relevance_score = content_relevance(hit.score, max_score); + enriched.snippet = hit.snippet.clone(); + enriched.match_source = Some("content".to_string()); + enriched_files.push(enriched); + added += 1; + } + if added > 0 { + sort_enriched_files(enriched_files, &criteria.sort_by); + } + Ok(added) + } + /// Quick suggestions search — returns up to `limit` name suggestions /// matching the query. Pushes filtering, relevance sort and LIMIT to SQL /// so only a handful of rows cross the DB→app boundary. @@ -305,6 +446,10 @@ impl SearchUseCase for SearchService { // Pre-compute once — avoids N heap allocations inside enrich_file/enrich_folder. let query_lower = query.to_lowercase(); + // Content-index candidates (first page only). Feature-off or an + // index failure yields an empty set — the search stays name-only. + let content_hits = self.lookup_content_hits(&criteria, user_id).await; + // For non-recursive searches, use efficient database-level pagination // This avoids loading all files into memory if !criteria.recursive { @@ -316,7 +461,7 @@ impl SearchUseCase for SearchService { // Convert to DTOs and enrich with metadata let file_dtos: Vec = files.into_iter().map(FileDto::from).collect(); - let enriched_files: Vec = file_dtos + let mut enriched_files: Vec = file_dtos .iter() .map(|f| Self::enrich_file(f, &query_lower)) .collect(); @@ -360,6 +505,12 @@ impl SearchUseCase for SearchService { } } + // Blend in content-discovered files before the pagination math. + let added = self + .merge_content_hits(content_hits, &mut enriched_files, &criteria, user_id) + .await?; + let total_file_count = total_file_count + added; + let folder_count = enriched_folders.len(); let total_count = total_file_count + folder_count; @@ -416,7 +567,7 @@ impl SearchUseCase for SearchService { // ── Convert to DTOs and enrich with server-computed metadata ── let file_dtos: Vec = found_files.into_iter().map(FileDto::from).collect(); - let enriched_files: Vec = file_dtos + let mut enriched_files: Vec = file_dtos .iter() .map(|f| Self::enrich_file(f, &query_lower)) .collect(); @@ -446,6 +597,12 @@ impl SearchUseCase for SearchService { } } + // Blend in content-discovered files before the pagination math. + let added = self + .merge_content_hits(content_hits, &mut enriched_files, &criteria, user_id) + .await?; + let total_file_count = total_file_count + added; + // ── Pagination (folders first, then files) ── let folder_count = enriched_folders.len(); let total_count = total_file_count + folder_count; @@ -535,3 +692,60 @@ impl SearchService { SearchServiceStub } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn content_relevance_stays_below_name_contains_band() { + // Best hit of the set caps at 45 — always under contains (50). + assert_eq!(content_relevance(8.0, 8.0), 45); + assert_eq!(content_relevance(4.0, 8.0), 28); + // Degenerate inputs fall to the floor instead of panicking. + assert_eq!(content_relevance(1.0, 0.0), 10); + assert_eq!(content_relevance(f32::NAN, 8.0), 10); + assert!(content_relevance(0.0, 8.0) >= 10); + } + + fn dto(name: &str, relevance: u32, size: u64, modified_at: u64) -> SearchFileResultDto { + SearchFileResultDto { + id: name.to_string(), + name: name.to_string(), + path: format!("/{name}"), + size, + mime_type: "text/plain".to_string(), + folder_id: None, + created_at: 0, + modified_at, + relevance_score: relevance, + size_formatted: String::new(), + icon_class: String::new(), + icon_special_class: String::new(), + category: String::new(), + blob_hash: String::new(), + snippet: None, + match_source: None, + } + } + + #[test] + fn merged_files_resort_by_relevance_and_by_column() { + let mut files = vec![ + dto("b-content.txt", 30, 10, 200), + dto("a-name.txt", 80, 99, 100), + ]; + sort_enriched_files(&mut files, "relevance"); + assert_eq!( + files[0].name, "a-name.txt", + "name match must outrank content match" + ); + + sort_enriched_files(&mut files, "size_desc"); + assert_eq!(files[0].name, "a-name.txt"); + sort_enriched_files(&mut files, "date"); + assert_eq!(files[0].name, "a-name.txt"); + sort_enriched_files(&mut files, "name_desc"); + assert_eq!(files[0].name, "b-content.txt"); + } +} diff --git a/src/common/config.rs b/src/common/config.rs index ffc89a93..963454e2 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -905,6 +905,44 @@ impl Default for FeaturesConfig { } } +/// Content-search configuration (embedded Tantivy index over file names and +/// extracted file content). +/// +/// The index is a derived artifact fed by a background worker on the +/// maintenance pool — none of these knobs affect request-path latency. +#[derive(Debug, Clone)] +pub struct ContentSearchConfig { + /// Master switch. When disabled, search falls back to name-only SQL and + /// a janitor keeps the (always-installed) dirty queue empty. + /// Env: `OXICLOUD_ENABLE_CONTENT_SEARCH`. + pub enabled: bool, + /// Index directory. Default: `{storage_path}/.search-index`. + /// Env: `OXICLOUD_CONTENT_INDEX_DIR`. + pub index_dir: Option, + /// Worker drain cadence in milliseconds — the upper bound on how long a + /// new upload takes to become content-searchable. Default: 1500. + /// Env: `OXICLOUD_CONTENT_INDEX_FLUSH_MS`. + pub flush_interval_ms: u64, + /// Files larger than this are indexed by NAME only (no text extraction). + /// Default: 32 MiB. Env: `OXICLOUD_CONTENT_INDEX_MAX_FILE_BYTES`. + pub max_extract_file_bytes: u64, + /// Hard cap on extracted text per blob fed to the index. Default: 1 MiB. + /// Env: `OXICLOUD_CONTENT_INDEX_MAX_TEXT_BYTES`. + pub max_text_bytes: usize, +} + +impl Default for ContentSearchConfig { + fn default() -> Self { + Self { + enabled: true, + index_dir: None, + flush_interval_ms: 1500, + max_extract_file_bytes: 32 * 1024 * 1024, + max_text_bytes: 1024 * 1024, + } + } +} + /// Global application configuration #[derive(Debug, Clone)] pub struct AppConfig { @@ -944,6 +982,8 @@ pub struct AppConfig { pub magic_link: MagicLinkConfig, /// I18n configuration (default locale for server-rendered surfaces) pub i18n: I18nConfig, + /// Content-search configuration (embedded full-text index) + pub content_search: ContentSearchConfig, } /// Server-side i18n knobs. @@ -995,6 +1035,7 @@ impl Default for AppConfig { smtp: SmtpConfig::default(), magic_link: MagicLinkConfig::default(), i18n: I18nConfig::default(), + content_search: ContentSearchConfig::default(), } } } @@ -1257,6 +1298,33 @@ impl AppConfig { config.features.enable_music = val; } + // Content search (embedded Tantivy index) + if let Ok(v) = env::var("OXICLOUD_ENABLE_CONTENT_SEARCH").map(|v| v.parse::()) + && let Ok(val) = v + { + config.content_search.enabled = val; + } + if let Ok(dir) = env::var("OXICLOUD_CONTENT_INDEX_DIR") + && !dir.trim().is_empty() + { + config.content_search.index_dir = Some(PathBuf::from(dir.trim())); + } + if let Ok(v) = env::var("OXICLOUD_CONTENT_INDEX_FLUSH_MS").map(|v| v.parse::()) + && let Ok(val) = v + { + config.content_search.flush_interval_ms = val; + } + if let Ok(v) = env::var("OXICLOUD_CONTENT_INDEX_MAX_FILE_BYTES").map(|v| v.parse::()) + && let Ok(val) = v + { + config.content_search.max_extract_file_bytes = val; + } + if let Ok(v) = env::var("OXICLOUD_CONTENT_INDEX_MAX_TEXT_BYTES").map(|v| v.parse::()) + && let Ok(val) = v + { + config.content_search.max_text_bytes = val; + } + if let Ok(v) = env::var("OXICLOUD_EXPOSE_SYSTEM_USERS").map(|v| v.parse::()) && let Ok(val) = v { diff --git a/src/common/di.rs b/src/common/di.rs index 08b483f0..0c9fcb19 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -40,6 +40,8 @@ use crate::infrastructure::services::file_system_i18n_service::FileSystemI18nSer use crate::infrastructure::services::nextcloud_chunked_upload_service::NextcloudChunkedUploadService; use crate::infrastructure::services::path_service::PathService; use crate::infrastructure::services::pg_acl_engine::PgAclEngine; +use crate::infrastructure::services::search_index::content_index_worker::ContentIndexWorker; +use crate::infrastructure::services::search_index::tantivy_content_index::TantivyContentIndex; use crate::infrastructure::services::trash_cleanup_service::TrashCleanupService; use crate::application::services::app_password_service::AppPasswordService; @@ -439,6 +441,7 @@ impl AppServiceFactory { repos: &RepositoryServices, trash_service: Option>, authz: &Arc, + content_index: Option>, ) -> ApplicationServices { // Main services let folder_service = Arc::new(FolderService::new( @@ -484,10 +487,15 @@ impl AppServiceFactory { let i18n_service = Arc::new(I18nApplicationService::new(repos.i18n_repository.clone())); - // Search service with cache + // Search service with cache. The optional content index widens the + // same `/api/search` endpoint to full-text content matches. + let content_index_port: Option< + Arc, + > = content_index.map(|idx| idx as _); let search_service: Option> = Some(Arc::new(SearchService::new( repos.file_read_repository.clone(), repos.folder_repository.clone(), + content_index_port, 300, // Cache TTL in seconds (5 minutes) 1000, // Maximum cache entries ))); @@ -688,6 +696,66 @@ impl AppServiceFactory { tracing::info!("Tree-ETag flush service initialized"); } + /// Opens (or rebuilds) the embedded Tantivy content index. Returns the + /// index plus a reseed flag (true when the on-disk index was missing or + /// version-stale and must be repopulated from `storage.files`). Any + /// failure degrades to name-only search instead of failing startup. + fn create_content_index(&self) -> Option<(Arc, bool)> { + if !self.config.content_search.enabled { + tracing::info!("Content search is disabled in configuration"); + return None; + } + let dir = self + .config + .content_search + .index_dir + .clone() + .unwrap_or_else(|| self.storage_path.join(".search-index")); + + match TantivyContentIndex::open_or_rebuild(&dir) { + Ok((index, needs_reseed)) => { + tracing::info!( + "Content index ready at {} ({} doc(s), reseed: {})", + dir.display(), + index.num_docs(), + needs_reseed + ); + Some((Arc::new(index), needs_reseed)) + } + Err(e) => { + tracing::error!("Content index unavailable — search will be name-only: {e}"); + None + } + } + } + + /// Starts the content-index pipeline on the maintenance pool. The + /// `storage.files` triggers enqueue unconditionally, so when the feature + /// is off (or the index failed to open) a discard-only janitor keeps the + /// dirty queue bounded instead. + fn start_content_index_job( + &self, + maintenance_pool: &Arc, + core: &CoreServices, + content_index: Option<(Arc, bool)>, + ) { + match content_index { + Some((index, needs_reseed)) => { + ContentIndexWorker::new( + maintenance_pool.clone(), + core.dedup_service.clone(), + index, + self.config.content_search.flush_interval_ms, + self.config.content_search.max_extract_file_bytes, + self.config.content_search.max_text_bytes, + ) + .start(needs_reseed); + tracing::info!("Content-index worker initialized"); + } + None => ContentIndexWorker::start_drain_only_janitor(maintenance_pool.clone()), + } + } + /// Builds the complete AppState using all factory services. /// /// This is the main entry point that replaces all manual logic in `main.rs`. @@ -731,9 +799,19 @@ impl AppServiceFactory { .create_trash_service(&repos, &core, &authorization) .await; + // 3c. Content index (embedded Tantivy) — opened before application + // services so SearchService can hold the query port; the feeding + // worker starts further down with the maintenance pool. + let content_index = self.create_content_index(); + // 4. Application services (with trash + authz already wired) - let mut apps = - self.create_application_services(&core, &repos, trash_service.clone(), &authorization); + let mut apps = self.create_application_services( + &core, + &repos, + trash_service.clone(), + &authorization, + content_index.as_ref().map(|(idx, _)| idx.clone()), + ); // 5. Share service let share_service = self.create_share_service(&repos, &pool, &authorization); @@ -778,6 +856,8 @@ impl AppServiceFactory { self.start_tree_etag_flush_job(&maintenance_pool); + self.start_content_index_job(&maintenance_pool, &core, content_index); + // User-lifecycle dispatcher. Hook order is registration order; // document dependencies inline if/when any arise. Today: // 1. AuditLifecycleHook — fires first so the diff --git a/src/infrastructure/repositories/pg/file_blob_read_repository.rs b/src/infrastructure/repositories/pg/file_blob_read_repository.rs index 35ffaf45..f2fefdfe 100644 --- a/src/infrastructure/repositories/pg/file_blob_read_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_read_repository.rs @@ -53,6 +53,93 @@ type FileRow = ( Option, ); +/// Append the optional type/date/size filters from `criteria` to +/// `conditions`, continuing placeholder numbering from `bind_idx`. Returns +/// the last placeholder index used. The name filter is NOT handled here — +/// it is search-flavour specific (ILIKE for name search, absent for +/// content-hit hydration). Mirror of [`bind_criteria_filters`]; the two +/// must stay in sync. +fn push_criteria_filters( + conditions: &mut Vec, + mut bind_idx: u32, + criteria: &SearchCriteriaDto, +) -> u32 { + if let Some(types) = &criteria.file_types + && !types.is_empty() + { + bind_idx += 1; + conditions.push(format!( + "LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})" + )); + } + if criteria.created_after.is_some() { + bind_idx += 1; + conditions.push(format!( + "EXTRACT(EPOCH FROM fi.created_at)::bigint >= ${bind_idx}" + )); + } + if criteria.created_before.is_some() { + bind_idx += 1; + conditions.push(format!( + "EXTRACT(EPOCH FROM fi.created_at)::bigint <= ${bind_idx}" + )); + } + if criteria.modified_after.is_some() { + bind_idx += 1; + conditions.push(format!( + "EXTRACT(EPOCH FROM fi.updated_at)::bigint >= ${bind_idx}" + )); + } + if criteria.modified_before.is_some() { + bind_idx += 1; + conditions.push(format!( + "EXTRACT(EPOCH FROM fi.updated_at)::bigint <= ${bind_idx}" + )); + } + if criteria.min_size.is_some() { + bind_idx += 1; + conditions.push(format!("fi.size >= ${bind_idx}")); + } + if criteria.max_size.is_some() { + bind_idx += 1; + conditions.push(format!("fi.size <= ${bind_idx}")); + } + bind_idx +} + +/// Bind the values for the filters appended by [`push_criteria_filters`], +/// in the same order. +fn bind_criteria_filters<'q, O>( + mut query: sqlx::query::QueryAs<'q, sqlx::Postgres, O, sqlx::postgres::PgArguments>, + criteria: &SearchCriteriaDto, +) -> sqlx::query::QueryAs<'q, sqlx::Postgres, O, sqlx::postgres::PgArguments> { + if let Some(types) = &criteria.file_types + && !types.is_empty() + { + let lower_types: Vec = types.iter().map(|t| t.to_lowercase()).collect(); + query = query.bind(lower_types); + } + if let Some(v) = criteria.created_after { + query = query.bind(v as i64); + } + if let Some(v) = criteria.created_before { + query = query.bind(v as i64); + } + if let Some(v) = criteria.modified_after { + query = query.bind(v as i64); + } + if let Some(v) = criteria.modified_before { + query = query.bind(v as i64); + } + if let Some(v) = criteria.min_size { + query = query.bind(v as i64); + } + if let Some(v) = criteria.max_size { + query = query.bind(v as i64); + } + query +} + /// File read repository backed by PostgreSQL metadata + blob storage. pub struct FileBlobReadRepository { pool: Arc, @@ -94,6 +181,80 @@ impl FileBlobReadRepository { self.hash_cache.clone() } + /// Hydrate content-index candidate ids into `File`s, re-applying the + /// caller's scope and the active search filters (owner, trash state, + /// folder scope, types, dates, sizes). The NAME filter is deliberately + /// NOT applied — content hits don't need to match it. Ids that fail any + /// filter (or no longer exist — the index is eventually consistent) + /// simply drop out, so a stale index can never leak a result. + pub async fn fetch_files_by_ids_filtered( + &self, + ids: &[String], + criteria: &SearchCriteriaDto, + user_id: Uuid, + ) -> Result, DomainError> { + // Index hits are externally produced strings — parse defensively. + let uuid_ids: Vec = ids.iter().filter_map(|id| id.parse().ok()).collect(); + if uuid_ids.is_empty() { + return Ok(Vec::new()); + } + + let mut conditions: Vec = vec![ + "fi.id = ANY($1)".to_string(), + "fi.user_id = $2".to_string(), + "fi.is_trashed = false".to_string(), + ]; + let mut bind_idx = 2u32; + + if criteria.folder_id.is_some() { + bind_idx += 1; + if criteria.recursive { + conditions.push(format!( + "fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = ${bind_idx}::uuid)" + )); + } else { + conditions.push(format!("fi.folder_id = ${bind_idx}::uuid")); + } + } + push_criteria_filters(&mut conditions, bind_idx, criteria); + + let where_clause = conditions.join(" AND "); + let sql = format!( + "SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path, \ + fi.size, fi.mime_type, \ + EXTRACT(EPOCH FROM fi.created_at)::bigint, \ + EXTRACT(EPOCH FROM fi.updated_at)::bigint, \ + fi.blob_hash, \ + fi.user_id \ + FROM storage.files fi \ + LEFT JOIN storage.folders fo ON fo.id = fi.folder_id \ + WHERE {where_clause}" + ); + + let mut query = sqlx::query_as::<_, FileRow>(&sql) + .bind(uuid_ids) + .bind(user_id); + if let Some(folder_id) = criteria.folder_id.as_deref() { + query = query.bind(folder_id); + } + query = bind_criteria_filters(query, criteria); + + let rows = query.fetch_all(self.pool.as_ref()).await.map_err(|e| { + DomainError::internal_error("FileBlobRead", format!("hydrate by ids: {e}")) + })?; + + rows.into_iter() + .map( + |(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| { + Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid) + }, + ) + .collect::, _>>() + .map_err(|e| { + DomainError::internal_error("FileBlobRead", format!("hydrate mapping: {e}")) + }) + } + /// Returns the user_id (owner) for a given file ID. /// Mirrors `FolderDbRepository::get_folder_user_id`. /// Used by the AuthorizationEngine for owner short-circuit. @@ -1021,46 +1182,7 @@ impl FileReadPort for FileBlobReadRepository { bind_idx += 1; conditions.push(format!("fi.name ILIKE ${bind_idx}")); } - if let Some(types) = &criteria.file_types - && !types.is_empty() - { - bind_idx += 1; - conditions.push(format!( - "LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})" - )); - } - if criteria.created_after.is_some() { - bind_idx += 1; - conditions.push(format!( - "EXTRACT(EPOCH FROM fi.created_at)::bigint >= ${bind_idx}" - )); - } - if criteria.created_before.is_some() { - bind_idx += 1; - conditions.push(format!( - "EXTRACT(EPOCH FROM fi.created_at)::bigint <= ${bind_idx}" - )); - } - if criteria.modified_after.is_some() { - bind_idx += 1; - conditions.push(format!( - "EXTRACT(EPOCH FROM fi.updated_at)::bigint >= ${bind_idx}" - )); - } - if criteria.modified_before.is_some() { - bind_idx += 1; - conditions.push(format!( - "EXTRACT(EPOCH FROM fi.updated_at)::bigint <= ${bind_idx}" - )); - } - if criteria.min_size.is_some() { - bind_idx += 1; - conditions.push(format!("fi.size >= ${bind_idx}")); - } - if criteria.max_size.is_some() { - bind_idx += 1; - conditions.push(format!("fi.size <= ${bind_idx}")); - } + bind_idx = push_criteria_filters(&mut conditions, bind_idx, criteria); let where_clause = conditions.join(" AND "); let limit_bind = bind_idx + 1; @@ -1107,30 +1229,7 @@ impl FileReadPort for FileBlobReadRepository { { query = query.bind(super::like_escape(name)); } - if let Some(types) = &criteria.file_types - && !types.is_empty() - { - let lower_types: Vec = types.iter().map(|t| t.to_lowercase()).collect(); - query = query.bind(lower_types); - } - if let Some(v) = criteria.created_after { - query = query.bind(v as i64); - } - if let Some(v) = criteria.created_before { - query = query.bind(v as i64); - } - if let Some(v) = criteria.modified_after { - query = query.bind(v as i64); - } - if let Some(v) = criteria.modified_before { - query = query.bind(v as i64); - } - if let Some(v) = criteria.min_size { - query = query.bind(v as i64); - } - if let Some(v) = criteria.max_size { - query = query.bind(v as i64); - } + query = bind_criteria_filters(query, criteria); query = query.bind(limit).bind(offset); diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 73144245..4e73e1b6 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -23,6 +23,7 @@ pub mod path_service; pub mod pg_acl_engine; pub mod retry_blob_backend; pub mod s3_blob_backend; +pub mod search_index; pub mod share_unlock_cookie; pub mod smtp_email_sender; pub mod thumbnail_service; diff --git a/src/infrastructure/services/search_index/content_index_worker.rs b/src/infrastructure/services/search_index/content_index_worker.rs new file mode 100644 index 00000000..bcc74781 --- /dev/null +++ b/src/infrastructure/services/search_index/content_index_worker.rs @@ -0,0 +1,415 @@ +//! Background drainer for `storage.search_index_dirty` — the asynchronous +//! half of content indexing (see migration `20260701000000_content_search_index`). +//! +//! The statement triggers on `storage.files` only append "index me" requests +//! to the queue, taking zero locks on user write paths. This worker turns the +//! requests into Tantivy mutations: every `interval_ms` it drains a batch, +//! re-reads the CURRENT file state (the queue row is a hint, not a payload), +//! extracts text once per unique blob, applies one batched Tantivy commit and +//! only then deletes the processed queue rows. +//! +//! Correctness invariants: +//! * At-least-once: queue rows are deleted AFTER the Tantivy commit. A +//! crash in between re-processes the batch — harmless, upserts are +//! idempotent (delete_term + add_document keyed by file_id). +//! * Deletes are selected by exact id (`id = ANY(...)`), never by range — +//! a transaction that began before our SELECT can commit a smaller id +//! afterwards, and a range delete would discard it unprocessed. +//! * Latest-op-wins per file within a batch; the authoritative state is +//! re-fetched from `storage.files` at drain time anyway (a file trashed +//! after its 'upsert' was queued simply turns into a delete). +//! * Extraction is keyed by blob hash (content-addressed): N files sharing +//! a blob cost ONE extraction, renames/moves cost zero re-extraction. +//! Terminal outcomes (ok/empty/failed/too_large) are cached in +//! `storage.blob_extracted_text`; transient blob-read errors store +//! nothing so the next event retries. +//! +//! Resource budget: single worker, one extraction at a time inside +//! `spawn_blocking`, single-threaded Tantivy writer — the pipeline trickles +//! along on the maintenance pool and never competes with request latency. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +use sqlx::PgPool; +use tracing::{debug, error, info, instrument, warn}; +use uuid::Uuid; + +use crate::infrastructure::services::dedup_service::DedupService; +use crate::infrastructure::services::search_index::tantivy_content_index::{ + EXTRACTOR_VERSION, IndexDocRecord, TantivyContentIndex, +}; +use crate::infrastructure::services::search_index::text_extractor::{self, ExtractedText}; + +/// Queue rows drained per batch. Each row may cost a blob read + extraction, +/// so this is far smaller than the tree-etag drain batch. +const DRAIN_BATCH: i64 = 256; + +/// Max batches per tick so a huge backlog (initial reseed) cannot monopolise +/// the maintenance connection within one tick. +const MAX_BATCHES_PER_TICK: u32 = 4; + +/// Stored preview head per document (snippet source). +const PREVIEW_BYTES: usize = 16 * 1024; + +/// Ticks between `blob_extracted_text` orphan sweeps (~1 h at the default +/// 1.5 s interval). +const ORPHAN_SWEEP_TICKS: u64 = 2400; + +pub struct ContentIndexWorker { + maintenance_pool: Arc, + dedup: Arc, + index: Arc, + interval_ms: u64, + max_extract_file_bytes: u64, + max_text_bytes: usize, +} + +impl ContentIndexWorker { + pub fn new( + maintenance_pool: Arc, + dedup: Arc, + index: Arc, + interval_ms: u64, + max_extract_file_bytes: u64, + max_text_bytes: usize, + ) -> Self { + Self { + maintenance_pool, + dedup, + index, + // Floor the cadence so a misconfiguration can't busy-loop the + // maintenance pool. + interval_ms: interval_ms.max(200), + max_extract_file_bytes, + max_text_bytes, + } + } + + /// Spawn the indexing loop. Fire-and-forget: the loop logs and survives + /// every error (an exited loop would silently freeze the index while the + /// queue grows), and the first drain runs immediately to absorb rows left + /// over from a previous run or the migration backfill. + #[instrument(skip(self))] + pub fn start(self, needs_reseed: bool) { + info!( + "Starting content-index worker (every {}ms, batch {}, reseed: {})", + self.interval_ms, DRAIN_BATCH, needs_reseed + ); + tokio::spawn(async move { + if let Err(e) = self.prepare(needs_reseed).await { + error!("Content-index prepare failed (continuing with queue as-is): {e}"); + } + + let mut ticker = + tokio::time::interval(std::time::Duration::from_millis(self.interval_ms)); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut ticks: u64 = 0; + loop { + ticker.tick().await; + for _ in 0..MAX_BATCHES_PER_TICK { + match self.drain_once().await { + Ok(0) => break, + Ok(drained) => { + debug!("Content-index drain: processed {drained} queue row(s)"); + if drained < DRAIN_BATCH as usize { + break; + } + } + Err(e) => { + error!("Content-index drain failed (queue preserved, will retry): {e}"); + break; + } + } + } + + ticks += 1; + if ticks.is_multiple_of(ORPHAN_SWEEP_TICKS) { + self.sweep_orphaned_text().await; + } + } + }); + } + + /// Spawn the discard-only janitor used when content search is DISABLED: + /// the triggers are always installed, so something must keep the queue + /// from growing unboundedly. Re-enabling the feature reseeds from scratch + /// (index version marker), so discarding here loses nothing. + pub fn start_drain_only_janitor(maintenance_pool: Arc) { + info!("Content search disabled — starting queue janitor (discard-only)"); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60)); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + ticker.tick().await; + if let Err(e) = sqlx::query("DELETE FROM storage.search_index_dirty") + .execute(maintenance_pool.as_ref()) + .await + { + error!("Content-search queue janitor failed: {e}"); + } + } + }); + } + + /// Startup housekeeping: drop extraction rows from other extractor + /// versions (the reseed re-extracts them) and, when the on-disk index was + /// wiped, re-enqueue every live file. + async fn prepare(&self, needs_reseed: bool) -> Result<(), sqlx::Error> { + let dropped = sqlx::query("DELETE FROM storage.blob_extracted_text WHERE extractor <> $1") + .bind(EXTRACTOR_VERSION) + .execute(self.maintenance_pool.as_ref()) + .await? + .rows_affected(); + if dropped > 0 { + info!("Dropped {dropped} extraction row(s) from a previous extractor version"); + } + + if needs_reseed { + let queued = sqlx::query( + "INSERT INTO storage.search_index_dirty (file_id, op) + SELECT id, 'upsert' FROM storage.files WHERE NOT is_trashed", + ) + .execute(self.maintenance_pool.as_ref()) + .await? + .rows_affected(); + info!("Content-index reseed: queued {queued} file(s) for indexing"); + } + Ok(()) + } + + /// Drain and process one queue batch. Returns the number of queue rows + /// consumed (0 = queue empty). + async fn drain_once(&self) -> Result { + let rows: Vec<(i64, Uuid, String)> = sqlx::query_as( + "SELECT id, file_id, op FROM storage.search_index_dirty ORDER BY id LIMIT $1", + ) + .bind(DRAIN_BATCH) + .fetch_all(self.maintenance_pool.as_ref()) + .await?; + + if rows.is_empty() { + return Ok(0); + } + let drained_ids: Vec = rows.iter().map(|r| r.0).collect(); + + // Latest op per file wins (rows are id-ordered). + let mut latest_op: HashMap = HashMap::with_capacity(rows.len()); + for (_, file_id, op) in &rows { + latest_op.insert(*file_id, op == "upsert"); + } + let upsert_candidates: Vec = latest_op + .iter() + .filter_map(|(id, &upsert)| upsert.then_some(*id)) + .collect(); + let mut deletes: HashSet = latest_op + .iter() + .filter_map(|(id, &upsert)| (!upsert).then_some(*id)) + .collect(); + + // Authoritative state re-read: a queued 'upsert' whose row vanished + // or got trashed in the meantime becomes a delete. + let files: Vec<(Uuid, String, String, String, String, i64)> = + if upsert_candidates.is_empty() { + Vec::new() + } else { + sqlx::query_as( + "SELECT fi.id, fi.user_id::text, fi.name, fi.blob_hash, fi.mime_type, fi.size + FROM storage.files fi + WHERE fi.id = ANY($1) AND NOT fi.is_trashed", + ) + .bind(&upsert_candidates) + .fetch_all(self.maintenance_pool.as_ref()) + .await? + }; + let found: HashSet = files.iter().map(|f| f.0).collect(); + deletes.extend(upsert_candidates.iter().filter(|id| !found.contains(id))); + + // Per-blob text: batch-read the extraction cache, extract misses. + let wanted_hashes: Vec = files + .iter() + .filter(|(_, _, name, _, mime, size)| { + text_extractor::supports(name, mime) && *size as u64 <= self.max_extract_file_bytes + }) + .map(|f| f.3.clone()) + .collect(); + let mut text_by_hash: HashMap> = HashMap::new(); + if !wanted_hashes.is_empty() { + let cached: Vec<(String, Option, String)> = sqlx::query_as( + "SELECT blob_hash, text, status FROM storage.blob_extracted_text + WHERE blob_hash = ANY($1)", + ) + .bind(&wanted_hashes) + .fetch_all(self.maintenance_pool.as_ref()) + .await?; + for (hash, text, status) in cached { + text_by_hash.insert(hash, (status == "ok").then_some(text.unwrap_or_default())); + } + } + + let mut records = Vec::with_capacity(files.len()); + for (file_id, user_id, name, blob_hash, mime, size) in files { + let supported = text_extractor::supports(&name, &mime); + let content = if !supported { + None + } else if let Some(cached) = text_by_hash.get(&blob_hash) { + cached.clone() + } else { + let extracted = self + .extract_and_cache(&blob_hash, &name, &mime, size as u64) + .await; + text_by_hash.insert(blob_hash.clone(), extracted.clone()); + extracted + }; + + let preview = content + .as_deref() + .map(|t| truncate_on_char(t, PREVIEW_BYTES)); + records.push(IndexDocRecord { + file_id: file_id.to_string(), + user_id, + name, + content, + preview, + }); + } + + // One batched Tantivy commit, off the async runtime. + let index = self.index.clone(); + let delete_ids: Vec = deletes.iter().map(Uuid::to_string).collect(); + let applied: Result<(), String> = + match tokio::task::spawn_blocking(move || index.apply_batch(records, delete_ids)).await + { + Ok(Ok(())) => Ok(()), + Ok(Err(e)) => Err(e.to_string()), + Err(e) => Err(format!("join: {e}")), + }; + if let Err(e) = applied { + // Queue rows survive — the next tick retries the whole batch. + error!("Tantivy batch apply failed (will retry): {e}"); + return Ok(0); + } + + // Only now is the work durable in the index — drop the queue rows. + sqlx::query("DELETE FROM storage.search_index_dirty WHERE id = ANY($1)") + .bind(&drained_ids) + .execute(self.maintenance_pool.as_ref()) + .await?; + + Ok(drained_ids.len()) + } + + /// Read the blob (already size-capped), run the extractor on the blocking + /// pool, and persist the terminal outcome keyed by blob hash. Transient + /// read failures persist nothing — the next queue event retries. + async fn extract_and_cache( + &self, + blob_hash: &str, + name: &str, + mime: &str, + size: u64, + ) -> Option { + if size > self.max_extract_file_bytes { + self.store_extraction(blob_hash, None, "too_large").await; + return None; + } + + let bytes = match self.dedup.read_blob_bytes(blob_hash).await { + Ok(bytes) => bytes, + Err(e) => { + warn!( + "Content-index blob read failed for {blob_hash} (will retry on next event): {e}" + ); + return None; + } + }; + + let (name, mime, max_text) = (name.to_owned(), mime.to_owned(), self.max_text_bytes); + let outcome = tokio::task::spawn_blocking(move || { + text_extractor::extract(&name, &mime, &bytes, max_text) + }) + .await + .unwrap_or_else(|e| ExtractedText::Failed(format!("join: {e}"))); + + match outcome { + ExtractedText::Text(text) => { + self.store_extraction(blob_hash, Some(&text), "ok").await; + Some(text) + } + ExtractedText::Empty => { + self.store_extraction(blob_hash, None, "empty").await; + None + } + ExtractedText::Failed(reason) => { + warn!("Text extraction failed for blob {blob_hash}: {reason}"); + self.store_extraction(blob_hash, None, "failed").await; + None + } + ExtractedText::Unsupported => None, + } + } + + async fn store_extraction(&self, blob_hash: &str, text: Option<&str>, status: &str) { + if let Err(e) = sqlx::query( + "INSERT INTO storage.blob_extracted_text (blob_hash, text, status, extractor) + VALUES ($1, $2, $3, $4) + ON CONFLICT (blob_hash) DO NOTHING", + ) + .bind(blob_hash) + .bind(text) + .bind(status) + .bind(EXTRACTOR_VERSION) + .execute(self.maintenance_pool.as_ref()) + .await + { + warn!("Failed to cache extraction for blob {blob_hash}: {e}"); + } + } + + /// Drop extraction rows whose blob no longer backs any live file. Uses + /// the `idx_files_blob_hash` index; runs hourly on the maintenance pool. + async fn sweep_orphaned_text(&self) { + match sqlx::query( + "DELETE FROM storage.blob_extracted_text bet + WHERE NOT EXISTS (SELECT 1 FROM storage.files f WHERE f.blob_hash = bet.blob_hash)", + ) + .execute(self.maintenance_pool.as_ref()) + .await + { + Ok(result) if result.rows_affected() > 0 => { + debug!( + "Content-index sweep: dropped {} orphaned extraction row(s)", + result.rows_affected() + ); + } + Ok(_) => {} + Err(e) => error!("Content-index orphan sweep failed: {e}"), + } + } +} + +/// Truncate on a char boundary at most `max_bytes` into `s`. +fn truncate_on_char(s: &str, max_bytes: usize) -> String { + if s.len() <= max_bytes { + return s.to_owned(); + } + let mut end = max_bytes; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + s[..end].to_owned() +} + +#[cfg(test)] +mod tests { + use super::truncate_on_char; + + #[test] + fn truncates_on_char_boundary() { + assert_eq!(truncate_on_char("patatas", 4), "pata"); + // 'ñ' is 2 bytes — a cut landing inside it must back off ("ñoño" is + // ñ:0-1 o:2 ñ:3-4 o:5, so a 4-byte cut falls mid-ñ and yields "ño"). + assert_eq!(truncate_on_char("ñoño", 4), "ño"); + assert_eq!(truncate_on_char("ok", 10), "ok"); + } +} diff --git a/src/infrastructure/services/search_index/mod.rs b/src/infrastructure/services/search_index/mod.rs new file mode 100644 index 00000000..78276f12 --- /dev/null +++ b/src/infrastructure/services/search_index/mod.rs @@ -0,0 +1,19 @@ +//! Embedded full-text content index (Tantivy) and its feeding pipeline. +//! +//! Three pieces, mirroring the thumbnail/tree-etag architecture: +//! +//! * [`tantivy_content_index`] — the embedded BM25 index over file names and +//! extracted content. Lives on local disk (`{storage}/.search-index`), +//! single-writer, microsecond queries. A DERIVED artifact: PostgreSQL is +//! the source of truth and the index is rebuilt (reseeded) whenever its +//! on-disk schema version differs from the binary's. +//! * [`text_extractor`] — pure-Rust text extraction (plain text/code, PDF, +//! Office OOXML/ODF). CPU-bound, runs only on the background worker. +//! * [`content_index_worker`] — drains `storage.search_index_dirty` (fed by +//! statement triggers on `storage.files`), extracts text once per unique +//! blob (BLAKE3-keyed cache in `storage.blob_extracted_text`), and applies +//! batched Tantivy mutations. Never touches a request path. + +pub mod content_index_worker; +pub mod tantivy_content_index; +pub mod text_extractor; diff --git a/src/infrastructure/services/search_index/tantivy_content_index.rs b/src/infrastructure/services/search_index/tantivy_content_index.rs new file mode 100644 index 00000000..fcfbcd3b --- /dev/null +++ b/src/infrastructure/services/search_index/tantivy_content_index.rs @@ -0,0 +1,545 @@ +//! Embedded Tantivy index over file names + extracted content. +//! +//! Performance contract (the reason Tantivy was chosen): +//! * queries are memory-mapped posting-list lookups — µs to low ms even at +//! millions of documents, executed on the blocking pool (never stalls the +//! Tokio reactor); +//! * the single `IndexWriter` is owned by the background worker; request +//! paths only ever touch the lock-free `IndexReader`. +//! +//! Index layout: one document per live file. +//! * `file_id` — raw term, stored. Identity for upsert (delete_term + add). +//! * `user_id` — raw term. Every query is `Must`-filtered by it, and hits +//! are re-validated through SQL hydration afterwards (defense in depth). +//! * `name` — tokenized file name (boosted 3x at query time). +//! * `content` — tokenized extracted text (never stored — the index stays +//! small; snippets come from `preview`). +//! * `preview` — stored-only head of the extracted text used to render a +//! snippet around the first matched term. + +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use tantivy::collector::TopDocs; +use tantivy::directory::MmapDirectory; +use tantivy::query::{BooleanQuery, BoostQuery, FuzzyTermQuery, Occur, Query, TermQuery}; +use tantivy::schema::{Field, IndexRecordOption, STORED, STRING, Schema, TEXT, Value as _}; +use tantivy::snippet::SnippetGenerator; +use tantivy::tokenizer::TextAnalyzer; +use tantivy::{Index, IndexReader, IndexWriter, ReloadPolicy, TantivyDocument, Term, doc}; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::application::ports::content_index_ports::{ContentHitDto, ContentIndexPort}; +use crate::common::errors::DomainError; + +/// Bump whenever the Tantivy schema OR the text extractor output changes in a +/// way that requires re-indexing. A mismatch with the on-disk marker wipes the +/// index directory and reseeds the dirty queue with every live file. +pub const INDEX_SCHEMA_VERSION: &str = "1"; + +/// Recorded in `storage.blob_extracted_text.extractor`; rows from another +/// version are dropped at worker startup (the reseed re-extracts them). +/// Keep in lockstep with [`INDEX_SCHEMA_VERSION`]. +pub const EXTRACTOR_VERSION: &str = "rust-native-1"; + +/// Marker file inside the index directory carrying the schema version. +const META_FILE: &str = "oxicloud-index.version"; + +/// RAM budget for the single-threaded writer. Indexing is a trickle-feed +/// background task — one thread and a small heap keep the footprint +/// negligible next to the request-serving process. +const WRITER_HEAP_BYTES: usize = 64 * 1024 * 1024; + +/// Hard cap on query tokens — a pathological query must not fan out into +/// dozens of fuzzy automata. +const MAX_QUERY_TOKENS: usize = 8; + +/// Snippet length target, in characters. +const SNIPPET_MAX_CHARS: usize = 180; + +/// Minimum token length for typo-tolerant (edit distance 1) matching. +/// Short tokens produce too many false positives under fuzzy matching. +const FUZZY_MIN_CHARS: usize = 5; + +/// Minimum token length for prefix expansion of the LAST query token +/// (search-as-you-type behaviour). +const PREFIX_MIN_CHARS: usize = 3; + +/// One file to (re-)index. `content`/`preview` are `None` for files without +/// extractable text (images, archives…) — their NAME is still indexed. +#[derive(Debug)] +pub struct IndexDocRecord { + pub file_id: String, + pub user_id: String, + pub name: String, + pub content: Option, + pub preview: Option, +} + +#[derive(Clone, Copy)] +struct IndexFields { + file_id: Field, + user_id: Field, + name: Field, + content: Field, + preview: Field, +} + +pub struct TantivyContentIndex { + /// Sole writer — owned by the background worker; the Mutex is never + /// contended on a request path. (Writer and reader each keep the + /// underlying `Index` alive.) + writer: Mutex, + reader: IndexReader, + /// Pre-cloned analyzer for query-side tokenization (matches the index + /// side: simple split + lowercase). + analyzer: TextAnalyzer, + fields: IndexFields, +} + +impl TantivyContentIndex { + fn build_schema() -> (Schema, IndexFields) { + let mut builder = Schema::builder(); + let fields = IndexFields { + file_id: builder.add_text_field("file_id", STRING | STORED), + user_id: builder.add_text_field("user_id", STRING), + name: builder.add_text_field("name", TEXT), + content: builder.add_text_field("content", TEXT), + preview: builder.add_text_field("preview", STORED), + }; + (builder.build(), fields) + } + + /// Open the index at `dir`, wiping and recreating it when the on-disk + /// version marker is absent or stale. Returns `(index, needs_reseed)`: + /// when `needs_reseed` is true the caller must re-enqueue every live file. + pub fn open_or_rebuild(dir: &Path) -> Result<(Self, bool), DomainError> { + let marker: PathBuf = dir.join(META_FILE); + let version_ok = std::fs::read_to_string(&marker) + .map(|v| v.trim() == INDEX_SCHEMA_VERSION) + .unwrap_or(false); + + if !version_ok && dir.exists() { + std::fs::remove_dir_all(dir).map_err(|e| { + DomainError::internal_error( + "ContentIndex", + format!("wiping stale index dir {}: {e}", dir.display()), + ) + })?; + } + std::fs::create_dir_all(dir).map_err(|e| { + DomainError::internal_error( + "ContentIndex", + format!("creating index dir {}: {e}", dir.display()), + ) + })?; + + let (schema, fields) = Self::build_schema(); + let mmap = MmapDirectory::open(dir) + .map_err(|e| DomainError::internal_error("ContentIndex", format!("mmap dir: {e}")))?; + let index = Index::open_or_create(mmap, schema) + .map_err(|e| DomainError::internal_error("ContentIndex", format!("open: {e}")))?; + + // Single writer thread: indexing is a background trickle, not a bulk + // load — keep the CPU/RAM footprint minimal. + let writer = index + .writer_with_num_threads::(1, WRITER_HEAP_BYTES) + .map_err(|e| DomainError::internal_error("ContentIndex", format!("writer: {e}")))?; + + let reader = index + .reader_builder() + .reload_policy(ReloadPolicy::OnCommitWithDelay) + .try_into() + .map_err(|e| DomainError::internal_error("ContentIndex", format!("reader: {e}")))?; + + let analyzer = index + .tokenizer_for_field(fields.content) + .map_err(|e| DomainError::internal_error("ContentIndex", format!("analyzer: {e}")))?; + + std::fs::write(&marker, INDEX_SCHEMA_VERSION).map_err(|e| { + DomainError::internal_error("ContentIndex", format!("writing version marker: {e}")) + })?; + + Ok(( + Self { + writer: Mutex::new(writer), + reader, + analyzer, + fields, + }, + !version_ok, + )) + } + + /// Apply one drained queue batch: deletes, then upserts, then ONE commit. + /// Blocking (disk I/O + segment serialization) — call from the worker via + /// `spawn_blocking`. The caller deletes the queue rows only after this + /// returns `Ok`, so a crash in between re-processes the batch + /// (idempotent: upsert = delete_term + add). + pub fn apply_batch( + &self, + upserts: Vec, + deletes: Vec, + ) -> Result<(), DomainError> { + let mut writer = self + .writer + .lock() + .map_err(|_| DomainError::internal_error("ContentIndex", "writer mutex poisoned"))?; + + for file_id in &deletes { + writer.delete_term(Term::from_field_text(self.fields.file_id, file_id)); + } + + for record in upserts { + writer.delete_term(Term::from_field_text(self.fields.file_id, &record.file_id)); + let mut document = doc!( + self.fields.file_id => record.file_id, + self.fields.user_id => record.user_id, + self.fields.name => record.name, + ); + if let Some(content) = record.content { + document.add_text(self.fields.content, content); + } + if let Some(preview) = record.preview { + document.add_text(self.fields.preview, preview); + } + writer + .add_document(document) + .map_err(|e| DomainError::internal_error("ContentIndex", format!("add: {e}")))?; + } + + writer + .commit() + .map_err(|e| DomainError::internal_error("ContentIndex", format!("commit: {e}")))?; + Ok(()) + } + + /// Number of live documents — used by tests and the startup log line. + pub fn num_docs(&self) -> u64 { + self.reader.searcher().num_docs() + } + + /// Tokenize `raw` with the index analyzer (simple split + lowercase). + fn query_tokens(analyzer: &TextAnalyzer, raw: &str) -> Vec { + let mut analyzer = analyzer.clone(); + let mut tokens = Vec::new(); + let mut stream = analyzer.token_stream(raw); + while stream.advance() && tokens.len() < MAX_QUERY_TOKENS { + tokens.push(stream.token().text.clone()); + } + tokens + } + + /// Build the scored query: every token must match (in name OR content, + /// exact OR fuzzy OR — for the last token — prefix), and the whole thing + /// is `Must`-scoped to the user. + fn build_query(fields: IndexFields, user_id: &str, tokens: &[String]) -> Box { + let mut clauses: Vec<(Occur, Box)> = vec![( + Occur::Must, + Box::new(TermQuery::new( + Term::from_field_text(fields.user_id, user_id), + IndexRecordOption::Basic, + )), + )]; + + let last = tokens.len().saturating_sub(1); + for (i, token) in tokens.iter().enumerate() { + let name_term = Term::from_field_text(fields.name, token); + let content_term = Term::from_field_text(fields.content, token); + + let mut alternatives: Vec<(Occur, Box)> = vec![ + ( + Occur::Should, + // Name matches outrank content matches for the same term. + Box::new(BoostQuery::new( + Box::new(TermQuery::new( + name_term.clone(), + IndexRecordOption::WithFreqs, + )), + 3.0, + )), + ), + ( + Occur::Should, + Box::new(TermQuery::new( + content_term.clone(), + IndexRecordOption::WithFreqs, + )), + ), + ]; + + if token.chars().count() >= FUZZY_MIN_CHARS { + // Edit distance 1 absorbs typos and most singular/plural + // morphology ("patata" ↔ "patatas") without a stemmer. + alternatives.push(( + Occur::Should, + Box::new(FuzzyTermQuery::new(name_term.clone(), 1, true)), + )); + alternatives.push(( + Occur::Should, + Box::new(FuzzyTermQuery::new(content_term.clone(), 1, true)), + )); + } + if i == last && token.chars().count() >= PREFIX_MIN_CHARS { + // Search-as-you-type: the token still being typed matches as + // a prefix ("pata" → "patatas"). + alternatives.push(( + Occur::Should, + Box::new(FuzzyTermQuery::new_prefix(name_term, 0, true)), + )); + alternatives.push(( + Occur::Should, + Box::new(FuzzyTermQuery::new_prefix(content_term, 0, true)), + )); + } + + clauses.push((Occur::Must, Box::new(BooleanQuery::new(alternatives)))); + } + + Box::new(BooleanQuery::new(clauses)) + } + + /// Blocking search core — runs on the blocking pool via the port impl. + fn search_blocking( + searcher: tantivy::Searcher, + analyzer: TextAnalyzer, + fields: IndexFields, + user_id: &str, + raw_query: &str, + limit: usize, + ) -> Result, DomainError> { + let tokens = Self::query_tokens(&analyzer, raw_query); + if tokens.is_empty() { + return Ok(Vec::new()); + } + + let query = Self::build_query(fields, user_id, &tokens); + let top_docs = searcher + .search(&query, &TopDocs::with_limit(limit.max(1)).order_by_score()) + .map_err(|e| DomainError::internal_error("ContentIndex", format!("search: {e}")))?; + + // Snippets highlight CONTENT matches; an empty fragment means the hit + // came from the name (or a fuzzy variant) — no snippet then. + let snippet_generator = SnippetGenerator::create(&searcher, &*query, fields.content) + .map(|mut g| { + g.set_max_num_chars(SNIPPET_MAX_CHARS); + g + }) + .ok(); + + let mut hits = Vec::with_capacity(top_docs.len()); + for (score, address) in top_docs { + let document: TantivyDocument = searcher.doc(address).map_err(|e| { + DomainError::internal_error("ContentIndex", format!("doc fetch: {e}")) + })?; + let Some(file_id) = document + .get_first(fields.file_id) + .and_then(|v| v.as_str()) + .map(str::to_owned) + else { + continue; + }; + + let snippet = document + .get_first(fields.preview) + .and_then(|v| v.as_str()) + .and_then(|preview| { + let generator = snippet_generator.as_ref()?; + let fragment = generator.snippet(preview).fragment().trim().to_owned(); + (!fragment.is_empty()).then_some(fragment) + }); + + hits.push(ContentHitDto { + file_id, + score, + snippet, + }); + } + Ok(hits) + } +} + +#[async_trait] +impl ContentIndexPort for TantivyContentIndex { + async fn search_content( + &self, + user_id: Uuid, + query: &str, + limit: usize, + ) -> Result, DomainError> { + let searcher = self.reader.searcher(); + let analyzer = self.analyzer.clone(); + let fields = self.fields; + let user_id = user_id.to_string(); + let query = query.to_owned(); + + tokio::task::spawn_blocking(move || { + Self::search_blocking(searcher, analyzer, fields, &user_id, &query, limit) + }) + .await + .map_err(|e| DomainError::internal_error("ContentIndex", format!("join: {e}")))? + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn record(file_id: &str, user_id: &str, name: &str, content: Option<&str>) -> IndexDocRecord { + IndexDocRecord { + file_id: file_id.to_owned(), + user_id: user_id.to_owned(), + name: name.to_owned(), + content: content.map(str::to_owned), + preview: content.map(str::to_owned), + } + } + + fn search(index: &TantivyContentIndex, user_id: &str, query: &str) -> Vec { + // Force a reader reload — OnCommitWithDelay is asynchronous and tests + // must observe the commit immediately. + index.reader.reload().unwrap(); + TantivyContentIndex::search_blocking( + index.reader.searcher(), + index.analyzer.clone(), + index.fields, + user_id, + query, + 32, + ) + .unwrap() + } + + #[test] + fn index_name_and_content_with_fuzzy_prefix_and_user_isolation() { + let dir = tempfile::tempdir().unwrap(); + let (index, needs_reseed) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap(); + assert!(needs_reseed, "fresh dir must request a reseed"); + + index + .apply_batch( + vec![ + record("f1", "user-a", "patatas-fritas.jpg", None), + record( + "f2", + "user-a", + "recetas.pdf", + Some("la mejor receta de patatas bravas del mundo"), + ), + record( + "f3", + "user-b", + "patatas-ajenas.txt", + Some("patatas de otro usuario"), + ), + record("f4", "user-a", "informe.txt", Some("nada relacionado aqui")), + ], + Vec::new(), + ) + .unwrap(); + + // Exact term: name hit + content hit for user-a only. + let hits = search(&index, "user-a", "patatas"); + let ids: Vec<&str> = hits.iter().map(|h| h.file_id.as_str()).collect(); + assert!(ids.contains(&"f1"), "name match expected: {ids:?}"); + assert!(ids.contains(&"f2"), "content match expected: {ids:?}"); + assert!(!ids.contains(&"f3"), "other user's file leaked: {ids:?}"); + assert!(!ids.contains(&"f4"), "non-matching file returned: {ids:?}"); + + // The content hit carries a snippet around the matched term. + let content_hit = hits.iter().find(|h| h.file_id == "f2").unwrap(); + assert!( + content_hit + .snippet + .as_deref() + .unwrap_or("") + .contains("patatas"), + "snippet should surround the match: {:?}", + content_hit.snippet + ); + + // Fuzzy (distance 1): singular finds plural. + let ids: Vec = search(&index, "user-a", "patata") + .into_iter() + .map(|h| h.file_id) + .collect(); + assert!( + ids.contains(&"f2".to_owned()), + "fuzzy match expected: {ids:?}" + ); + + // Prefix on the last token (search-as-you-type). + let ids: Vec = search(&index, "user-a", "pata") + .into_iter() + .map(|h| h.file_id) + .collect(); + assert!( + ids.contains(&"f1".to_owned()), + "prefix match expected: {ids:?}" + ); + } + + #[test] + fn upsert_replaces_and_delete_removes() { + let dir = tempfile::tempdir().unwrap(); + let (index, _) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap(); + + index + .apply_batch( + vec![record( + "f1", + "u", + "old-name.txt", + Some("contenido original"), + )], + Vec::new(), + ) + .unwrap(); + index + .apply_batch( + vec![record("f1", "u", "renamed.txt", Some("contenido original"))], + Vec::new(), + ) + .unwrap(); + + assert!( + search(&index, "u", "old").is_empty(), + "stale doc survived upsert" + ); + assert_eq!(search(&index, "u", "renamed").len(), 1); + + index + .apply_batch(Vec::new(), vec!["f1".to_owned()]) + .unwrap(); + assert!( + search(&index, "u", "renamed").is_empty(), + "deleted doc still found" + ); + } + + #[test] + fn reopen_preserves_documents_and_version_mismatch_wipes() { + let dir = tempfile::tempdir().unwrap(); + { + let (index, _) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap(); + index + .apply_batch(vec![record("f1", "u", "persistente.txt", None)], Vec::new()) + .unwrap(); + } + + // Same version: documents survive, no reseed requested. + { + let (index, needs_reseed) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap(); + assert!(!needs_reseed); + assert_eq!(index.num_docs(), 1); + } + + // Stale version marker: wipe + reseed. + std::fs::write(dir.path().join(META_FILE), "0-stale").unwrap(); + let (index, needs_reseed) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap(); + assert!(needs_reseed); + assert_eq!(index.num_docs(), 0); + } +} diff --git a/src/infrastructure/services/search_index/text_extractor.rs b/src/infrastructure/services/search_index/text_extractor.rs new file mode 100644 index 00000000..470986aa --- /dev/null +++ b/src/infrastructure/services/search_index/text_extractor.rs @@ -0,0 +1,418 @@ +//! Pure-Rust text extraction for the content index. +//! +//! Supported: plain text/code/markup, PDF (text layer), Office OOXML +//! (docx/xlsx/pptx) and OpenDocument (odt/ods/odp). Images, media and +//! archives are reported as [`ExtractedText::Unsupported`] WITHOUT reading +//! the blob (the worker checks [`supports`] first). +//! +//! Everything here is CPU-bound and synchronous — the worker runs it inside +//! `spawn_blocking`, one extraction at a time, so user-facing latency is +//! never affected. Output is whitespace-normalized and hard-capped at the +//! caller-provided byte budget. + +use std::io::{BufReader, Cursor}; +use std::panic::{AssertUnwindSafe, catch_unwind}; + +use quick_xml::events::Event; + +/// Outcome of one extraction attempt. +#[derive(Debug)] +pub enum ExtractedText { + /// Usable text (normalized, capped). + Text(String), + /// Extractor ran fine but produced no text (e.g. empty document, + /// scanned PDF without a text layer, binary masquerading as text). + Empty, + /// No extractor handles this name/MIME combination. + Unsupported, + /// Extractor failed or panicked — terminal for this blob (recorded so it + /// is never retried until the extractor version bumps). + Failed(String), +} + +#[derive(Debug, Clone, Copy, PartialEq)] +enum Kind { + Plain, + Pdf, + Docx, + Xlsx, + Pptx, + Odf, +} + +/// MIME types (beyond `text/*`) parsed as plain text. +const TEXTUAL_MIMES: &[&str] = &[ + "application/json", + "application/ld+json", + "application/xml", + "application/javascript", + "application/x-javascript", + "application/x-yaml", + "application/yaml", + "application/toml", + "application/x-sh", + "application/x-shellscript", + "application/sql", + "image/svg+xml", +]; + +/// Extensions parsed as plain text when the MIME type is generic +/// (`application/octet-stream` uploads are common on WebDAV clients). +const TEXTUAL_EXTENSIONS: &[&str] = &[ + "txt", "md", "markdown", "csv", "tsv", "json", "xml", "yaml", "yml", "toml", "ini", "cfg", + "conf", "log", "rs", "js", "mjs", "ts", "jsx", "tsx", "css", "scss", "html", "htm", "py", "rb", + "go", "java", "c", "h", "cpp", "hpp", "cs", "php", "sh", "sql", "tex", "svg", +]; + +fn extension_of(name: &str) -> Option { + name.rsplit_once('.').map(|(_, ext)| ext.to_lowercase()) +} + +fn classify(name: &str, mime: &str) -> Option { + let mime = mime + .split(';') + .next() + .unwrap_or_default() + .trim() + .to_lowercase(); + + if mime.starts_with("text/") || TEXTUAL_MIMES.contains(&mime.as_str()) { + return Some(Kind::Plain); + } + match mime.as_str() { + "application/pdf" => return Some(Kind::Pdf), + "application/vnd.openxmlformats-officedocument.wordprocessingml.document" => { + return Some(Kind::Docx); + } + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => { + return Some(Kind::Xlsx); + } + "application/vnd.openxmlformats-officedocument.presentationml.presentation" => { + return Some(Kind::Pptx); + } + "application/vnd.oasis.opendocument.text" + | "application/vnd.oasis.opendocument.spreadsheet" + | "application/vnd.oasis.opendocument.presentation" => return Some(Kind::Odf), + _ => {} + } + + // Generic MIME — fall back to the extension. + match extension_of(name)?.as_str() { + ext if TEXTUAL_EXTENSIONS.contains(&ext) => Some(Kind::Plain), + "pdf" => Some(Kind::Pdf), + "docx" => Some(Kind::Docx), + "xlsx" => Some(Kind::Xlsx), + "pptx" => Some(Kind::Pptx), + "odt" | "ods" | "odp" => Some(Kind::Odf), + _ => None, + } +} + +/// Whether [`extract`] has an extractor for this file — the worker calls this +/// BEFORE reading the blob, so unsupported content (photos, video, archives) +/// costs zero I/O. +pub fn supports(name: &str, mime: &str) -> bool { + classify(name, mime).is_some() +} + +/// Extract plain text from `bytes`, capped at `max_text_bytes` of UTF-8. +pub fn extract(name: &str, mime: &str, bytes: &[u8], max_text_bytes: usize) -> ExtractedText { + let Some(kind) = classify(name, mime) else { + return ExtractedText::Unsupported; + }; + + let result = match kind { + Kind::Plain => extract_plain(bytes, max_text_bytes), + Kind::Pdf => extract_pdf(bytes, max_text_bytes), + Kind::Docx => { + extract_zipped_xml(bytes, ZipSource::Fixed("word/document.xml"), max_text_bytes) + } + Kind::Xlsx => extract_zipped_xml( + bytes, + ZipSource::Fixed("xl/sharedStrings.xml"), + max_text_bytes, + ), + Kind::Pptx => extract_zipped_xml(bytes, ZipSource::Slides, max_text_bytes), + Kind::Odf => extract_zipped_xml(bytes, ZipSource::Fixed("content.xml"), max_text_bytes), + }; + + match result { + Ok(text) if text.is_empty() => ExtractedText::Empty, + Ok(text) => ExtractedText::Text(text), + Err(reason) => ExtractedText::Failed(reason), + } +} + +/// Collapse whitespace runs and cap at `max_bytes` (on a char boundary). +/// Normalization keeps the index lean and makes stored previews readable. +fn normalize_and_cap(text: &str, max_bytes: usize) -> String { + let mut out = String::with_capacity(text.len().min(max_bytes)); + for word in text.split_whitespace() { + if out.len() + word.len() + 1 > max_bytes { + break; + } + if !out.is_empty() { + out.push(' '); + } + out.push_str(word); + } + out +} + +fn extract_plain(bytes: &[u8], max_text_bytes: usize) -> Result { + // NUL byte in the head = binary masquerading under a textual name/MIME. + if bytes.iter().take(8192).any(|&b| b == 0) { + return Ok(String::new()); + } + // Decode at most ~2x the budget — normalization only shrinks text, so + // anything beyond that can never reach the output. + let slice_end = bytes.len().min(max_text_bytes.saturating_mul(2)); + let text = String::from_utf8_lossy(&bytes[..slice_end]); + Ok(normalize_and_cap(&text, max_text_bytes)) +} + +fn extract_pdf(bytes: &[u8], max_text_bytes: usize) -> Result { + // pdf-extract is known to panic on malformed documents; a poisoned blob + // must mark itself 'failed' instead of taking the worker down. + let outcome = catch_unwind(AssertUnwindSafe(|| { + pdf_extract::extract_text_from_mem(bytes) + })); + match outcome { + Ok(Ok(text)) => Ok(normalize_and_cap(&text, max_text_bytes)), + Ok(Err(e)) => Err(format!("pdf: {e}")), + Err(_) => Err("pdf: extractor panicked".to_owned()), + } +} + +enum ZipSource { + /// One well-known entry (docx body, xlsx shared strings, ODF content). + Fixed(&'static str), + /// Every `ppt/slides/slideN.xml` entry. + Slides, +} + +fn extract_zipped_xml( + bytes: &[u8], + source: ZipSource, + max_text_bytes: usize, +) -> Result { + let mut archive = zip::ZipArchive::new(Cursor::new(bytes)).map_err(|e| format!("zip: {e}"))?; + + let entries: Vec = match source { + ZipSource::Fixed(name) => vec![name.to_owned()], + ZipSource::Slides => { + let mut slides: Vec = archive + .file_names() + .filter(|n| n.starts_with("ppt/slides/slide") && n.ends_with(".xml")) + .map(str::to_owned) + .collect(); + slides.sort(); + slides + } + }; + + let mut text = String::new(); + for entry in entries { + let Ok(file) = archive.by_name(&entry) else { + // Tolerated: e.g. an xlsx with no shared strings table. + continue; + }; + collect_xml_text(BufReader::new(file), &mut text, max_text_bytes) + .map_err(|e| format!("{entry}: {e}"))?; + if text.len() >= max_text_bytes { + break; + } + } + Ok(normalize_and_cap(&text, max_text_bytes)) +} + +/// Append every XML text node to `out` (capped). Text RUNS are concatenated +/// without separators — OOXML splits words across `` runs arbitrarily — +/// while paragraph/cell boundaries insert whitespace so distinct words never +/// fuse together. +fn collect_xml_text( + reader: R, + out: &mut String, + max_bytes: usize, +) -> Result<(), String> { + let mut xml = quick_xml::Reader::from_reader(reader); + let mut buf = Vec::new(); + loop { + if out.len() >= max_bytes { + return Ok(()); + } + match xml.read_event_into(&mut buf) { + Ok(Event::Text(t)) => { + if let Ok(decoded) = t.xml_content() { + out.push_str(&decoded); + } + } + Ok(Event::GeneralRef(r)) => { + // quick-xml emits entity references as separate events. + // Character refs (A) and the predefined five resolve to + // their literal character; unknown custom entities are + // dropped (no DTD resolution). + if let Ok(Some(ch)) = r.resolve_char_ref() { + out.push(ch); + } else if let Ok(name) = r.decode() { + match name.as_ref() { + "amp" => out.push('&'), + "lt" => out.push('<'), + "gt" => out.push('>'), + "apos" => out.push('\''), + "quot" => out.push('"'), + _ => {} + } + } + } + Ok(Event::End(e)) => { + // Paragraphs (docx w:p, pptx a:p, ODF text:p/text:h), table + // rows and xlsx shared-string items all separate words. + let local = e.local_name(); + if matches!(local.as_ref(), b"p" | b"h" | b"si" | b"row" | b"br") { + out.push('\n'); + } + } + Ok(Event::Empty(e)) => { + // Self-closing breaks/tabs inside a paragraph (, ). + let local = e.local_name(); + if matches!(local.as_ref(), b"br" | b"tab") { + out.push(' '); + } + } + Ok(Event::Eof) => return Ok(()), + Ok(_) => {} + Err(e) => return Err(format!("xml: {e}")), + } + buf.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use zip::write::SimpleFileOptions; + + fn build_zip(entries: &[(&str, &str)]) -> Vec { + let mut writer = zip::ZipWriter::new(Cursor::new(Vec::new())); + for (name, content) in entries { + writer + .start_file(*name, SimpleFileOptions::default()) + .unwrap(); + writer.write_all(content.as_bytes()).unwrap(); + } + writer.finish().unwrap().into_inner() + } + + fn text_of(outcome: ExtractedText) -> String { + match outcome { + ExtractedText::Text(t) => t, + other => panic!("expected Text, got {other:?}"), + } + } + + #[test] + fn plain_text_is_normalized_and_capped() { + let out = extract( + "notas.txt", + "text/plain", + b"receta de\n\npatatas bravas", + 1024, + ); + assert_eq!(text_of(out), "receta de patatas bravas"); + + let big = "palabra ".repeat(1000); + let out = text_of(extract("big.txt", "text/plain", big.as_bytes(), 64)); + assert!(out.len() <= 64, "cap exceeded: {}", out.len()); + assert!(out.ends_with("palabra"), "must cut on word boundary"); + } + + #[test] + fn binary_masquerading_as_text_yields_empty() { + let mut bytes = b"PK\x03\x04".to_vec(); + bytes.extend_from_slice(&[0u8; 64]); + assert!(matches!( + extract("raro.txt", "text/plain", &bytes, 1024), + ExtractedText::Empty + )); + } + + #[test] + fn unsupported_types_are_reported_without_reading() { + assert!(!supports("foto.jpg", "image/jpeg")); + assert!(matches!( + extract("foto.jpg", "image/jpeg", &[0xFF, 0xD8], 1024), + ExtractedText::Unsupported + )); + assert!(supports("recetas.pdf", "application/pdf")); + assert!( + supports("notas", "text/plain"), + "MIME wins without extension" + ); + } + + #[test] + fn docx_runs_concatenate_and_paragraphs_separate() { + let body = r#" + + + patatas + bravas & ali oli + + "#; + let bytes = build_zip(&[("word/document.xml", body)]); + let out = text_of(extract( + "receta.docx", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + &bytes, + 4096, + )); + assert_eq!(out, "patatas bravas & ali oli"); + } + + #[test] + fn xlsx_shared_strings_extract() { + let shared = r#" + + patatas + 900 kg + "#; + let bytes = build_zip(&[("xl/sharedStrings.xml", shared)]); + let out = text_of(extract( + "stock.xlsx", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + &bytes, + 4096, + )); + assert_eq!(out, "patatas 900 kg"); + } + + #[test] + fn odt_content_extracts_by_extension_fallback() { + let content = r#" + + + tortilla de patatas + + "#; + let bytes = build_zip(&[("content.xml", content)]); + // Generic MIME — classification must fall back to the .odt extension. + let out = text_of(extract( + "receta.odt", + "application/octet-stream", + &bytes, + 4096, + )); + assert_eq!(out, "tortilla de patatas"); + } + + #[test] + fn corrupt_pdf_fails_terminally_instead_of_panicking() { + assert!(matches!( + extract("roto.pdf", "application/pdf", b"definitely not a pdf", 4096), + ExtractedText::Failed(_) + )); + } +} diff --git a/static/css/components/resourceList.css b/static/css/components/resourceList.css index 9a907da0..e462e184 100644 --- a/static/css/components/resourceList.css +++ b/static/css/components/resourceList.css @@ -269,6 +269,26 @@ text-overflow: ellipsis; } +/* Content-search snippet — fragment of the file body that matched the + query. Wraps onto a second line inside the flex name-cell. */ +.files-list-view .file-item .name-cell:has(.file-item__snippet) { + flex-wrap: wrap; + row-gap: 2px; +} + +.files-list-view .file-item .file-item__snippet { + flex-basis: 100%; + /* icon width (36px) + cell gap (12px) — aligns under the name */ + padding-left: 48px; + font-size: 12px; + color: var(--color-text-muted); +} + +/* Grid cards are too compact for body fragments. */ +.files-grid-view .file-item .file-item__snippet { + display: none; +} + .files-list-view .file-item .file-icon { width: 36px; height: 36px; diff --git a/static/js/components/resourceList.js b/static/js/components/resourceList.js index 63c82380..916a9b39 100644 --- a/static/js/components/resourceList.js +++ b/static/js/components/resourceList.js @@ -739,6 +739,7 @@ export class ResourceListComponent { ${escapeHtml(file.name)} ${cfg.showFavorite ? `
` : ''} ${cfg.showShareBadge ? `
` : ''} + ${file.snippet ? `${escapeHtml(file.snippet)}` : ''}
${cfg.showPath ? `
${escapeHtml(file.path || '')}
` : ''} diff --git a/static/js/core/types.js b/static/js/core/types.js index 99ba6f02..143b75ee 100644 --- a/static/js/core/types.js +++ b/static/js/core/types.js @@ -47,6 +47,8 @@ * @property {number} sort_date * @property {string} etag opaque HTTP ETag, for If-Match / If-None-Match * @property {string} content_hash raw BLAKE3 content hash, for dedup checks + * @property {string} [snippet] plain-text fragment around a content match (search results only) + * @property {string} [match_source] "name" or "content" — how the search found this file */ /**