From 77cbb62d84dd0357303b13739cf11c9a1fed14a5 Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Mon, 10 Aug 2026 01:44:40 +0100 Subject: [PATCH] Added similarity search --- .gitignore | 2 + Cargo.lock | 619 ++++++++++++++- Cargo.toml | 10 +- flake.lock | 27 + flake.nix | 38 + src/admin/mod.rs | 14 + src/admin/v2.rs | 119 +++ src/config.rs | 37 + src/federation/capabilities.rs | 12 +- src/federation/client.rs | 60 ++ src/federation/mod.rs | 39 +- src/federation/similarity.rs | 310 ++++++++ src/i18n/phrases.rs | 6 + src/main.rs | 8 + src/music/mod.rs | 66 ++ src/player/mod.rs | 140 ++++ src/similarity.rs | 1323 ++++++++++++++++++++++++++++++++ templates/admin/v2.html | 182 ++++- templates/player/modals.html | 11 + templates/player/scripts.html | 58 ++ templates/player/shell.html | 13 +- templates/player/styles.html | 15 + 22 files changed, 3092 insertions(+), 17 deletions(-) create mode 100644 flake.lock create mode 100644 flake.nix create mode 100644 src/federation/similarity.rs create mode 100644 src/similarity.rs diff --git a/.gitignore b/.gitignore index 20c612d..3346236 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,5 @@ /.claude /media /federation +/similarity-models +/federation-cache diff --git a/Cargo.lock b/Cargo.lock index ee64341..6414ad0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -159,6 +159,12 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "anymap3" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb5dfbc6d8d2675589ccbe4d0fd61df2419075625f8c1a62325e718e2b0049f9" + [[package]] name = "arc-swap" version = "1.9.2" @@ -488,6 +494,24 @@ dependencies = [ "virtue", ] +[[package]] +name = "bit-set" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2f926cc3060f09db9ebc5b52823d85268d24bb917e472c0c4bea35780a7d" +dependencies = [ + "bit-vec", +] + +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -1069,6 +1093,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-epoch" version = "0.9.20" @@ -1093,6 +1127,12 @@ version = "0.8.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" +[[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.5.5" @@ -1284,6 +1324,12 @@ dependencies = [ "serde", ] +[[package]] +name = "dasp_sample" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c87e182de0887fd5361989c677c4e8f5000cd9491d6d563161a8f3a5519fc7f" + [[package]] name = "data-encoding" version = "2.11.0" @@ -1341,6 +1387,17 @@ dependencies = [ "serde_core", ] +[[package]] +name = "derive-new" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2cdc8d50f426189eef89dac62fabfa0abb27d5cc008f25bf4156a0203325becc" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "derive_builder" version = "0.20.2" @@ -1493,12 +1550,30 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "downcast-rs" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "117240f60069e65410b3ae1bb213295bd828f707b5bec6596a1afc8793ce0cbc" + [[package]] name = "dyn-clone" version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "dyn-eq" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c2d035d21af5cde1a6f5c7b444a5bf963520a9f142e5d06931178433d7d5388" + +[[package]] +name = "dyn-hash" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fdab65db9274e0168143841eb8f864a0a21f8b1b8d2ba6812bbe6024346e99e" + [[package]] name = "ecdsa" version = "0.16.9" @@ -1641,6 +1716,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" @@ -1748,6 +1834,16 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" +[[package]] +name = "filetime" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" +dependencies = [ + "cfg-if", + "libc", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -1764,6 +1860,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "float-ord" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ce81f49ae8a0482e4c55ea62ebbd7e5a686af544c00b9d090bba3ff9be97b3d" + [[package]] name = "flume" version = "0.11.1" @@ -1836,7 +1938,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" [[package]] name = "furumusic" -version = "0.9.7" +version = "0.9.8" dependencies = [ "anyhow", "async-stream", @@ -1848,6 +1950,7 @@ dependencies = [ "cot", "croner", "encoding_rs", + "futures-util", "id3", "image", "librqbit", @@ -1856,6 +1959,8 @@ dependencies = [ "openidconnect", "postcard", "reqwest 0.12.28", + "rodio", + "rustfft", "schemars 0.9.0", "serde", "serde_json", @@ -1867,6 +1972,7 @@ dependencies = [ "tower", "tracing", "tracing-subscriber", + "tract-onnx", "uuid", ] @@ -2191,6 +2297,18 @@ dependencies = [ "tracing", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "num-traits", + "zerocopy", +] + [[package]] name = "hash32" version = "0.2.1" @@ -2236,6 +2354,8 @@ dependencies = [ "allocator-api2", "equivalent", "foldhash 0.2.0", + "serde", + "serde_core", ] [[package]] @@ -2799,6 +2919,15 @@ dependencies = [ "smallvec", ] +[[package]] +name = "inventory" +version = "0.3.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4f0c30c76f2f4ccee3fe55a2435f691ca00c0e4bd87abe4f4a851b1d4dac39b" +dependencies = [ + "rustversion", +] + [[package]] name = "ipconfig" version = "0.3.4" @@ -3491,6 +3620,12 @@ dependencies = [ "tracing", ] +[[package]] +name = "maplit" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d" + [[package]] name = "matchers" version = "0.2.0" @@ -3506,6 +3641,16 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "matrixmultiply" +version = "0.3.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f607c237553f086e7043417a51df26b2eb899d3caff94e6a67592ff992fedc7" +dependencies = [ + "autocfg", + "rawpointer", +] + [[package]] name = "md-5" version = "0.10.6" @@ -3531,6 +3676,12 @@ dependencies = [ "libc", ] +[[package]] +name = "memo-map" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38d1115007560874e373613744c6fba374c17688327a71c1476d1a5954cc857b" + [[package]] name = "mime" version = "0.3.17" @@ -3547,6 +3698,16 @@ dependencies = [ "unicase", ] +[[package]] +name = "minijinja" +version = "2.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42d74234349a775546a83af0f0c0c0e3a73227dee4950a542cda81a47240b3e6" +dependencies = [ + "memo-map", + "serde", +] + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -3614,9 +3775,9 @@ dependencies = [ [[package]] name = "music-dht" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f32daa9edf769fb5686e92ae6884e9fda6ea082452e4208309fc14301e26aef" +checksum = "91592b40de9f3c2158a39da105c821e9bf17f461fe142a56a8607fb0faf56a9c" dependencies = [ "async-trait", "blake3", @@ -3704,6 +3865,21 @@ dependencies = [ "tempfile", ] +[[package]] +name = "ndarray" +version = "0.17.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "520080814a7a6b4a6e9070823bb24b4531daac8c4627e08ba5de8c5ef2f2752d" +dependencies = [ + "matrixmultiply", + "num-complex 0.4.6", + "num-integer", + "num-traits", + "portable-atomic", + "portable-atomic-util", + "rawpointer", +] + [[package]] name = "ndk-context" version = "0.1.1" @@ -3835,6 +4011,24 @@ dependencies = [ "winapi", ] +[[package]] +name = "nom" +version = "8.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405" +dependencies = [ + "memchr", +] + +[[package]] +name = "nom-language" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2de2bc5b451bfedaef92c90b8939a8fff5770bdcc1fafd6239d086aab8fa6b29" +dependencies = [ + "nom", +] + [[package]] name = "nonzero_ext" version = "0.3.0" @@ -3918,10 +4112,20 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8536030f9fea7127f841b45bb6243b27255787fb4eb83958aa1ef9d2fdc0c36" dependencies = [ - "num-complex", + "num-complex 0.2.4", "num-integer", "num-iter", - "num-rational", + "num-rational 0.2.4", + "num-traits", +] + +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", "num-traits", ] @@ -3951,6 +4155,15 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-complex" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" +dependencies = [ + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" @@ -3998,6 +4211,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-rational" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824" +dependencies = [ + "num-bigint", + "num-integer", + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -4352,6 +4576,12 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" +[[package]] +name = "pastey" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4" + [[package]] name = "pem-rfc7468" version = "0.7.0" @@ -4562,6 +4792,15 @@ dependencies = [ "serde", ] +[[package]] +name = "portable-atomic-util" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618" +dependencies = [ + "portable-atomic", +] + [[package]] name = "portmapper" version = "0.19.1" @@ -4651,6 +4890,15 @@ dependencies = [ "num-traits", ] +[[package]] +name = "primal-check" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc0d895b311e3af9902528fbb8f928688abbd95872819320517cc24ca6b2bd08" +dependencies = [ + "num-integer", +] + [[package]] name = "primeorder" version = "0.13.6" @@ -4678,6 +4926,29 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools 0.14.0", + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "pxfm" version = "0.1.30" @@ -4883,6 +5154,16 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_distr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d431c2703ccf129de4d45253c03f49ebb22b97d6ad79ee3ecfc7e3f4862c1d8" +dependencies = [ + "num-traits", + "rand 0.10.2", +] + [[package]] name = "rand_pcg" version = "0.10.2" @@ -4901,6 +5182,32 @@ dependencies = [ "bitflags 2.13.1", ] +[[package]] +name = "rawpointer" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "60a357793950651c4ed0f3f52338f53b2f809f32d83a07f72909fa13e4c6c1e3" + +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -5099,6 +5406,18 @@ dependencies = [ "libc", ] +[[package]] +name = "rodio" +version = "0.22.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a536bb79db59098ef71a4dd4246c02eb87b316deceb1b68e0cde7167ec01eb" +dependencies = [ + "dasp_sample", + "num-rational 0.4.2", + "symphonia", + "thiserror 2.0.19", +] + [[package]] name = "rsa" version = "0.9.10" @@ -5154,6 +5473,20 @@ dependencies = [ "semver", ] +[[package]] +name = "rustfft" +version = "6.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21db5f9893e91f41798c88680037dba611ca6674703c1a18601b01a72c8adb89" +dependencies = [ + "num-complex 0.4.6", + "num-integer", + "num-traits", + "primal-check", + "strength_reduce", + "transpose", +] + [[package]] name = "rustix" version = "1.1.4" @@ -5254,6 +5587,19 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "safetensors" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79b079b829cb27a1c3c374341345ed2e8b2c0c839034522cee576c140bd7f846" +dependencies = [ + "hashbrown 0.16.1", + "libc", + "serde", + "serde_json", + "tempfile", +] + [[package]] name = "same-file" version = "1.0.6" @@ -5263,6 +5609,15 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "scan_fmt" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b53b0a5db882a8e2fdaae0a43f7b39e7e9082389e978398bdf223a55b581248" +dependencies = [ + "regex", +] + [[package]] name = "schannel" version = "0.1.29" @@ -5996,6 +6351,22 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "strength_reduce" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe895eb47f22e2ddd4dabc02bce419d2e643c8e3b585c78158b349195bc24d82" + +[[package]] +name = "string-interner" +version = "0.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad3df9b59e2eded8d825c7c4363ad339a20fb6bc0b9a4778560f518f59910b15" +dependencies = [ + "hashbrown 0.16.1", + "serde", +] + [[package]] name = "stringprep" version = "0.1.5" @@ -6329,6 +6700,17 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" +[[package]] +name = "tar" +version = "0.4.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840" +dependencies = [ + "filetime", + "libc", + "xattr", +] + [[package]] name = "tempfile" version = "3.27.0" @@ -6788,12 +7170,229 @@ dependencies = [ "tracing-log", ] +[[package]] +name = "tract-core" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "608e176a669d5da02cccc92bbfe5ee4e57686ed8841022608a9eba014d3b7886" +dependencies = [ + "anyhow", + "anymap3", + "bit-set", + "derive-new", + "downcast-rs", + "dyn-clone", + "dyn-eq", + "erased-serde", + "inventory", + "lazy_static", + "log", + "maplit", + "ndarray", + "num-complex 0.4.6", + "num-integer", + "num-traits", + "pastey", + "rustfft", + "serde", + "smallvec", + "tract-data", + "tract-linalg", +] + +[[package]] +name = "tract-data" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "870236dd45aaeb1381023cb709a67ff14ece608ee0b37f99aa166d166db9b0d0" +dependencies = [ + "anyhow", + "downcast-rs", + "dyn-clone", + "dyn-eq", + "dyn-hash", + "half", + "inventory", + "itertools 0.14.0", + "lazy_static", + "libm", + "maplit", + "ndarray", + "nom", + "nom-language", + "num-integer", + "num-traits", + "parking_lot", + "scan_fmt", + "smallvec", + "string-interner", +] + +[[package]] +name = "tract-extra" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "effcae1ebfce133e8bf6c79ad31298cbee0a9eceb3e4642fe1d14cb54cca78e6" +dependencies = [ + "tract-nnef", + "tract-pulse", +] + +[[package]] +name = "tract-hir" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28783b2bb583177685f65016a866b01eb3d383bce4fa7ae1b9692b325b64449d" +dependencies = [ + "derive-new", + "log", + "tract-core", +] + +[[package]] +name = "tract-linalg" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3e01491f7360806ef061af4c016a2d0a801d586768896394a0b8a7d6872c2b0" +dependencies = [ + "byteorder", + "cc", + "derive-new", + "downcast-rs", + "dyn-clone", + "dyn-eq", + "dyn-hash", + "half", + "lazy_static", + "log", + "minijinja", + "num-traits", + "pastey", + "scan_fmt", + "tract-data", + "walkdir", +] + +[[package]] +name = "tract-nnef" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00417fabf01aeea7bc56107367862e5bb8da18c9ad850c9061d3823700479ecc" +dependencies = [ + "byteorder", + "erased-serde", + "flate2", + "log", + "minijinja", + "nom", + "nom-language", + "safetensors", + "serde", + "serde_json", + "simd-adler32", + "tar", + "tract-core", + "walkdir", +] + +[[package]] +name = "tract-onnx" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3215dd27bddd2a041a20fee750013b400135186d3485501c9274c755b19ceb0" +dependencies = [ + "bytes", + "derive-new", + "dyn-eq", + "log", + "memmap2", + "num-integer", + "prost", + "smallvec", + "tract-extra", + "tract-hir", + "tract-nnef", + "tract-onnx-opl", + "tract-transformers", +] + +[[package]] +name = "tract-onnx-opl" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44f549ad3f245c1c00ce710c66cc0e086be0b637f35934d37b655ff175e1ab3e" +dependencies = [ + "dyn-eq", + "getrandom 0.4.3", + "log", + "rand 0.10.2", + "rand_distr", + "rustfft", + "tract-extra", + "tract-nnef", +] + +[[package]] +name = "tract-pulse" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a164e22e96ab9b5c90458fa270700570d87963e3dec9ccaac1ab18d9fa763ce2" +dependencies = [ + "downcast-rs", + "dyn-eq", + "erased-serde", + "lazy_static", + "log", + "serde", + "tract-pulse-opl", + "tract-transformers", +] + +[[package]] +name = "tract-pulse-opl" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65418f9e93e0af0d567f4f2f1bf2309b53930fa2543c635a69fda10c87a04987" +dependencies = [ + "downcast-rs", + "dyn-eq", + "lazy_static", + "tract-nnef", +] + +[[package]] +name = "tract-transformers" +version = "0.23.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8471ebf7f52d226552283d9130539595540c28ed41b8e032df3830507d3ccdd5" +dependencies = [ + "float-ord", + "rayon", + "tract-nnef", +] + +[[package]] +name = "transpose" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad61aed86bc3faea4300c7aee358b4c6d0c8d6ccc36524c96e4c92ccf26e77e" +dependencies = [ + "num-integer", + "strength_reduce", +] + [[package]] name = "try-lock" version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[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.1" @@ -7559,6 +8158,16 @@ dependencies = [ "tap", ] +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix", +] + [[package]] name = "xml-rs" version = "0.8.28" diff --git a/Cargo.toml b/Cargo.toml index 2574686..a83da2d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumusic" -version = "0.9.8" +version = "0.10.0" edition = "2024" description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL" @@ -12,7 +12,7 @@ cot = { version = "0.6.0", default-features = false, features = ["postgres", "js schemars = { version = "0.9", features = ["derive"] } serde = { version = "1", features = ["derive"] } openidconnect = "4.0" -reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "stream"] } tokio = { version = "1", features = ["sync", "fs", "io-util"] } async-stream = "0.3" bytes = "1" @@ -31,6 +31,10 @@ md-5 = "0.10" image = { version = "0.25", default-features = false, features = ["jpeg", "png", "webp", "gif", "bmp"] } sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres"] } anyhow = "1.0" +futures-util = "0.3" +rodio = { version = "0.22.2", default-features = false, features = ["mp3", "flac", "vorbis", "wav", "symphonia-aac", "symphonia-isomp4", "symphonia-alac", "symphonia-adpcm", "symphonia-aiff", "symphonia-mkv", "symphonia-pcm"] } +rustfft = "6.4.1" +tract-onnx = "0.23.4" tokio-cron-scheduler = "0.15" croner = "3" async-trait = "0.1" @@ -39,4 +43,4 @@ uuid = "1" librqbit = { version = "8.1.1", features = ["disable-upload"] } # P2P federation: publishes the library into a shared DHT and serves audio / # catalogs to furumi peers (TUI clients) over the frid stack. -music-dht = "0.3" +music-dht = "0.3.1" diff --git a/flake.lock b/flake.lock new file mode 100644 index 0000000..379045c --- /dev/null +++ b/flake.lock @@ -0,0 +1,27 @@ +{ + "nodes": { + "nixpkgs": { + "locked": { + "lastModified": 1786106723, + "narHash": "sha256-zDSUbpoeo/9ZmD2+wXnzxoo1+uhL8vxc0b8yuYMKYq0=", + "owner": "NixOS", + "repo": "nixpkgs", + "rev": "f13ff45afd1bb73e640eaa08a7066dbed07e3238", + "type": "github" + }, + "original": { + "owner": "NixOS", + "ref": "nixos-unstable", + "repo": "nixpkgs", + "type": "github" + } + }, + "root": { + "inputs": { + "nixpkgs": "nixpkgs" + } + } + }, + "root": "root", + "version": 7 +} diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..706296b --- /dev/null +++ b/flake.nix @@ -0,0 +1,38 @@ +{ + description = "Furumusic development environment"; + + inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable"; + + outputs = { nixpkgs, ... }: + let + supportedSystems = [ + "x86_64-linux" + "aarch64-linux" + "x86_64-darwin" + "aarch64-darwin" + ]; + forEachSystem = function: + nixpkgs.lib.genAttrs supportedSystems (system: + function (import nixpkgs { inherit system; })); + in + { + devShells = forEachSystem (pkgs: { + default = pkgs.mkShell { + nativeBuildInputs = with pkgs; [ + cargo + clippy + pkg-config + rustc + rustfmt + ]; + + buildInputs = with pkgs; [ + cacert + openssl + ] ++ lib.optionals stdenv.isDarwin [ libiconv ]; + + RUST_SRC_PATH = "${pkgs.rustPlatform.rustLibSrc}"; + }; + }); + }; +} diff --git a/src/admin/mod.rs b/src/admin/mod.rs index cb95ac6..08bea5d 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -422,6 +422,20 @@ impl App for AdminApp { }), "admin_v2_federation_status", ), + Route::with_handler_and_name( + "/v2/api/similarity", + get(move |session: Session, db: Database| async move { + v2::similarity_status(session, db).await + }), + "admin_v2_similarity_status", + ), + Route::with_handler_and_name( + "/v2/api/similarity/clear", + cot::router::method::post(move |session: Session, db: Database| async move { + v2::similarity_clear(session, db).await + }), + "admin_v2_similarity_clear", + ), Route::with_handler_and_name( "/v2/api/federation/sync", cot::router::method::post(move |session: Session, db: Database| async move { diff --git a/src/admin/v2.rs b/src/admin/v2.rs index d4526c7..2d27ba0 100644 --- a/src/admin/v2.rs +++ b/src/admin/v2.rs @@ -454,6 +454,14 @@ struct AdminSettingsValues { federation_network_id: String, #[serde(default)] federation_save_on_listen: bool, + #[serde(default)] + similarity_enabled: bool, + #[serde(default = "default_similarity_model")] + similarity_model: String, + #[serde(default = "default_similarity_profile")] + similarity_profile: String, + #[serde(default = "default_similarity_workers")] + similarity_workers: String, } #[derive(Debug, Clone, Serialize, JsonSchema)] @@ -481,6 +489,10 @@ struct AdminSettingsSources { federation_enabled: &'static str, federation_network_id: &'static str, federation_save_on_listen: &'static str, + similarity_enabled: &'static str, + similarity_model: &'static str, + similarity_profile: &'static str, + similarity_workers: &'static str, } #[derive(Debug, Deserialize)] @@ -511,6 +523,26 @@ pub(super) struct UpdateSettingsRequest { federation_network_id: String, #[serde(default)] federation_save_on_listen: bool, + #[serde(default)] + similarity_enabled: bool, + #[serde(default = "default_similarity_model")] + similarity_model: String, + #[serde(default = "default_similarity_profile")] + similarity_profile: String, + #[serde(default = "default_similarity_workers")] + similarity_workers: String, +} + +fn default_similarity_model() -> String { + crate::similarity::DEFAULT_MODEL_ID.to_owned() +} + +fn default_similarity_profile() -> String { + crate::similarity::DEFAULT_PROFILE_ID.to_owned() +} + +fn default_similarity_workers() -> String { + "1".to_owned() } #[derive(Debug, Serialize, JsonSchema)] @@ -948,6 +980,29 @@ pub async fn update_settings( if let Err(response) = require_admin_json(&session, &db).await { return Ok(response); } + let similarity_model = body.similarity_model.trim(); + if crate::similarity::model_by_id(similarity_model).is_none() { + return Ok(json_error( + StatusCode::BAD_REQUEST, + "unknown similarity model", + )); + } + let similarity_profile = body.similarity_profile.trim(); + if crate::similarity::profile_by_id(similarity_profile).is_none() { + return Ok(json_error( + StatusCode::BAD_REQUEST, + "unknown similarity preprocessing profile", + )); + } + let similarity_workers = match body.similarity_workers.trim().parse::() { + Ok(workers @ 1..=16) => workers, + _ => { + return Ok(json_error( + StatusCode::BAD_REQUEST, + "similarity workers must be an integer from 1 to 16", + )); + } + }; let fields = [ ( "auth_password_enabled", @@ -1002,6 +1057,10 @@ pub async fn update_settings( "federation_save_on_listen", body.federation_save_on_listen.to_string(), ), + ("similarity_enabled", body.similarity_enabled.to_string()), + ("similarity_model", similarity_model.to_string()), + ("similarity_profile", similarity_profile.to_string()), + ("similarity_workers", similarity_workers.to_string()), ]; for (key, value) in fields { let mut entry = ConfigEntry::new(key.to_string(), value); @@ -1014,6 +1073,7 @@ pub async fn update_settings( // the freshly saved settings — no server restart involved. let (fresh, _) = AppConfig::load_with_db(&db).await; tokio::spawn(async move { + crate::similarity::handle().apply(&fresh); crate::federation::handle().apply(&fresh).await; }); Json(serde_json::json!({ "ok": true })).into_response() @@ -1033,6 +1093,57 @@ pub async fn federation_status( Json(crate::federation::handle().status().await).into_response() } +pub async fn similarity_status( + session: Session, + db: Database, +) -> cot::Result { + if let Err(response) = require_admin_json(&session, &db).await { + return Ok(response); + } + let manager = crate::similarity::handle(); + let status = manager.status(); + let model_id = if status.model.is_empty() { + crate::similarity::DEFAULT_MODEL_ID + } else { + &status.model + }; + let profiles = crate::similarity::PROFILES + .iter() + .map(|profile| { + serde_json::json!({ + "id": profile.id, + "title": profile.title, + "details": crate::similarity::profile_details( + profile.id, + model_id, + ).unwrap_or_default(), + }) + }) + .collect::>(); + Json(serde_json::json!({ + "status": status, + "models": crate::similarity::MODELS, + "profiles": profiles, + })) + .into_response() +} + +pub async fn similarity_clear( + session: Session, + db: Database, +) -> cot::Result { + if let Err(response) = require_admin_json(&session, &db).await { + return Ok(response); + } + match crate::similarity::handle().clear().await { + Ok(()) => Json(serde_json::json!({ "ok": true })).into_response(), + Err(error) => Ok(json_error( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("embedding cleanup failed: {error:#}"), + )), + } +} + pub async fn federation_sync( session: Session, db: Database, @@ -1153,6 +1264,10 @@ fn settings_dto(config: AppConfig, sources: ConfigSources) -> AdminSettingsDto { federation_enabled: config.federation_enabled, federation_network_id: config.federation_network_id, federation_save_on_listen: config.federation_save_on_listen, + similarity_enabled: config.similarity_enabled, + similarity_model: config.similarity_model, + similarity_profile: config.similarity_profile, + similarity_workers: config.similarity_workers.to_string(), }, sources: AdminSettingsSources { auth_password_enabled: sources.auth_password_enabled.code(), @@ -1178,6 +1293,10 @@ fn settings_dto(config: AppConfig, sources: ConfigSources) -> AdminSettingsDto { federation_enabled: sources.federation_enabled.code(), federation_network_id: sources.federation_network_id.code(), federation_save_on_listen: sources.federation_save_on_listen.code(), + similarity_enabled: sources.similarity_enabled.code(), + similarity_model: sources.similarity_model.code(), + similarity_profile: sources.similarity_profile.code(), + similarity_workers: sources.similarity_workers.code(), }, } } diff --git a/src/config.rs b/src/config.rs index 3d000e7..a4c7fbd 100644 --- a/src/config.rs +++ b/src/config.rs @@ -138,6 +138,10 @@ pub struct ConfigSources { pub federation_enabled: ConfigSource, pub federation_network_id: ConfigSource, pub federation_save_on_listen: ConfigSource, + pub similarity_enabled: ConfigSource, + pub similarity_model: ConfigSource, + pub similarity_profile: ConfigSource, + pub similarity_workers: ConfigSource, } impl Default for ConfigSources { @@ -168,6 +172,10 @@ impl Default for ConfigSources { federation_enabled: ConfigSource::Default, federation_network_id: ConfigSource::Default, federation_save_on_listen: ConfigSource::Default, + similarity_enabled: ConfigSource::Default, + similarity_model: ConfigSource::Default, + similarity_profile: ConfigSource::Default, + similarity_workers: ConfigSource::Default, } } } @@ -285,6 +293,14 @@ pub struct AppConfig { /// Whether a federated track requested for playback is imported into the /// shared local library. This is a server-wide administrator policy. pub federation_save_on_listen: bool, + /// Whether local embedding calculation and similarity search are enabled. + pub similarity_enabled: bool, + /// Embedding model selected by the administrator. + pub similarity_model: String, + /// Audio preprocessing profile selected by the administrator. + pub similarity_profile: String, + /// Maximum number of concurrent CPU embedding workers. + pub similarity_workers: u64, } impl Default for AppConfig { @@ -315,6 +331,12 @@ impl Default for AppConfig { federation_enabled: false, federation_network_id: String::new(), federation_save_on_listen: false, + similarity_enabled: false, + similarity_model: "discogs-effnet-bsdynamic-1".into(), + similarity_profile: "furumi-full-track-v1".into(), + similarity_workers: std::thread::available_parallelism() + .map(|count| (count.get() / 2).clamp(1, 4) as u64) + .unwrap_or(1), } } } @@ -346,6 +368,10 @@ impl_env_overrides!( federation_enabled, federation_network_id, federation_save_on_listen, + similarity_enabled, + similarity_model, + similarity_profile, + similarity_workers, ); impl AppConfig { @@ -476,6 +502,10 @@ impl AppConfig { apply_db_field!(federation_enabled); apply_db_field!(federation_network_id); apply_db_field!(federation_save_on_listen); + apply_db_field!(similarity_enabled); + apply_db_field!(similarity_model); + apply_db_field!(similarity_profile); + apply_db_field!(similarity_workers); } } @@ -495,6 +525,13 @@ mod tests { let cfg = AppConfig::default(); assert!(cfg.database_url.is_empty()); assert_eq!(cfg.log_level, "info"); + assert!(!cfg.similarity_enabled); + assert_eq!(cfg.similarity_model, crate::similarity::DEFAULT_MODEL_ID); + assert_eq!( + cfg.similarity_profile, + crate::similarity::DEFAULT_PROFILE_ID + ); + assert!((1..=4).contains(&cfg.similarity_workers)); } #[test] diff --git a/src/federation/capabilities.rs b/src/federation/capabilities.rs index 4bdbd61..a2fcb96 100644 --- a/src/federation/capabilities.rs +++ b/src/federation/capabilities.rs @@ -5,8 +5,8 @@ use std::time::Duration; use anyhow::Result; use music_dht::StreamAcceptor; use music_dht::capabilities::{ - CAPABILITIES_PROTOCOL_VERSION, CapabilityManifest, CapabilityMessage, JAM_ID, read_message, - write_message, + CAPABILITIES_PROTOCOL_VERSION, CapabilityManifest, CapabilityMessage, JAM_ID, SIMILARITY_ID, + read_message, write_message, }; use super::serve::AUDIO_PROTOCOL_VERSION; @@ -16,6 +16,10 @@ fn local_manifest() -> CapabilityManifest { // The web server does not expose federation Jam yet. .without_protocol(JAM_ID) .with_protocol("audio", AUDIO_PROTOCOL_VERSION) + .with_protocol( + SIMILARITY_ID, + music_dht::similarity::SIMILARITY_PROTOCOL_VERSION, + ) } pub async fn serve(mut acceptor: StreamAcceptor) { @@ -61,6 +65,10 @@ mod tests { Some(&AUDIO_PROTOCOL_VERSION) ); assert!(!manifest.protocols.contains_key(JAM_ID)); + assert_eq!( + manifest.protocols.get(SIMILARITY_ID), + Some(&music_dht::similarity::SIMILARITY_PROTOCOL_VERSION) + ); manifest.validate().unwrap(); } } diff --git a/src/federation/client.rs b/src/federation/client.rs index 041925d..5d052f6 100644 --- a/src/federation/client.rs +++ b/src/federation/client.rs @@ -98,6 +98,66 @@ pub struct SearchEvent { } impl Federation { + pub async fn prepare_similarity_tracks( + &self, + tracks: Vec, + ) -> Result> { + let pool = self.pool().await?; + let mut prepared = Vec::new(); + for track in tracks { + let Some(content_id) = track.content_id.as_deref().and_then(normalize_content_id) + else { + continue; + }; + let local = local_availability(&pool, &content_id).await?; + // A local result is already present in the first result section. + if local.is_some() { + continue; + } + let owner = track.owner; + let item_id = track.item_id; + let dto = TrackDto { + key: TrackKeyDto { + content_id: content_id.clone(), + }, + metadata: TrackMetadataDto { + title: track.title, + artists: artist_refs(&track.artist_names), + featured_artists: artist_refs(&track.featured_artist_names), + release: track.release_title.map(|title| ReleaseRefDto { + key: ReleaseKeyDto { + normalized_title: music_dht::normalize_name(&title), + primary_artists: track + .artist_names + .iter() + .map(|artist| music_dht::normalize_name(artist)) + .collect(), + release_type: None, + year: track.year, + }, + local_id: None, + title, + }), + year: track.year, + duration_seconds: track.duration_seconds.map(|value| value as f64), + track_number: track.track_number, + disc_number: track.disc_number, + cover_url: Some(format!( + "/api/player/federation/tracks/artwork?owner={owner}&item_id={item_id}" + )), + }, + availability: TrackAvailabilityDto { + state: "federated", + local: None, + federation: vec![FederationSourceDto { owner, item_id }], + }, + }; + persist_track_ref(&pool, &dto).await?; + prepared.push(dto); + } + Ok(prepared) + } + pub fn stream_artist_catalogs( self: &std::sync::Arc, name: String, diff --git a/src/federation/mod.rs b/src/federation/mod.rs index 73e2064..a1f1a15 100644 --- a/src/federation/mod.rs +++ b/src/federation/mod.rs @@ -5,18 +5,21 @@ //! releases, tracks — names and small metadata, never files) into the //! shared DHT and serves audio, track metadata, cover art and per-artist //! catalogs to other peers (TUI clients) over the same wire protocols the -//! clients speak among themselves. Serve-only: the server does not search -//! or download from other peers. +//! clients speak among themselves. The web player also searches known peers +//! for catalog metadata and, when enabled, compatible similarity embeddings; +//! local playback and the local library remain independent of the network. //! //! Settings are the regular admin config entries (`federation_enabled`, -//! `federation_network_id`, `federation_save_on_listen`) and apply on the fly — saving the settings -//! starts, stops or re-joins the node without a server restart. +//! `federation_network_id`, `federation_save_on_listen`) and apply on the fly — +//! saving the settings starts, stops or re-joins the node without a server +//! restart. mod capabilities; pub mod client; pub mod devices; mod receive; mod serve; +mod similarity; mod storage; use std::collections::{HashMap, HashSet, VecDeque}; @@ -38,6 +41,7 @@ use crate::config::AppConfig; use storage::PostgresFederationStorage; pub use serve::{AUDIO_ALPN, CATALOG_ALPN}; +pub use similarity::SIMILARITY_ALPN; /// How often the published library is re-synchronized with the database. const SYNC_INTERVAL: Duration = Duration::from_secs(60); @@ -124,6 +128,7 @@ struct TransportStatsState { audio_samples: u64, catalog_samples: u64, sync_samples: u64, + similarity_samples: u64, last: VecDeque, } @@ -157,6 +162,7 @@ impl TransportStats { "audio" => state.audio_samples += 1, "catalog" => state.catalog_samples += 1, "device-sync" => state.sync_samples += 1, + "similarity" => state.similarity_samples += 1, _ => {} } state.last.push_front(sample); @@ -177,6 +183,7 @@ impl TransportStats { "audio_samples": state.audio_samples, "catalog_samples": state.catalog_samples, "sync_samples": state.sync_samples, + "similarity_samples": state.similarity_samples, "last_path": latest.map(|sample| sample.selected_path.clone()), "last_rtt_ms": latest.and_then(|sample| sample.selected_rtt_ms), "last_peer": latest.map(|sample| sample.peer_id.clone()), @@ -382,6 +389,7 @@ impl Federation { .stream_protocol(AUDIO_ALPN) .stream_protocol(CATALOG_ALPN) .stream_protocol(devices::SYNC_ALPN) + .stream_protocol(SIMILARITY_ALPN) .schema_independent_stream_protocol(CAPABILITIES_ALPN) .build() .map_err(|err| anyhow::anyhow!("invalid federation config: {err}"))?; @@ -454,6 +462,15 @@ impl Federation { .stream_acceptor(CAPABILITIES_ALPN) .map_err(|err| anyhow::anyhow!("failed to take the capabilities acceptor: {err}"))?; let capabilities_task = tokio::spawn(capabilities::serve(capabilities_acceptor)); + let similarity_acceptor = service + .stream_acceptor(SIMILARITY_ALPN) + .map_err(|err| anyhow::anyhow!("failed to take the similarity acceptor: {err}"))?; + let similarity_task = tokio::spawn(similarity::serve_peers( + similarity_acceptor, + crate::similarity::handle(), + service.endpoint_id(), + Arc::clone(&self.transport_stats), + )); *guard = Some(Running { service, @@ -466,6 +483,7 @@ impl Federation { device_task, device_sync_task, capabilities_task, + similarity_task, ], }); self.set_error(None); @@ -827,6 +845,19 @@ impl Federation { Ok(peer.to_string()) } + pub async fn search_similarity( + &self, + query: crate::similarity::QueryVector, + limit: usize, + ) -> Result> { + anyhow::ensure!( + crate::similarity::handle().enabled(), + "similarity search is disabled" + ); + let service = self.service().await?; + similarity::search(service, query, limit, Arc::clone(&self.transport_stats)).await + } + pub async fn fed_device_status( &self, user_id: i64, diff --git a/src/federation/similarity.rs b/src/federation/similarity.rs new file mode 100644 index 0000000..512c0ec --- /dev/null +++ b/src/federation/similarity.rs @@ -0,0 +1,310 @@ +//! Furumusic policy and PostgreSQL adapter for the shared similarity protocol. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context as _, Result}; +use futures_util::stream::{self, StreamExt as _}; +use music_dht::similarity::{self as wire, SimilarityHit, SimilarityRequest, SimilarityResponse}; +use music_dht::{ByteStream, EndpointId, ItemId, ItemKind, MusicDhtService, StreamAcceptor}; + +use crate::similarity::{Manager, QueryVector}; + +use super::TransportStats; + +pub use music_dht::similarity::SIMILARITY_ALPN; + +const MAX_QUERY_PEERS: usize = 16; +const QUERY_CONCURRENCY: usize = 6; +const QUERY_TIMEOUT: Duration = Duration::from_secs(5); +const MAX_PER_ARTIST: usize = 3; +const MAX_NEAR_DUPLICATE_SIGNATURE_DISTANCE: u32 = 8; + +#[derive(Debug, Clone)] +pub struct RemoteSimilarityTrack { + pub owner: String, + pub item_id: String, + pub title: String, + pub artist_names: Vec, + pub featured_artist_names: Vec, + pub year: Option, + pub duration_seconds: Option, + pub content_id: Option, + pub release_title: Option, + pub track_number: Option, + pub disc_number: Option, +} + +pub async fn serve_peers( + mut acceptor: StreamAcceptor, + manager: Arc, + own: EndpointId, + transport: Arc, +) { + while let Some(stream) = acceptor.accept().await { + let manager = Arc::clone(&manager); + let transport = Arc::clone(&transport); + tokio::spawn(async move { + let peer = stream.peer_id; + if let Err(error) = serve_one(stream, manager, own, transport).await { + tracing::warn!(peer = %peer, "similarity request failed: {error:#}"); + } + }); + } +} + +async fn serve_one( + mut stream: ByteStream, + manager: Arc, + own: EndpointId, + transport: Arc, +) -> Result<()> { + super::record_stream_transport(&transport, "similarity", "inbound", "open", &stream); + let request = wire::read_request(&mut stream).await?; + let response = if !manager.enabled() { + SimilarityResponse::refused("similarity search is disabled on this instance")? + } else { + let profile_id = request.profile_id; + let vector = request.vector; + let limit = request.limit; + let rank_manager = Arc::clone(&manager); + let ranked = tokio::task::spawn_blocking(move || { + rank_manager.rank_vector(&profile_id, &vector, None, None, limit) + }) + .await + .context("local similarity task failed") + .and_then(|result| result); + match ranked { + Ok(ranked) => { + let ids = ranked + .iter() + .map(|track| track.track_id) + .collect::>(); + match manager.metadata_for_tracks(&ids).await { + Ok(metadata) => { + let by_id = ranked + .into_iter() + .map(|track| (track.track_id, track)) + .collect::>(); + let hits = metadata + .into_iter() + .filter_map(|track| { + let ranked = by_id.get(&track.track_id)?; + let hit = SimilarityHit { + score: ranked.score, + item_id: hex( + ItemId::derive( + &own, + ItemKind::Track, + &format!("track:{}", track.track_id), + ) + .as_bytes(), + ), + title: track.title, + artist_names: track.artist_names, + featured_artist_names: track.featured_artist_names, + year: track.year, + duration_seconds: Some(track.duration_seconds.round() as i64), + content_id: track.content_id, + release_title: Some(track.release_title), + track_number: track.track_number, + disc_number: track.disc_number, + embedding_signature: Some(ranked.embedding_signature), + }; + match hit.validate() { + Ok(()) => Some(hit), + Err(error) => { + tracing::debug!(%error, "invalid local similarity metadata skipped"); + None + } + } + }) + .collect(); + SimilarityResponse::success(hits)? + } + Err(error) => SimilarityResponse::refused(format!( + "similarity metadata is unavailable: {error:#}" + ))?, + } + } + Err(error) => { + SimilarityResponse::refused(format!("similarity query is unavailable: {error:#}"))? + } + } + }; + wire::write_response(&mut stream, &response).await?; + stream.send.finish()?; + let _ = stream.send.stopped().await; + super::record_stream_transport(&transport, "similarity", "inbound", "done", &stream); + Ok(()) +} + +pub async fn search( + service: Arc, + query: QueryVector, + limit: usize, + transport: Arc, +) -> Result> { + let own = service.endpoint_id(); + let mut peers = Vec::new(); + let mut seen = HashSet::new(); + for peer in service + .connected_peers() + .into_iter() + .chain(service.known_peers().into_iter().map(|peer| peer.peer_id)) + { + if peer != own && seen.insert(peer) { + peers.push(peer); + } + if peers.len() >= MAX_QUERY_PEERS { + break; + } + } + let query_signature = wire::embedding_signature(&query.vector)?; + let request = Arc::new(SimilarityRequest::new( + query.profile_id, + query.vector, + limit.clamp(1, wire::MAX_SIMILARITY_RESULTS), + )?); + + let responses = stream::iter(peers.into_iter().map(|peer| { + let service = Arc::clone(&service); + let request = Arc::clone(&request); + let transport = Arc::clone(&transport); + async move { + tokio::time::timeout( + QUERY_TIMEOUT, + query_peer(service, peer, &request, transport), + ) + .await + .map_err(|_| anyhow::anyhow!("similarity peer timed out"))? + } + })) + .buffer_unordered(QUERY_CONCURRENCY) + .collect::>() + .await; + + let mut hits = Vec::new(); + for response in responses { + match response { + Ok(peer_hits) => hits.extend(peer_hits), + Err(error) => tracing::debug!(%error, "similarity peer query skipped"), + } + } + hits.sort_by(|left, right| right.1.total_cmp(&left.1)); + let mut dedup = HashSet::new(); + let mut signatures = vec![query_signature]; + let mut artist_counts: HashMap = HashMap::new(); + let mut tracks = Vec::new(); + for (track, _, signature) in hits { + if query + .source_content_id + .as_deref() + .is_some_and(|source| track.content_id.as_deref() == Some(source)) + { + continue; + } + let key = track + .content_id + .clone() + .unwrap_or_else(|| format!("{}:{}", track.owner, track.item_id)); + if !dedup.insert(key) { + continue; + } + if signature.is_some_and(|candidate| { + signatures.iter().any(|existing| { + wire::signature_distance(&candidate, existing) + <= MAX_NEAR_DUPLICATE_SIGNATURE_DISTANCE + }) + }) { + continue; + } + let artist = track + .artist_names + .first() + .map(|name| music_dht::normalize_name(name)) + .unwrap_or_default(); + let count = artist_counts.entry(artist.clone()).or_default(); + if !artist.is_empty() && *count >= MAX_PER_ARTIST { + continue; + } + *count += 1; + if let Some(signature) = signature { + signatures.push(signature); + } + tracks.push(track); + if tracks.len() >= limit.min(wire::MAX_SIMILARITY_RESULTS) { + break; + } + } + Ok(tracks) +} + +async fn query_peer( + service: Arc, + owner: EndpointId, + request: &SimilarityRequest, + transport: Arc, +) -> Result< + Vec<( + RemoteSimilarityTrack, + f32, + Option<[u8; wire::SIMILARITY_SIGNATURE_BYTES]>, + )>, +> { + let mut stream = service + .open_stream(owner, SIMILARITY_ALPN) + .await + .map_err(|error| anyhow::anyhow!("cannot reach similarity peer: {error}"))?; + super::record_stream_transport(&transport, "similarity", "outbound", "open", &stream); + let response = wire::exchange(&mut stream, request).await?; + super::record_stream_transport(&transport, "similarity", "outbound", "done", &stream); + anyhow::ensure!( + response.ok, + "peer refused similarity query: {}", + response.error.unwrap_or_default() + ); + Ok(response + .hits + .into_iter() + .map(|hit| { + let score = hit.score; + let signature = hit.embedding_signature; + ( + RemoteSimilarityTrack { + owner: owner.to_string(), + item_id: hit.item_id, + title: hit.title, + artist_names: hit.artist_names, + featured_artist_names: hit.featured_artist_names, + year: hit.year, + duration_seconds: hit.duration_seconds, + content_id: hit.content_id, + release_title: hit.release_title, + track_number: hit.track_number, + disc_number: hit.disc_number, + }, + score, + signature, + ) + }) + .collect()) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hamming_threshold_keeps_exact_and_near_duplicates_out() { + let query = [0u8; wire::SIMILARITY_SIGNATURE_BYTES]; + let mut near = query; + near[0] = 0b0000_0111; + assert!(wire::signature_distance(&query, &near) <= MAX_NEAR_DUPLICATE_SIGNATURE_DISTANCE); + } +} diff --git a/src/i18n/phrases.rs b/src/i18n/phrases.rs index 700fdbb..241607f 100644 --- a/src/i18n/phrases.rs +++ b/src/i18n/phrases.rs @@ -298,6 +298,9 @@ translations! { player_likes_playlist: "Likes" , "Лайки"; player_listened: "listened" , "прослушано"; player_search_placeholder: "Search artists, releases, tracks..." , "Поиск артистов, релизов, треков..."; + player_search_similar_to: "Search similar to:" , "Поиск похожих на:"; + player_find_similar: "Find similar tracks" , "Найти похожие треки"; + player_similarity_failed: "Similarity search failed" , "Не удалось найти похожие треки"; player_connection_lost: "Server connection lost" , "Нет соединения с сервером"; player_connection_lost_detail: "Player cannot reach the server. Retrying..." , "Плеер не может связаться с сервером. Повторяю..."; player_active_device: "Active device" , "Активный девайс"; @@ -314,6 +317,9 @@ translations! { player_delete_playlist_confirm: "Delete this playlist?" , "Удалить этот плейлист?"; player_rename: "Rename" , "Переименовать"; player_close: "Close" , "Закрыть"; + player_interface_language: "Interface language" , "Язык интерфейса"; + player_language_description: "Choose the language used by the web player." , "Выберите язык интерфейса веб-плеера."; + player_switch_language: "Русский" , "English"; player_log_out: "Log out" , "Выйти"; player_admin_panel: "Admin Panel" , "Админка"; player_info: "Info" , "Информация"; diff --git a/src/main.rs b/src/main.rs index a3ed554..b0bc809 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,6 +13,7 @@ mod music; mod oidc; mod player; mod scheduler; +mod similarity; mod torrents; mod user; @@ -567,6 +568,13 @@ impl Project for FuruProject { federation::handle().boot(&fed_config).await; }); + // Embedding calculation is an independent, server-wide background + // service. It remains useful locally when federation is disabled. + let similarity_config = Arc::clone(&self.app_config); + tokio::spawn(async move { + similarity::handle().boot(&similarity_config).await; + }); + apps.register(cot::session::db::SessionApp::new()); apps.register_with_views( FuruApp { diff --git a/src/music/mod.rs b/src/music/mod.rs index eeb3284..f2123c2 100644 --- a/src/music/mod.rs +++ b/src/music/mod.rs @@ -2476,6 +2476,71 @@ pub mod db_migrations { &[Operation::custom(repair_legacy_listen_qualification).build()]; } + #[cot::db::migrations::migration_op] + async fn create_similarity_embeddings( + ctx: migrations::MigrationContext<'_>, + ) -> cot::db::Result<()> { + ctx.db + .raw( + "CREATE TABLE IF NOT EXISTS furumusic__similarity_profile ( + profile_id TEXT PRIMARY KEY, + model_id TEXT NOT NULL, + model_version TEXT NOT NULL, + model_sha256 TEXT NOT NULL, + preprocessing TEXT NOT NULL, + dimensions INTEGER NOT NULL, + active BOOLEAN NOT NULL DEFAULT FALSE, + created_at TEXT NOT NULL + )", + ) + .await?; + ctx.db + .raw( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_similarity_profile_active + ON furumusic__similarity_profile (active) + WHERE active = TRUE", + ) + .await?; + ctx.db + .raw( + "CREATE TABLE IF NOT EXISTS furumusic__track_embedding ( + track_id BIGINT NOT NULL REFERENCES furumusic__track(id) + ON DELETE CASCADE, + profile_id TEXT NOT NULL REFERENCES furumusic__similarity_profile(profile_id) + ON DELETE CASCADE, + dimensions INTEGER NOT NULL, + vector BYTEA NOT NULL, + source_sha256 TEXT NOT NULL, + source_content_id TEXT, + computed_at TEXT NOT NULL, + PRIMARY KEY (track_id, profile_id) + )", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_track_embedding_profile + ON furumusic__track_embedding (profile_id, track_id)", + ) + .await?; + Ok(()) + } + + #[derive(Debug, Copy, Clone)] + pub struct M0043CreateSimilarityEmbeddings; + + impl migrations::Migration for M0043CreateSimilarityEmbeddings { + const APP_NAME: &'static str = "furumusic"; + const MIGRATION_NAME: &'static str = "m_0043_create_similarity_embeddings"; + const DEPENDENCIES: &'static [migrations::MigrationDependency] = + &[migrations::MigrationDependency::migration( + "furumusic", + "m_0042_repair_legacy_listen_qualification", + )]; + const OPERATIONS: &'static [Operation] = + &[Operation::custom(create_similarity_embeddings).build()]; + } + pub const MIGRATIONS: &[&SyncDynMigration] = &[ &M0006CreateMediaFile, &M0007CreateArtist, @@ -2509,5 +2574,6 @@ pub mod db_migrations { &M0040CreateContentAddressedMusicRefs, &M0041CreateSyncedListenHistory, &M0042RepairLegacyListenQualification, + &M0043CreateSimilarityEmbeddings, ]; } diff --git a/src/player/mod.rs b/src/player/mod.rs index 60b223a..4fb2dec 100644 --- a/src/player/mod.rs +++ b/src/player/mod.rs @@ -14,6 +14,7 @@ use cot::router::method::{delete, get, post}; use cot::router::{Route, Router}; use cot::session::Session; use cot::{App, Body, Template}; +use serde::Serialize; use sqlx::Row as _; use crate::auth; @@ -4314,6 +4315,118 @@ async fn load_track_items_by_ids(pool: &sqlx::PgPool, ids: &[i64]) -> cot::Resul .collect()) } +#[derive(Debug, Serialize)] +struct SimilaritySearchResponse { + label: String, + tracks: Vec, + federation_tracks: Vec, + federation_error: Option, +} + +async fn similarity_search_handler( + auth_ctx: auth::AuthContext, + session: Session, + db: Database, + pool: &sqlx::PgPool, + Path(path): Path, +) -> cot::Result { + let Some(_user) = auth::get_request_user(&auth_ctx, &session, &db).await else { + return Ok(json_error(StatusCode::UNAUTHORIZED, "not authenticated")); + }; + if path.id <= 0 { + return Ok(json_error(StatusCode::BAD_REQUEST, "invalid track id")); + } + + let mut source = load_track_items_by_ids(pool, &[path.id]).await?; + let Some(source_track) = source.pop() else { + return Ok(json_error(StatusCode::NOT_FOUND, "local track not found")); + }; + let manager = crate::similarity::handle(); + let query = match manager.query_for_track(path.id).await { + Ok(query) => query, + Err(error) => { + return Ok(json_error( + StatusCode::SERVICE_UNAVAILABLE, + &format!("similarity search is not ready: {error:#}"), + )); + } + }; + let rank_manager = std::sync::Arc::clone(&manager); + let profile_id = query.profile_id.clone(); + let vector = query.vector.clone(); + let source_content_id = query.source_content_id.clone(); + let ranked = match tokio::task::spawn_blocking(move || { + rank_manager.rank_vector( + &profile_id, + &vector, + Some(path.id), + source_content_id.as_deref(), + 49, + ) + }) + .await + { + Ok(Ok(ranked)) => ranked, + Ok(Err(error)) => { + return Ok(json_error( + StatusCode::SERVICE_UNAVAILABLE, + &format!("similarity search failed: {error:#}"), + )); + } + Err(error) => { + return Ok(json_error( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("similarity worker failed: {error}"), + )); + } + }; + + let ids = ranked + .iter() + .map(|track| track.track_id) + .collect::>(); + let mut tracks = Vec::with_capacity(ids.len() + 1); + tracks.push(source_track.clone()); + tracks.extend(load_track_items_by_ids(pool, &ids).await?); + + let (config, _) = AppConfig::load_with_db(&db).await; + let (federation_tracks, federation_error) = if config.federation_enabled { + match crate::federation::handle() + .search_similarity(query, 50) + .await + { + Ok(remote) => match crate::federation::handle() + .prepare_similarity_tracks(remote) + .await + { + Ok(tracks) => (tracks, None), + Err(error) => (Vec::new(), Some(format!("{error:#}"))), + }, + Err(error) => (Vec::new(), Some(format!("{error:#}"))), + } + } else { + (Vec::new(), None) + }; + let artists = source_track + .artists + .iter() + .map(|artist| artist.name.as_str()) + .collect::>() + .join(", "); + let label = if artists.is_empty() { + source_track.title.clone() + } else { + format!("{} — {artists}", source_track.title) + }; + Json(SimilaritySearchResponse { + label, + tracks, + federation_tracks, + federation_error, + }) + .into_response() +} + // --------------------------------------------------------------------------- // POST /api/player/share-playlist // --------------------------------------------------------------------------- @@ -9813,6 +9926,33 @@ impl App for PlayerApp { }), "player_search", ), + Route::with_handler_and_name( + "/similarity/{id}", + get({ + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + move |auth_ctx: auth::AuthContext, + session: Session, + db: Database, + path: Path| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + async move { + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + similarity_search_handler(auth_ctx, session, db, pg_pool, path).await + } + } + }), + "player_similarity_search", + ), Route::with_handler_and_name( "/federation/search/events", get( diff --git a/src/similarity.rs b/src/similarity.rs new file mode 100644 index 0000000..6419129 --- /dev/null +++ b/src/similarity.rs @@ -0,0 +1,1323 @@ +//! Local, server-wide music embeddings and exact cosine search. +//! +//! PostgreSQL is the durable source of truth. The active profile is mirrored +//! into a replaceable in-memory index so ordinary searches do not require a +//! vector extension or a second database. + +use std::collections::{HashMap, HashSet}; +use std::fs::File; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, OnceLock, RwLock}; +use std::time::{Duration, Instant}; + +use anyhow::{Context as _, Result}; +use futures_util::StreamExt as _; +use rodio::{Decoder, Source as _}; +use rustfft::FftPlanner; +use rustfft::num_complex::Complex; +use serde::Serialize; +use sha2::{Digest as _, Sha256}; +use sqlx::{PgPool, Row as _}; +use tokio::io::AsyncWriteExt as _; +use tract_onnx::prelude::*; +use tract_onnx::tract_core::dims; + +use crate::config::AppConfig; + +const SAMPLE_RATE: usize = 16_000; +const FRAME_SIZE: usize = 512; +const HOP_SIZE: usize = 256; +const MEL_BANDS: usize = 96; +const PATCH_FRAMES: usize = 128; +const PATCH_HOP: usize = 62; +const EMBEDDING_DIMENSIONS: usize = 1280; +const MODEL_BATCH: usize = 8; +const MAX_MODEL_BYTES: usize = 64 * 1024 * 1024; +const RESULT_LIMIT: usize = 50; +const MAX_PER_ARTIST: usize = 3; +const NEAR_DUPLICATE_COSINE: f32 = 0.995; +const FULL_TRACK_MAX_SECONDS: u32 = 5 * 60; +const LONG_TRACK_WINDOW_SECONDS: u32 = 60; +const PIPELINE_POLL_INTERVAL: Duration = Duration::from_secs(30); + +pub const DEFAULT_MODEL_ID: &str = "discogs-effnet-bsdynamic-1"; +pub const DEFAULT_PROFILE_ID: &str = "furumi-full-track-v1"; + +#[derive(Debug, Clone, Copy, Serialize)] +pub struct ProfileSpec { + pub id: &'static str, + pub title: &'static str, +} + +pub const PROFILES: &[ProfileSpec] = &[ProfileSpec { + id: DEFAULT_PROFILE_ID, + title: "Full track / balanced long track", +}]; + +#[derive(Debug, Clone, Copy, Serialize)] +pub struct ModelSpec { + pub id: &'static str, + pub version: &'static str, + pub filename: &'static str, + pub url: &'static str, + pub sha256: &'static str, + pub dimensions: usize, + pub license: &'static str, +} + +pub const MODELS: &[ModelSpec] = &[ModelSpec { + id: DEFAULT_MODEL_ID, + version: "1", + filename: "discogs-effnet-bsdynamic-1.onnx", + url: "https://essentia.upf.edu/models/feature-extractors/discogs-effnet/discogs-effnet-bsdynamic-1.onnx", + sha256: "a280825b334797cf677939db8cd5762c0392aedd0ca6415dbc1cd083f045e43c", + dimensions: EMBEDDING_DIMENSIONS, + license: "CC BY-NC-SA 4.0 (or proprietary from MTG)", +}]; + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Phase { + #[default] + Disabled, + Downloading, + Loading, + Processing, + Ready, + Error, +} + +#[derive(Debug, Clone, Default, Serialize)] +pub struct SimilarityStatus { + pub phase: Phase, + pub active_profile: Option, + pub target_profile: Option, + pub model: String, + pub total_tracks: usize, + pub completed_tracks: usize, + pub failed_tracks: usize, + pub stored_vectors: usize, + pub stored_bytes: u64, + pub current_track: Option, + pub last_error: Option, +} + +#[derive(Debug, Clone)] +pub struct QueryVector { + pub profile_id: String, + pub vector: Vec, + pub source_content_id: Option, +} + +#[derive(Debug, Clone)] +pub struct RankedTrack { + pub track_id: i64, + pub score: f32, + pub embedding_signature: [u8; music_dht::similarity::SIMILARITY_SIGNATURE_BYTES], +} + +#[derive(Debug, Clone)] +pub struct TrackMetadata { + pub track_id: i64, + pub title: String, + pub artist_names: Vec, + pub featured_artist_names: Vec, + pub year: Option, + pub duration_seconds: f64, + pub content_id: Option, + pub release_title: String, + pub track_number: Option, + pub disc_number: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Settings { + enabled: bool, + model: String, + profile: String, + workers: usize, +} + +impl Settings { + fn from_config(config: &AppConfig) -> Self { + Self { + enabled: config.similarity_enabled, + model: config.similarity_model.clone(), + profile: config.similarity_profile.clone(), + workers: (config.similarity_workers as usize).clamp(1, 16), + } + } +} + +impl Default for Settings { + fn default() -> Self { + Self { + enabled: false, + model: DEFAULT_MODEL_ID.to_owned(), + profile: DEFAULT_PROFILE_ID.to_owned(), + workers: 1, + } + } +} + +#[derive(Debug, Clone)] +struct SimilarityTrack { + id: i64, + title: String, + file_path: PathBuf, + source_sha256: String, + source_content_id: Option, + duration_seconds: f64, +} + +#[derive(Debug, Clone)] +struct StoredEmbedding { + track_id: i64, + vector: Vec, + artist_key: String, + content_id: Option, +} + +#[derive(Default)] +struct Index { + profile_id: Option, + entries: Vec, +} + +#[derive(Debug, Default)] +struct StorageStats { + total_tracks: usize, + embedded_tracks: usize, + stored_vectors: usize, + stored_bytes: u64, +} + +type RunnableModel = Arc; + +pub struct Manager { + database_url: Mutex, + storage_dir: Mutex, + pool: tokio::sync::OnceCell, + settings: Mutex, + workers: AtomicUsize, + generation: AtomicU64, + status: Mutex, + index: RwLock, + model: Mutex>, + model_dir: PathBuf, +} + +pub fn handle() -> Arc { + static HANDLE: OnceLock> = OnceLock::new(); + Arc::clone(HANDLE.get_or_init(|| { + Arc::new(Manager { + database_url: Mutex::new(String::new()), + storage_dir: Mutex::new(String::new()), + pool: tokio::sync::OnceCell::new(), + settings: Mutex::new(Settings::default()), + workers: AtomicUsize::new(1), + generation: AtomicU64::new(0), + status: Mutex::new(SimilarityStatus::default()), + index: RwLock::new(Index::default()), + model: Mutex::new(None), + model_dir: PathBuf::from(crate::media_paths::resolve_config_path("similarity-models")), + }) + })) +} + +impl Manager { + pub async fn boot(self: &Arc, config: &AppConfig) { + *lock(&self.database_url) = config.database_url.clone(); + *lock(&self.storage_dir) = config.agent_storage_dir.clone(); + if config.database_url.trim().is_empty() { + return; + } + let pool = match self.pool().await { + Ok(pool) => pool, + Err(error) => { + tracing::warn!(%error, "similarity boot: database unavailable"); + self.update_status(|status| { + status.phase = Phase::Error; + status.last_error = Some(format!("database unavailable: {error}")); + }); + return; + } + }; + if let Err(error) = self.restore_stored_status(&pool).await { + tracing::warn!(%error, "similarity boot: stored status unavailable"); + } + self.apply(config); + } + + pub fn apply(self: &Arc, config: &AppConfig) { + *lock(&self.database_url) = config.database_url.clone(); + *lock(&self.storage_dir) = config.agent_storage_dir.clone(); + let settings = Settings::from_config(config); + self.workers.store(settings.workers, Ordering::Release); + let previous = std::mem::replace(&mut *lock(&self.settings), settings.clone()); + self.update_status(|status| status.model = settings.model.clone()); + if !settings.enabled { + self.generation.fetch_add(1, Ordering::AcqRel); + self.update_status(|status| { + status.phase = Phase::Disabled; + status.target_profile = None; + status.current_track = None; + status.last_error = None; + }); + return; + } + if !previous.enabled + || previous.model != settings.model + || previous.profile != settings.profile + { + self.start(); + } + } + + pub fn enabled(&self) -> bool { + lock(&self.settings).enabled + } + + pub fn status(&self) -> SimilarityStatus { + lock(&self.status).clone() + } + + pub fn start(self: &Arc) { + let generation = self.generation.fetch_add(1, Ordering::AcqRel) + 1; + let manager = Arc::clone(self); + tokio::spawn(async move { + if let Err(error) = manager.run_pipeline(generation).await + && manager.generation.load(Ordering::Acquire) == generation + { + tracing::error!(%error, "similarity pipeline failed"); + manager.update_status(|status| { + status.phase = Phase::Error; + status.current_track = None; + status.last_error = Some(format!("{error:#}")); + }); + } + }); + } + + pub async fn clear(self: &Arc) -> Result<()> { + self.generation.fetch_add(1, Ordering::AcqRel); + let pool = self.pool().await?; + sqlx::query("DELETE FROM furumusic__similarity_profile") + .execute(&pool) + .await?; + *write(&self.index) = Index::default(); + self.update_status(|status| { + *status = SimilarityStatus { + phase: if self.enabled() { + Phase::Loading + } else { + Phase::Disabled + }, + model: lock(&self.settings).model.clone(), + ..SimilarityStatus::default() + }; + }); + if self.enabled() { + self.start(); + } + Ok(()) + } + + pub async fn query_for_track(&self, track_id: i64) -> Result { + anyhow::ensure!(self.enabled(), "similarity search is disabled"); + let profile_id = read(&self.index) + .profile_id + .clone() + .context("no similarity profile is ready yet")?; + let pool = self.pool().await?; + let row = sqlx::query( + "SELECT e.dimensions, e.vector, c.content_id + FROM furumusic__track_embedding e + JOIN furumusic__track t ON t.id = e.track_id + JOIN furumusic__release r ON r.id = t.release_id + JOIN furumusic__media_file m ON m.id = t.audio_file_id + LEFT JOIN furumusic__federation_content_id_cache c + ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash + WHERE e.track_id = $1 AND e.profile_id = $2 + AND e.source_sha256 = m.sha256_hash + AND t.is_hidden = FALSE AND r.is_hidden = FALSE", + ) + .bind(track_id) + .bind(&profile_id) + .fetch_optional(&pool) + .await? + .context("this track has not been processed yet")?; + let dimensions: i32 = row.get(0); + let bytes: Vec = row.get(1); + let source_content_id: Option = row.get(2); + Ok(QueryVector { + profile_id, + vector: embedding_from_bytes(dimensions, &bytes)?, + source_content_id, + }) + } + + pub fn rank_vector( + &self, + profile_id: &str, + vector: &[f32], + exclude_track_id: Option, + exclude_content_id: Option<&str>, + limit: usize, + ) -> Result> { + anyhow::ensure!( + !vector.is_empty() && vector.len() <= 4096, + "wrong embedding dimensions" + ); + anyhow::ensure!( + vector.iter().all(|value| value.is_finite()), + "invalid embedding" + ); + let index = read(&self.index); + anyhow::ensure!( + index.profile_id.as_deref() == Some(profile_id), + "the requested similarity profile is not active" + ); + let mut scores: Vec<(&StoredEmbedding, f32)> = index + .entries + .iter() + .filter(|entry| { + Some(entry.track_id) != exclude_track_id + && entry.vector.len() == vector.len() + && !exclude_content_id + .is_some_and(|source| entry.content_id.as_deref() == Some(source)) + }) + .map(|entry| (entry, dot(vector, &entry.vector))) + .filter(|(_, score)| score.is_finite()) + .collect(); + scores.sort_by(|left, right| right.1.total_cmp(&left.1)); + + let mut artist_counts: HashMap = HashMap::new(); + let mut kept_vectors: Vec<&[f32]> = vec![vector]; + let mut selected = Vec::new(); + for (entry, score) in scores { + if is_near_duplicate(&entry.vector, &kept_vectors) { + continue; + } + let count = artist_counts.entry(entry.artist_key.clone()).or_default(); + if !entry.artist_key.is_empty() && *count >= MAX_PER_ARTIST { + continue; + } + *count += 1; + let embedding_signature = music_dht::similarity::embedding_signature(&entry.vector)?; + kept_vectors.push(&entry.vector); + selected.push(RankedTrack { + track_id: entry.track_id, + score, + embedding_signature, + }); + if selected.len() >= limit.clamp(1, RESULT_LIMIT) { + break; + } + } + Ok(selected) + } + + pub async fn metadata_for_tracks(&self, ids: &[i64]) -> Result> { + if ids.is_empty() { + return Ok(Vec::new()); + } + let pool = self.pool().await?; + let rows = sqlx::query( + "SELECT t.id, t.title::text, COALESCE(t.year, r.year), + t.duration_seconds, c.content_id, r.title::text, + t.track_number, t.disc_number, + COALESCE(array_agg(a.name::text ORDER BY ta.position) + FILTER (WHERE ta.role = 'main'), ARRAY[]::text[]), + COALESCE(array_agg(a.name::text ORDER BY ta.position) + FILTER (WHERE ta.role = 'featuring'), ARRAY[]::text[]) + FROM furumusic__track t + JOIN furumusic__release r ON r.id = t.release_id + JOIN furumusic__media_file m ON m.id = t.audio_file_id + LEFT JOIN furumusic__federation_content_id_cache c + ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash + LEFT JOIN furumusic__track_artist ta ON ta.track_id = t.id + LEFT JOIN furumusic__artist a ON a.id = ta.artist_id + WHERE t.id = ANY($1) AND t.is_hidden = FALSE AND r.is_hidden = FALSE + GROUP BY t.id, r.id, c.content_id", + ) + .bind(ids) + .fetch_all(&pool) + .await?; + let by_id: HashMap = rows + .into_iter() + .map(|row| { + let track = TrackMetadata { + track_id: row.get(0), + title: row.get(1), + year: row.get(2), + duration_seconds: row.get(3), + content_id: row.get(4), + release_title: row.get(5), + track_number: row.get(6), + disc_number: row.get(7), + artist_names: row.get(8), + featured_artist_names: row.get(9), + }; + (track.track_id, track) + }) + .collect(); + Ok(ids.iter().filter_map(|id| by_id.get(id).cloned()).collect()) + } + + async fn run_pipeline(self: &Arc, generation: u64) -> Result<()> { + let settings = lock(&self.settings).clone(); + if !settings.enabled { + return Ok(()); + } + let spec = model_by_id(&settings.model) + .with_context(|| format!("unknown similarity model '{}'", settings.model))?; + anyhow::ensure!( + profile_by_id(&settings.profile).is_some(), + "unknown preprocessing profile '{}'", + settings.profile + ); + let profile_id = profile_fingerprint(spec, &settings.profile); + let pool = self.pool().await?; + self.restore_active_index(&pool).await?; + ensure_similarity_profile(&pool, &profile_id, spec, &settings.profile).await?; + + let stats = storage_stats(&pool, &profile_id).await?; + self.update_status(|status| { + status.phase = Phase::Downloading; + status.target_profile = Some(profile_id.clone()); + status.model = spec.id.to_owned(); + status.total_tracks = stats.total_tracks; + status.completed_tracks = stats.embedded_tracks; + status.failed_tracks = 0; + status.stored_vectors = stats.stored_vectors; + status.stored_bytes = stats.stored_bytes; + status.current_track = None; + status.last_error = None; + }); + let model_path = self.ensure_model(spec, generation).await?; + self.ensure_generation(generation)?; + self.update_status(|status| status.phase = Phase::Loading); + let model = self.load_model(&profile_id, &model_path).await?; + self.ensure_generation(generation)?; + + let mut failures: HashSet<(i64, String)> = HashSet::new(); + loop { + self.ensure_generation(generation)?; + let storage_dir = lock(&self.storage_dir).clone(); + let mut pending = pending_tracks(&pool, &profile_id, &storage_dir).await?; + pending.retain(|track| !failures.contains(&(track.id, track.source_sha256.clone()))); + if !pending.is_empty() { + self.update_status(|status| status.phase = Phase::Processing); + let mut queue: std::collections::VecDeque<_> = pending.into(); + let mut jobs = tokio::task::JoinSet::new(); + while !queue.is_empty() || !jobs.is_empty() { + self.ensure_generation(generation)?; + let workers = self.workers.load(Ordering::Acquire).clamp(1, 16); + while jobs.len() < workers { + let Some(track) = queue.pop_front() else { + break; + }; + self.update_status(|status| { + status.current_track = Some(track.title.clone()) + }); + let model = Arc::clone(&model); + jobs.spawn_blocking(move || { + let started = Instant::now(); + let result = + embed_track(&model, &track.file_path, track.duration_seconds); + (track, result, started.elapsed()) + }); + } + let Some(result) = jobs.join_next().await else { + continue; + }; + let (track, result, elapsed) = result.context("embedding worker panicked")?; + self.ensure_generation(generation)?; + match result { + Ok(vector) => { + store_embedding(&pool, &track, &profile_id, &vector).await?; + tracing::info!( + track_id = track.id, + title = %track.title, + elapsed_ms = elapsed.as_millis(), + profile = %profile_id, + "track embedding calculated" + ); + self.update_status(|status| status.completed_tracks += 1); + } + Err(error) => { + tracing::warn!( + track_id = track.id, + title = %track.title, + %error, + "track embedding failed" + ); + failures.insert((track.id, track.source_sha256.clone())); + self.update_status(|status| { + status.failed_tracks += 1; + status.last_error = Some(format!("{}: {error:#}", track.title)); + }); + } + } + } + } + + self.ensure_generation(generation)?; + let entries = load_index(&pool, &profile_id).await?; + let stats = storage_stats(&pool, &profile_id).await?; + anyhow::ensure!( + stats.total_tracks == 0 || !entries.is_empty(), + "no visible tracks could be processed" + ); + activate_profile(&pool, &profile_id).await?; + *write(&self.index) = Index { + profile_id: Some(profile_id.clone()), + entries, + }; + self.update_status(|status| { + status.phase = Phase::Ready; + status.active_profile = Some(profile_id.clone()); + status.target_profile = Some(profile_id.clone()); + status.total_tracks = stats.total_tracks; + status.completed_tracks = stats.embedded_tracks; + status.stored_vectors = stats.stored_vectors; + status.stored_bytes = stats.stored_bytes; + status.current_track = None; + }); + tokio::time::sleep(PIPELINE_POLL_INTERVAL).await; + } + } + + async fn restore_active_index(&self, pool: &PgPool) -> Result<()> { + let active: Option = sqlx::query_scalar( + "SELECT profile_id FROM furumusic__similarity_profile + WHERE active = TRUE LIMIT 1", + ) + .fetch_optional(pool) + .await?; + let Some(profile_id) = active else { + return Ok(()); + }; + if read(&self.index).profile_id.as_deref() == Some(&profile_id) { + return Ok(()); + } + let entries = load_index(pool, &profile_id).await?; + *write(&self.index) = Index { + profile_id: Some(profile_id.clone()), + entries, + }; + self.update_status(|status| status.active_profile = Some(profile_id)); + Ok(()) + } + + async fn restore_stored_status(&self, pool: &PgPool) -> Result<()> { + let active_profile: Option = sqlx::query_scalar( + "SELECT profile_id FROM furumusic__similarity_profile + WHERE active = TRUE LIMIT 1", + ) + .fetch_optional(pool) + .await?; + let stats = storage_stats(pool, active_profile.as_deref().unwrap_or_default()).await?; + self.update_status(|status| { + status.active_profile = active_profile; + status.total_tracks = stats.total_tracks; + status.completed_tracks = stats.embedded_tracks; + status.stored_vectors = stats.stored_vectors; + status.stored_bytes = stats.stored_bytes; + }); + Ok(()) + } + + fn ensure_generation(&self, generation: u64) -> Result<()> { + anyhow::ensure!( + self.generation.load(Ordering::Acquire) == generation, + "similarity processing superseded by newer settings" + ); + Ok(()) + } + + async fn ensure_model(&self, spec: &ModelSpec, generation: u64) -> Result { + tokio::fs::create_dir_all(&self.model_dir).await?; + let path = self.model_dir.join(spec.filename); + if path.exists() { + let verify_path = path.clone(); + let expected = spec.sha256.to_owned(); + let valid = tokio::task::spawn_blocking(move || sha256_file(&verify_path)) + .await + .context("model hash task failed")?? + == expected; + if valid { + return Ok(path); + } + tokio::fs::remove_file(&path).await?; + } + + let response = reqwest::get(spec.url).await?.error_for_status()?; + let temporary = path.with_extension(format!("part-{}-{generation}", std::process::id())); + let mut file = tokio::fs::File::create(&temporary).await?; + let mut hasher = Sha256::new(); + let mut received = 0usize; + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + self.ensure_generation(generation)?; + let chunk = chunk?; + received = received.saturating_add(chunk.len()); + anyhow::ensure!( + received <= MAX_MODEL_BYTES, + "model download exceeds size limit" + ); + hasher.update(&chunk); + file.write_all(&chunk).await?; + } + file.flush().await?; + drop(file); + let actual = format!("{:x}", hasher.finalize()); + if actual != spec.sha256 { + let _ = tokio::fs::remove_file(&temporary).await; + anyhow::bail!("downloaded model hash mismatch"); + } + if let Err(error) = tokio::fs::rename(&temporary, &path).await { + if path.exists() { + let _ = tokio::fs::remove_file(&temporary).await; + } else { + return Err(error.into()); + } + } + Ok(path) + } + + async fn load_model(&self, profile_id: &str, path: &Path) -> Result { + if let Some((cached_profile, model)) = lock(&self.model).as_ref() + && cached_profile == profile_id + { + return Ok(Arc::clone(model)); + } + let path = path.to_path_buf(); + let model = tokio::task::spawn_blocking(move || load_onnx(&path)) + .await + .context("model loading task failed")??; + *lock(&self.model) = Some((profile_id.to_owned(), Arc::clone(&model))); + Ok(model) + } + + async fn pool(&self) -> Result { + let url = lock(&self.database_url).clone(); + anyhow::ensure!(!url.trim().is_empty(), "database is not configured"); + let pool = self + .pool + .get_or_try_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(8) + .connect(&url) + .await + }) + .await?; + Ok(pool.clone()) + } + + fn update_status(&self, update: impl FnOnce(&mut SimilarityStatus)) { + update(&mut lock(&self.status)); + } +} + +pub fn model_by_id(id: &str) -> Option<&'static ModelSpec> { + MODELS.iter().find(|model| model.id == id) +} + +pub fn profile_by_id(id: &str) -> Option<&'static ProfileSpec> { + PROFILES.iter().find(|profile| profile.id == id) +} + +pub fn profile_details(profile_id: &str, model_id: &str) -> Option { + let profile = profile_by_id(profile_id)?; + let dimensions = model_by_id(model_id) + .map(|model| model.dimensions.to_string()) + .unwrap_or_else(|| "model-defined".to_owned()); + Some(format!( + "{}\n\nTrack selection:\n• Up to {} seconds: entire track.\n• Longer: 3 × {}-second windows (start, middle, end).\n\nAudio: mono, {} Hz; 16-tap windowed-sinc resampling.\nSpectrogram: Hann window; FFT {}, hop {} samples (16 ms).\nMel: {} Slaney bands, 0–8 kHz, unit-triangle normalization.\nCompression: log10(1 + 10000 × energy).\nPatches: {} frames (~2.05 s), hop {} frames (~0.99 s).\nAggregation: mean of patch embeddings, then L2 normalization.\nOutput dimensions for selected model: {}.\n\nCompatibility includes the exact model version and SHA-256; peers compare only matching profiles.", + profile.title, + FULL_TRACK_MAX_SECONDS, + LONG_TRACK_WINDOW_SECONDS, + SAMPLE_RATE, + FRAME_SIZE, + HOP_SIZE, + MEL_BANDS, + PATCH_FRAMES, + PATCH_HOP, + dimensions, + )) +} + +pub fn profile_fingerprint(model: &ModelSpec, profile: &str) -> String { + let contract = format!( + "furumi-similarity-v1\nmodel={}\nversion={}\nsha256={}\nprofile={}\ninput={}-mono-windowed-sinc16\nselection=full-to-{}s-else-first-middle-last-{}s\nframe={}\nhop={}\nmel=slaney-{}-unit-tri\npatch={}\npatch-hop={}\naggregate=mean-l2\ndimensions={}", + model.id, + model.version, + model.sha256, + profile, + SAMPLE_RATE, + FULL_TRACK_MAX_SECONDS, + LONG_TRACK_WINDOW_SECONDS, + FRAME_SIZE, + HOP_SIZE, + MEL_BANDS, + PATCH_FRAMES, + PATCH_HOP, + model.dimensions + ); + format!("sim1:{}", blake3::hash(contract.as_bytes()).to_hex()) +} + +async fn ensure_similarity_profile( + pool: &PgPool, + profile_id: &str, + model: &ModelSpec, + preprocessing: &str, +) -> Result<()> { + sqlx::query( + "INSERT INTO furumusic__similarity_profile + (profile_id, model_id, model_version, model_sha256, + preprocessing, dimensions, active, created_at) + VALUES ($1, $2, $3, $4, $5, $6, FALSE, $7) + ON CONFLICT (profile_id) DO NOTHING", + ) + .bind(profile_id) + .bind(model.id) + .bind(model.version) + .bind(model.sha256) + .bind(preprocessing) + .bind(model.dimensions as i32) + .bind(now_iso()) + .execute(pool) + .await?; + Ok(()) +} + +async fn pending_tracks( + pool: &PgPool, + profile_id: &str, + storage_dir: &str, +) -> Result> { + let rows = sqlx::query( + "SELECT t.id, t.title::text, m.file_path, m.sha256_hash::text, + c.content_id, t.duration_seconds + FROM furumusic__track t + JOIN furumusic__release r ON r.id = t.release_id + JOIN furumusic__media_file m ON m.id = t.audio_file_id + LEFT JOIN furumusic__federation_content_id_cache c + ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash + WHERE t.is_hidden = FALSE AND r.is_hidden = FALSE + AND NOT EXISTS ( + SELECT 1 FROM furumusic__track_embedding e + WHERE e.track_id = t.id AND e.profile_id = $1 + AND e.source_sha256 = m.sha256_hash + ) + ORDER BY t.id", + ) + .bind(profile_id) + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .map(|row| SimilarityTrack { + id: row.get(0), + title: row.get(1), + file_path: crate::media_paths::resolve_media_file_path( + storage_dir, + row.get::(2).as_str(), + ), + source_sha256: row.get(3), + source_content_id: row.get(4), + duration_seconds: row.get(5), + }) + .collect()) +} + +async fn store_embedding( + pool: &PgPool, + track: &SimilarityTrack, + profile_id: &str, + vector: &[f32], +) -> Result<()> { + anyhow::ensure!(!vector.is_empty(), "embedding vector is empty"); + anyhow::ensure!( + vector.iter().all(|value| value.is_finite()), + "embedding contains a non-finite value" + ); + sqlx::query( + "INSERT INTO furumusic__track_embedding + (track_id, profile_id, dimensions, vector, source_sha256, + source_content_id, computed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (track_id, profile_id) DO UPDATE SET + dimensions = EXCLUDED.dimensions, + vector = EXCLUDED.vector, + source_sha256 = EXCLUDED.source_sha256, + source_content_id = EXCLUDED.source_content_id, + computed_at = EXCLUDED.computed_at", + ) + .bind(track.id) + .bind(profile_id) + .bind(vector.len() as i32) + .bind(embedding_to_bytes(vector)) + .bind(&track.source_sha256) + .bind(&track.source_content_id) + .bind(now_iso()) + .execute(pool) + .await?; + Ok(()) +} + +async fn load_index(pool: &PgPool, profile_id: &str) -> Result> { + let rows = sqlx::query( + "SELECT e.track_id, e.dimensions, e.vector, + COALESCE(( + SELECT a.name::text + FROM furumusic__track_artist ta + JOIN furumusic__artist a ON a.id = ta.artist_id + WHERE ta.track_id = e.track_id AND ta.role = 'main' + ORDER BY ta.position LIMIT 1 + ), ''), c.content_id + FROM furumusic__track_embedding e + JOIN furumusic__track t ON t.id = e.track_id + JOIN furumusic__release r ON r.id = t.release_id + JOIN furumusic__media_file m ON m.id = t.audio_file_id + LEFT JOIN furumusic__federation_content_id_cache c + ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash + WHERE e.profile_id = $1 AND e.source_sha256 = m.sha256_hash + AND t.is_hidden = FALSE AND r.is_hidden = FALSE + ORDER BY e.track_id", + ) + .bind(profile_id) + .fetch_all(pool) + .await?; + rows.into_iter() + .map(|row| { + let dimensions: i32 = row.get(1); + let bytes: Vec = row.get(2); + Ok(StoredEmbedding { + track_id: row.get(0), + vector: embedding_from_bytes(dimensions, &bytes)?, + artist_key: music_dht::normalize_name(&row.get::(3)), + content_id: row.get(4), + }) + }) + .collect() +} + +async fn storage_stats(pool: &PgPool, profile_id: &str) -> Result { + let total_tracks: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM furumusic__track t + JOIN furumusic__release r ON r.id = t.release_id + WHERE t.is_hidden = FALSE AND r.is_hidden = FALSE", + ) + .fetch_one(pool) + .await?; + let embedded_tracks: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM furumusic__track_embedding e + JOIN furumusic__track t ON t.id = e.track_id + JOIN furumusic__release r ON r.id = t.release_id + JOIN furumusic__media_file m ON m.id = t.audio_file_id + WHERE e.profile_id = $1 AND e.source_sha256 = m.sha256_hash + AND t.is_hidden = FALSE AND r.is_hidden = FALSE", + ) + .bind(profile_id) + .fetch_one(pool) + .await?; + let row = sqlx::query( + "SELECT COUNT(*), COALESCE(SUM(octet_length(vector)), 0) + FROM furumusic__track_embedding", + ) + .fetch_one(pool) + .await?; + let stored_vectors: i64 = row.get(0); + let stored_bytes: i64 = row.get(1); + Ok(StorageStats { + total_tracks: total_tracks.max(0) as usize, + embedded_tracks: embedded_tracks.max(0) as usize, + stored_vectors: stored_vectors.max(0) as usize, + stored_bytes: stored_bytes.max(0) as u64, + }) +} + +async fn activate_profile(pool: &PgPool, profile_id: &str) -> Result<()> { + let mut transaction = pool.begin().await?; + sqlx::query("UPDATE furumusic__similarity_profile SET active = FALSE WHERE active = TRUE") + .execute(&mut *transaction) + .await?; + sqlx::query("UPDATE furumusic__similarity_profile SET active = TRUE WHERE profile_id = $1") + .bind(profile_id) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + Ok(()) +} + +fn embedding_to_bytes(vector: &[f32]) -> Vec { + vector + .iter() + .flat_map(|value| value.to_le_bytes()) + .collect() +} + +fn embedding_from_bytes(dimensions: i32, bytes: &[u8]) -> Result> { + let dimensions = usize::try_from(dimensions).context("negative embedding dimensions")?; + anyhow::ensure!( + dimensions > 0 && dimensions <= 4096 && bytes.len() == dimensions * 4, + "invalid stored embedding dimensions" + ); + Ok(bytes + .chunks_exact(4) + .map(|bytes| f32::from_le_bytes(bytes.try_into().expect("four-byte chunk"))) + .collect()) +} + +fn now_iso() -> String { + chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string() +} + +fn load_onnx(path: &Path) -> Result { + let model = tract_onnx::onnx().model_for_path(path)?; + let batch = model.sym("batch_size"); + let model = model + .with_input_fact(0, f32::fact(dims!(batch, PATCH_FRAMES, MEL_BANDS)).into())? + .into_optimized()? + .into_runnable()?; + Ok(model) +} + +fn embed_track(model: &RunnableModel, path: &Path, duration_seconds: f64) -> Result> { + let signal = decode_mono_16k(path, duration_seconds)?; + let mel = mel_spectrogram(&signal)?; + anyhow::ensure!( + mel.len() >= PATCH_FRAMES, + "track is too short for the model" + ); + let starts: Vec = (0..=mel.len() - PATCH_FRAMES).step_by(PATCH_HOP).collect(); + let mut sum = vec![0.0f32; EMBEDDING_DIMENSIONS]; + let mut count = 0usize; + for batch in starts.chunks(MODEL_BATCH) { + let mut input = vec![0.0f32; MODEL_BATCH * PATCH_FRAMES * MEL_BANDS]; + for (batch_index, &start) in batch.iter().enumerate() { + let offset = batch_index * PATCH_FRAMES * MEL_BANDS; + for frame in 0..PATCH_FRAMES { + let destination = offset + frame * MEL_BANDS; + input[destination..destination + MEL_BANDS].copy_from_slice(&mel[start + frame]); + } + } + let tensor = Tensor::from_shape(&[MODEL_BATCH, PATCH_FRAMES, MEL_BANDS], &input)?; + let outputs = model.run(tvec!(tensor.into_tvalue()))?; + let embedding = outputs + .iter() + .find(|output| output.len() == MODEL_BATCH * EMBEDDING_DIMENSIONS) + .context("model did not return its 1280-dimensional embedding output")? + .to_plain_array_view::()?; + let values = embedding + .as_slice() + .context("model embedding output is not contiguous")?; + for batch_index in 0..batch.len() { + let row = &values + [batch_index * EMBEDDING_DIMENSIONS..(batch_index + 1) * EMBEDDING_DIMENSIONS]; + for (total, value) in sum.iter_mut().zip(row) { + *total += *value; + } + count += 1; + } + } + anyhow::ensure!(count > 0, "model produced no patches"); + for value in &mut sum { + *value /= count as f32; + } + normalize(&mut sum)?; + Ok(sum) +} + +fn decode_mono_16k(path: &Path, duration_seconds: f64) -> Result> { + if duration_seconds.is_finite() && duration_seconds > f64::from(FULL_TRACK_MAX_SECONDS) { + let window = f64::from(LONG_TRACK_WINDOW_SECONDS); + let starts = [ + 0.0, + (duration_seconds / 2.0 - window / 2.0).max(0.0), + (duration_seconds - window).max(0.0), + ]; + let mut selected = Vec::new(); + for start in starts { + selected.extend(decode_mono_window(path, start, Some(window))?); + } + anyhow::ensure!(!selected.is_empty(), "decoded track is empty"); + return Ok(selected); + } + decode_mono_window(path, 0.0, None) +} + +fn decode_mono_window( + path: &Path, + start_seconds: f64, + length_seconds: Option, +) -> Result> { + let file = File::open(path).with_context(|| format!("opening {}", path.display()))?; + let mut decoder = + Decoder::try_from(file).with_context(|| format!("decoding {}", path.display()))?; + let channels = decoder.channels().get() as usize; + let source_rate = decoder.sample_rate().get() as usize; + if start_seconds > 0.0 { + decoder + .try_seek(Duration::from_secs_f64(start_seconds)) + .with_context(|| format!("seeking {}", path.display()))?; + } + let max_samples = + length_seconds.map(|seconds| (seconds * source_rate as f64).ceil() as usize * channels); + let mut mono = Vec::new(); + let mut channel_sum = 0.0f32; + let mut channel_index = 0usize; + for (sample_index, sample) in decoder.enumerate() { + if max_samples.is_some_and(|limit| sample_index >= limit) { + break; + } + channel_sum += sample; + channel_index += 1; + if channel_index == channels { + mono.push(channel_sum / channels as f32); + channel_sum = 0.0; + channel_index = 0; + } + } + anyhow::ensure!(!mono.is_empty(), "decoded track is empty"); + if source_rate == SAMPLE_RATE { + return Ok(mono); + } + Ok(resample_sinc(&mono, source_rate, SAMPLE_RATE)) +} + +fn resample_sinc(input: &[f32], source_rate: usize, target_rate: usize) -> Vec { + if input.len() < 2 || source_rate == 0 { + return input.to_vec(); + } + let output_len = input + .len() + .saturating_mul(target_rate) + .checked_div(source_rate) + .unwrap_or(0) + .max(1); + let ratio = source_rate as f64 / target_rate as f64; + let cutoff = (target_rate as f64 / source_rate as f64).min(1.0) * 0.95; + const HALF_TAPS: isize = 8; + (0..output_len) + .map(|index| { + let position = index as f64 * ratio; + let center = position.floor() as isize; + let mut value = 0.0f64; + let mut weight_sum = 0.0f64; + for sample_index in center - HALF_TAPS + 1..=center + HALF_TAPS { + if sample_index < 0 || sample_index >= input.len() as isize { + continue; + } + let distance = position - sample_index as f64; + let phase = std::f64::consts::PI * distance * cutoff; + let sinc = if phase.abs() < 1e-12 { + 1.0 + } else { + phase.sin() / phase + }; + let window_position = distance / HALF_TAPS as f64; + let window = if window_position.abs() <= 1.0 { + 0.5 + 0.5 * (std::f64::consts::PI * window_position).cos() + } else { + 0.0 + }; + let weight = cutoff * sinc * window; + value += input[sample_index as usize] as f64 * weight; + weight_sum += weight; + } + if weight_sum.abs() < 1e-12 { + input[center.clamp(0, input.len() as isize - 1) as usize] + } else { + (value / weight_sum) as f32 + } + }) + .collect() +} + +fn mel_spectrogram(signal: &[f32]) -> Result> { + let frame_count = 1 + signal + .len() + .saturating_sub(FRAME_SIZE / 2) + .div_ceil(HOP_SIZE); + let filters = mel_filters(); + let mut planner = FftPlanner::::new(); + let fft = planner.plan_fft_forward(FRAME_SIZE); + let mut mel = Vec::with_capacity(frame_count); + let mut spectrum = vec![Complex::new(0.0f32, 0.0); FRAME_SIZE]; + for frame_index in 0..frame_count { + let start = frame_index as isize * HOP_SIZE as isize - (FRAME_SIZE / 2) as isize; + for (index, value) in spectrum.iter_mut().enumerate() { + let source = start + index as isize; + let sample = if source >= 0 { + signal.get(source as usize).copied().unwrap_or(0.0) + } else { + 0.0 + }; + let window = 0.5 + - 0.5 * (2.0 * std::f32::consts::PI * index as f32 / (FRAME_SIZE - 1) as f32).cos(); + *value = Complex::new(sample * window, 0.0); + } + fft.process(&mut spectrum); + let powers: Vec = spectrum[..=FRAME_SIZE / 2] + .iter() + .map(|value| value.norm_sqr()) + .collect(); + let mut bands = [0.0f32; MEL_BANDS]; + for (band, weights) in filters.iter().enumerate() { + let energy: f32 = powers + .iter() + .zip(weights) + .map(|(power, weight)| power * weight) + .sum(); + bands[band] = (1.0 + 10_000.0 * energy.max(0.0)).log10(); + } + mel.push(bands); + } + Ok(mel) +} + +fn mel_filters() -> Vec> { + let low = hz_to_mel_slaney(0.0); + let high = hz_to_mel_slaney((SAMPLE_RATE / 2) as f32); + let points: Vec = (0..MEL_BANDS + 2) + .map(|index| mel_to_hz_slaney(low + (high - low) * index as f32 / (MEL_BANDS + 1) as f32)) + .collect(); + let frequency_scale = (SAMPLE_RATE as f32 / 2.0) / (FRAME_SIZE / 2) as f32; + (0..MEL_BANDS) + .map(|band| { + let left = points[band]; + let center = points[band + 1]; + let right = points[band + 2]; + let area = ((center - left) + (right - center)) / 2.0; + (0..=FRAME_SIZE / 2) + .map(|bin| { + let frequency = bin as f32 * frequency_scale; + let triangle = if frequency < left || frequency > right { + 0.0 + } else if frequency < center { + (frequency - left) / (center - left) + } else { + (right - frequency) / (right - center) + }; + triangle.max(0.0) / area + }) + .collect() + }) + .collect() +} + +fn hz_to_mel_slaney(hz: f32) -> f32 { + if hz < 1000.0 { + hz / (200.0 / 3.0) + } else { + 15.0 + 27.0 * (hz / 1000.0).ln() / 6.4f32.ln() + } +} + +fn mel_to_hz_slaney(mel: f32) -> f32 { + if mel < 15.0 { + mel * (200.0 / 3.0) + } else { + 1000.0 * (6.4f32.ln() * (mel - 15.0) / 27.0).exp() + } +} + +fn normalize(vector: &mut [f32]) -> Result<()> { + let norm = vector.iter().map(|value| value * value).sum::().sqrt(); + anyhow::ensure!( + norm.is_finite() && norm > f32::EPSILON, + "zero or invalid embedding" + ); + for value in vector { + *value /= norm; + } + Ok(()) +} + +fn dot(left: &[f32], right: &[f32]) -> f32 { + left.iter().zip(right).map(|(a, b)| a * b).sum() +} + +fn is_near_duplicate(candidate: &[f32], kept: &[&[f32]]) -> bool { + kept.iter() + .any(|existing| dot(candidate, existing) >= NEAR_DUPLICATE_COSINE) +} + +fn sha256_file(path: &Path) -> Result { + use std::io::Read as _; + + let mut file = File::open(path)?; + let mut hasher = Sha256::new(); + let mut buffer = [0u8; 64 * 1024]; + loop { + let read = file.read(&mut buffer)?; + if read == 0 { + break; + } + hasher.update(&buffer[..read]); + } + Ok(format!("{:x}", hasher.finalize())) +} + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn read(lock: &RwLock) -> std::sync::RwLockReadGuard<'_, T> { + lock.read() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn write(lock: &RwLock) -> std::sync::RwLockWriteGuard<'_, T> { + lock.write() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn profile_fingerprint_matches_the_tui_contract() { + assert_eq!( + profile_fingerprint(&MODELS[0], DEFAULT_PROFILE_ID), + "sim1:9293527b186f2f7e8b3dc2d6b05ce57721299840e80e0e7aaaa922f85c37b0e3" + ); + assert_ne!( + profile_fingerprint(&MODELS[0], DEFAULT_PROFILE_ID), + profile_fingerprint(&MODELS[0], "another-profile") + ); + } + + #[test] + fn embedding_bytes_round_trip() { + let vector = vec![0.1, -0.2, 0.3]; + let bytes = embedding_to_bytes(&vector); + assert_eq!(embedding_from_bytes(3, &bytes).unwrap(), vector); + assert!(embedding_from_bytes(4, &bytes).is_err()); + } + + #[test] + fn near_duplicates_are_filtered() { + let query = [1.0, 0.0, 0.0]; + let near_duplicate = [0.99995, 0.01, 0.0]; + let distinct = [0.0, 1.0, 0.0]; + assert!(is_near_duplicate(&near_duplicate, &[&query])); + assert!(!is_near_duplicate(&distinct, &[&query])); + } + + #[test] + fn resampling_keeps_a_constant_signal() { + let output = resample_sinc(&vec![0.25; 441], 44_100, 16_000); + assert_eq!(output.len(), 160); + assert!(output.iter().all(|value| (*value - 0.25).abs() < 1e-6)); + } +} diff --git a/templates/admin/v2.html b/templates/admin/v2.html index 9aecbd9..c606911 100644 --- a/templates/admin/v2.html +++ b/templates/admin/v2.html @@ -2192,6 +2192,63 @@ tbody tr:hover { + +
+
+
+ Similarity Search + Local audio embeddings and federated nearest-track search +
+
+
+
+ +
+ + +
+
Downloads the selected model and processes every visible local track. When federation is also enabled, this instance sends anonymized query embeddings to peers and answers their searches.
+
+
+ + +
+
+
+ + +
+ Show profile details + 

+                                    

+                                

+                                

+                                    
+                                    
+                                    
Applied immediately after saving.

+                                

+                            

+                        

                     
 
                     

@@ -2323,7 +2380,7 @@ tbody tr:hover {
                                     

                                     
RTT

                                     
Path samples

-                                    
Protocols

+                                    
Protocols

                                     
Last peer

                                 
                                 

@@ -2366,6 +2423,44 @@ tbody tr:hover {
                             

                         
 
+                        

+                            

+                                

+                                    Similarity Status
+                                    Model download, indexing, and active profile
+                                

+                                
+                            

+                            

+                                

+                                    
Model

+                                    
Active profile

+                                    

+                                        Processing profile
+                                    

+                                    
Visible tracks

+                                    
Processed

+                                    
Failed

+                                    
Stored

+                                    
Current track

+                                

+                                

+                                    

+                                

+                                

+                                

+                                    
+                                    
+                                

+                            

+                        

+
                         

                             

                                 

@@ -2936,7 +3031,11 @@ function adminV2() {
             agent_concurrency: '',
             federation_enabled: false,
             federation_network_id: '',
-            federation_save_on_listen: false
+            federation_save_on_listen: false,
+            similarity_enabled: false,
+            similarity_model: 'discogs-effnet-bsdynamic-1',
+            similarity_profile: 'furumi-full-track-v1',
+            similarity_workers: '1'
         },
         settingsProbe: { status: 'idle', ok: false },
         settingsProbeLoading: false,
@@ -2944,6 +3043,8 @@ function adminV2() {
         federationLoading: false,
         federationTicket: '',
         fedConnectTicket: '',
+        similarityStatus: { status: { phase: 'disabled' }, models: [], profiles: [] },
+        similarityLoading: false,
         settingsSaving: false,
         routeReady: false,
         poller: null,
@@ -3007,6 +3108,11 @@ function adminV2() {
                 ]);
             } else if (this.activeView === 'users') {
                 await Promise.allSettled([this.loadUsers(false)]);
+            } else if (this.activeView === 'settings') {
+                await Promise.allSettled([
+                    this.loadFederation(false),
+                    this.loadSimilarity(false)
+                ]);
             } else {
                 await Promise.allSettled([this.loadJobs(false), this.loadReviews(false)]);
             }
@@ -3069,6 +3175,10 @@ function adminV2() {
             } else if (this.activeView === 'settings') {
                 if (updateRoute) this.setRoute('#settings');
                 await this.loadSettings();
+                await Promise.allSettled([
+                    this.loadFederation(false),
+                    this.loadSimilarity(false)
+                ]);
                 if (!this.settingsProbe.status || this.settingsProbe.status === 'idle') {
                     await this.loadSettingsProbe(false);
                 }
@@ -3216,6 +3326,21 @@ function adminV2() {
 
         async saveSettings() {
             if (this.settingsSaving) return;
+            const networkSimilarityWasEnabled = Boolean(
+                this.settings.values?.similarity_enabled
+                && this.settings.values?.federation_enabled
+            );
+            const networkSimilarityWillBeEnabled = Boolean(
+                this.settingsDraft.similarity_enabled
+                && this.settingsDraft.federation_enabled
+            );
+            if (!networkSimilarityWasEnabled && networkSimilarityWillBeEnabled) {
+                const accepted = window.confirm(
+                    'Similarity searches send an anonymized embedding of the selected track to federation peers. '
+                    + 'Peers do not receive a user identity, but the query can technically reveal what is being searched. Enable this for the instance?'
+                );
+                if (!accepted) return;
+            }
             this.settingsSaving = true;
             try {
                 await this.request(`${this.apiBase}/settings`, {
@@ -3224,6 +3349,7 @@ function adminV2() {
                 });
                 await this.loadSettings(false);
                 await this.loadFederation(false);
+                await this.loadSimilarity(false);
                 this.showToast('Settings saved');
             } catch (error) {
                 this.showToast(error.message);
@@ -3238,6 +3364,7 @@ function adminV2() {
             this.setRoute('#settings');
             await this.loadSettings();
             await this.loadFederation(false);
+            await this.loadSimilarity(false);
             if (!this.settingsProbe.status || this.settingsProbe.status === 'idle') {
                 await this.loadSettingsProbe(false);
             }
@@ -3255,6 +3382,57 @@ function adminV2() {
             }
         },
 
+        async loadSimilarity(showErrors = true) {
+            this.similarityLoading = true;
+            try {
+                this.similarityStatus = await this.request(`${this.apiBase}/similarity`);
+            } catch (error) {
+                if (showErrors) this.showToast(error.message);
+            } finally {
+                this.similarityLoading = false;
+                this.icons();
+            }
+        },
+
+        async clearSimilarityEmbeddings() {
+            if (!window.confirm('Delete embeddings for every model and preprocessing profile? Audio files are not affected.')) return;
+            this.similarityLoading = true;
+            try {
+                await this.request(`${this.apiBase}/similarity/clear`, {
+                    method: 'POST',
+                    body: '{}'
+                });
+                await this.loadSimilarity(false);
+                this.showToast('All embeddings cleared');
+            } catch (error) {
+                this.showToast(error.message);
+            } finally {
+                this.similarityLoading = false;
+                this.icons();
+            }
+        },
+
+        selectedSimilarityProfile() {
+            return (this.similarityStatus.profiles || []).find(
+                profile => profile.id === this.settingsDraft.similarity_profile
+            ) || null;
+        },
+
+        similarityBadge() {
+            const phase = this.similarityStatus.status?.phase || 'disabled';
+            if (phase === 'ready') return 'ok';
+            if (phase === 'error') return 'failed';
+            if (['downloading', 'loading', 'processing'].includes(phase)) return 'running';
+            return 'disabled';
+        },
+
+        similarityProgress() {
+            const status = this.similarityStatus.status || {};
+            const total = Number(status.total_tracks || 0);
+            const completed = Number(status.completed_tracks || 0);
+            return total > 0 ? Math.min(100, Math.round(completed * 100 / total)) : 0;
+        },
+
         async fedSyncNow() {
             this.federationLoading = true;
             try {
diff --git a/templates/player/modals.html b/templates/player/modals.html
index f691662..cc54e6a 100644
--- a/templates/player/modals.html
+++ b/templates/player/modals.html
@@ -653,6 +653,17 @@
                     
                 

 
+