diff --git a/.cargo/config.toml b/.cargo/config.toml index 6e0446f..c8c679d 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -4,3 +4,6 @@ low-speed-limit = 1 [net] retry = 10 + +[resolver] +incompatible-rust-versions = "fallback" diff --git a/.env.example b/.env.example index 1f7026e..4aeaf41 100644 --- a/.env.example +++ b/.env.example @@ -30,8 +30,7 @@ JWT_EXPIRY_HOURS=168 # API Key（仅 Pro/Business 可创建） API_KEY_PEPPER=please-change-this-in-production -# 存储（当前实现仅支持 local） -STORAGE_TYPE=local +# 本地回退与批处理临时目录；S3 端点在管理后台配置 STORAGE_PATH=./uploads # 计费（已确认：Stripe） diff --git a/Cargo.lock b/Cargo.lock index 86b862a..84ebfe1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -294,6 +294,350 @@ dependencies = [ "arrayvec", ] +[[package]] +name = "aws-credential-types" +version = "1.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e26bbf46abc608f2dc61fd6cb3b7b0665497cc259a21520151ed98f8b37d2c79" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-types", + "zeroize", +] + +[[package]] +name = "aws-lc-rs" +version = "1.17.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.43.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + +[[package]] +name = "aws-runtime" +version = "1.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0f92058d22a46adf53ec57a6a96f34447daf02bff52e8fb956c66bcd5c6ac12" +dependencies = [ + "aws-credential-types", + "aws-sigv4", + "aws-smithy-async", + "aws-smithy-eventstream", + "aws-smithy-http 0.63.4", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "bytes-utils", + "fastrand", + "http 0.2.12", + "http 1.4.0", + "http-body 0.4.6", + "http-body 1.0.1", + "percent-encoding", + "pin-project-lite", + "tracing", + "uuid", +] + +[[package]] +name = "aws-sdk-s3" +version = "1.119.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d65fddc3844f902dfe1864acb8494db5f9342015ee3ab7890270d36fbd2e01c" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-sigv4", + "aws-smithy-async", + "aws-smithy-checksums", + "aws-smithy-eventstream", + "aws-smithy-http 0.62.6", + "aws-smithy-json", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-smithy-xml", + "aws-types", + "bytes", + "fastrand", + "hex", + "hmac", + "http 0.2.12", + "http 1.4.0", + "http-body 0.4.6", + "lru", + "percent-encoding", + "regex-lite", + "sha2", + "tracing", + "url", +] + +[[package]] +name = "aws-sigv4" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68f6ae9b71597dc5fd115d52849d7a5556ad9265885ad3492ea8d73b93bbc46e" +dependencies = [ + "aws-credential-types", + "aws-smithy-eventstream", + "aws-smithy-http 0.63.4", + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "form_urlencoded", + "hex", + "hmac", + "http 0.2.12", + "http 1.4.0", + "percent-encoding", + "sha2", + "time", + "tracing", +] + +[[package]] +name = "aws-smithy-async" +version = "1.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cba48474f1d6807384d06fec085b909f5807e16653c5af5c45dfe89539f0b70" +dependencies = [ + "futures-util", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "aws-smithy-checksums" +version = "0.63.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87294a084b43d649d967efe58aa1f9e0adc260e13a6938eb904c0ae9b45824ae" +dependencies = [ + "aws-smithy-http 0.62.6", + "aws-smithy-types", + "bytes", + "crc-fast", + "hex", + "http 0.2.12", + "http-body 0.4.6", + "md-5", + "pin-project-lite", + "sha1", + "sha2", + "tracing", +] + +[[package]] +name = "aws-smithy-eventstream" +version = "0.60.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c0b3e587fbaa5d7f7e870544508af8ce82ea47cd30376e69e1e37c4ac746f79" +dependencies = [ + "aws-smithy-types", + "bytes", + "crc32fast", +] + +[[package]] +name = "aws-smithy-http" +version = "0.62.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "826141069295752372f8203c17f28e30c464d22899a43a0c9fd9c458d469c88b" +dependencies = [ + "aws-smithy-eventstream", + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "bytes-utils", + "futures-core", + "futures-util", + "http 0.2.12", + "http 1.4.0", + "http-body 0.4.6", + "percent-encoding", + "pin-project-lite", + "pin-utils", + "tracing", +] + +[[package]] +name = "aws-smithy-http" +version = "0.63.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af4a8a5fe3e4ac7ee871237c340bbce13e982d37543b65700f4419e039f5d78e" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "bytes-utils", + "futures-core", + "futures-util", + "http 1.4.0", + "http-body 1.0.1", + "http-body-util", + "percent-encoding", + "pin-project-lite", + "pin-utils", + "tracing", +] + +[[package]] +name = "aws-smithy-http-client" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0709f0083aa19b704132684bc26d3c868e06bd428ccc4373b0b55c3e8748a58b" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-types", + "h2 0.3.27", + "h2 0.4.15", + "http 0.2.12", + "http 1.4.0", + "http-body 0.4.6", + "hyper 0.14.32", + "hyper 1.8.1", + "hyper-rustls 0.24.2", + "hyper-rustls 0.27.7", + "hyper-util", + "pin-project-lite", + "rustls 0.21.12", + "rustls 0.23.35", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-rustls 0.26.4", + "tower", + "tracing", +] + +[[package]] +name = "aws-smithy-json" +version = "0.61.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49fa1213db31ac95288d981476f78d05d9cbb0353d22cdf3472cc05bb02f6551" +dependencies = [ + "aws-smithy-types", +] + +[[package]] +name = "aws-smithy-observability" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3f39d5bb871aaf461d59144557f16d5927a5248a983a40654d9cf3b9ba183b" +dependencies = [ + "aws-smithy-runtime-api", +] + +[[package]] +name = "aws-smithy-runtime" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd3dfc18c1ce097cf81fced7192731e63809829c6cbf933c1ec47452d08e1aa" +dependencies = [ + "aws-smithy-async", + "aws-smithy-http 0.63.4", + "aws-smithy-http-client", + "aws-smithy-observability", + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "fastrand", + "http 0.2.12", + "http 1.4.0", + "http-body 0.4.6", + "http-body 1.0.1", + "http-body-util", + "pin-project-lite", + "pin-utils", + "tokio", + "tracing", +] + +[[package]] +name = "aws-smithy-runtime-api" +version = "1.11.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c55e0837e9b8526f49e0b9bfa9ee18ddee70e853f5bc09c5d11ebceddcb0fec" +dependencies = [ + "aws-smithy-async", + "aws-smithy-types", + "bytes", + "http 0.2.12", + "http 1.4.0", + "pin-project-lite", + "tokio", + "tracing", + "zeroize", +] + +[[package]] +name = "aws-smithy-types" +version = "1.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "576b0d6991c9c32bc14fc340582ef148311f924d41815f641a308b5d11e8e7cd" +dependencies = [ + "base64-simd", + "bytes", + "bytes-utils", + "futures-core", + "http 0.2.12", + "http 1.4.0", + "http-body 0.4.6", + "http-body 1.0.1", + "http-body-util", + "itoa", + "num-integer", + "pin-project-lite", + "pin-utils", + "ryu", + "serde", + "time", + "tokio", + "tokio-util", +] + +[[package]] +name = "aws-smithy-xml" +version = "0.60.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ce02add1aa3677d022f8adf81dcbe3046a95f17a1b1e8979c145cd21d3d22b3" +dependencies = [ + "xmlparser", +] + +[[package]] +name = "aws-types" +version = "1.3.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c50f3cdf47caa8d01f2be4a6663ea02418e892f9bbfd82c7b9a3a37eaccdd3a" +dependencies = [ + "aws-credential-types", + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-types", + "rustc_version", + "tracing", +] + [[package]] name = "axum" version = "0.8.7" @@ -304,10 +648,10 @@ dependencies = [ "bytes", "form_urlencoded", "futures-util", - "http", - "http-body", + "http 1.4.0", + "http-body 1.0.1", "http-body-util", - "hyper", + "hyper 1.8.1", "hyper-util", "itoa", "matchit", @@ -336,8 +680,8 @@ checksum = "59446ce19cd142f8833f856eb31f3eb097812d1479ab224f54d72428ca21ea22" dependencies = [ "bytes", "futures-core", - "http", - "http-body", + "http 1.4.0", + "http-body 1.0.1", "http-body-util", "mime", "pin-project-lite", @@ -359,8 +703,8 @@ dependencies = [ "cookie", "futures-core", "futures-util", - "http", - "http-body", + "http 1.4.0", + "http-body 1.0.1", "http-body-util", "mime", "pin-project-lite", @@ -381,6 +725,16 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64-simd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "339abbe78e73178762e23bea9dfd08e697eb3f3301cd4be981c0f78ba5859195" +dependencies = [ + "outref", + "vsimd", +] + [[package]] name = "base64ct" version = "1.8.1" @@ -504,6 +858,16 @@ version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3" +[[package]] +name = "bytes-utils" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35" +dependencies = [ + "bytes", + "either", +] + [[package]] name = "bzip2" version = "0.5.2" @@ -629,6 +993,15 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a1d728cc89cf3aee9ff92b05e62b19ee65a02b5702cff7d5a377e32c6ae29d8d" +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + [[package]] name = "color_quant" version = "1.1.0" @@ -698,6 +1071,16 @@ dependencies = [ "version_check", ] +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -728,6 +1111,19 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" +[[package]] +name = "crc-fast" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ddc2d09feefeee8bd78101665bd8645637828fa9317f9f292496dbbd8c65ff3" +dependencies = [ + "crc", + "digest", + "rand 0.9.2", + "regex", + "rustversion", +] + [[package]] name = "crc32fast" version = "1.5.0" @@ -894,6 +1290,12 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "either" version = "1.15.0" @@ -1085,6 +1487,18 @@ dependencies = [ "spin", ] +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -1094,6 +1508,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "funty" version = "2.0.0" @@ -1263,6 +1683,44 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" +[[package]] +name = "h2" +version = "0.3.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0beca50380b1fc32983fc1cb4587bfa4bb9e78fc259aad4a0032d2080309222d" +dependencies = [ + "bytes", + "fnv", + "futures-core", + "futures-sink", + "futures-util", + "http 0.2.12", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http 1.4.0", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "half" version = "2.7.1" @@ -1293,6 +1751,17 @@ dependencies = [ "allocator-api2", ] +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + [[package]] name = "hashbrown" version = "0.16.1" @@ -1356,6 +1825,17 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "http" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + [[package]] name = "http" version = "1.4.0" @@ -1366,6 +1846,17 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" +dependencies = [ + "bytes", + "http 0.2.12", + "pin-project-lite", +] + [[package]] name = "http-body" version = "1.0.1" @@ -1373,7 +1864,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" dependencies = [ "bytes", - "http", + "http 1.4.0", ] [[package]] @@ -1384,8 +1875,8 @@ checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" dependencies = [ "bytes", "futures-core", - "http", - "http-body", + "http 1.4.0", + "http-body 1.0.1", "pin-project-lite", ] @@ -1407,6 +1898,30 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "hyper" +version = "0.14.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41dfc780fdec9373c01bae43289ea34c972e40ee3c9f6b3c8801a35f35586ce7" +dependencies = [ + "bytes", + "futures-channel", + "futures-core", + "futures-util", + "h2 0.3.27", + "http 0.2.12", + "http-body 0.4.6", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "socket2 0.4.10", + "tokio", + "tower-service", + "tracing", + "want", +] + [[package]] name = "hyper" version = "1.8.1" @@ -1417,8 +1932,9 @@ dependencies = [ "bytes", "futures-channel", "futures-core", - "http", - "http-body", + "h2 0.4.15", + "http 1.4.0", + "http-body 1.0.1", "httparse", "httpdate", "itoa", @@ -1429,19 +1945,35 @@ dependencies = [ "want", ] +[[package]] +name = "hyper-rustls" +version = "0.24.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec3efd23720e2049821a693cbc7e65ea87c72f1c58ff2f9522ff332b1491e590" +dependencies = [ + "futures-util", + "http 0.2.12", + "hyper 0.14.32", + "log", + "rustls 0.21.12", + "tokio", + "tokio-rustls 0.24.1", +] + [[package]] name = "hyper-rustls" version = "0.27.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58" dependencies = [ - "http", - "hyper", + "http 1.4.0", + "hyper 1.8.1", "hyper-util", "rustls 0.23.35", + "rustls-native-certs", "rustls-pki-types", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.4", "tower-service", "webpki-roots 1.0.4", ] @@ -1457,9 +1989,9 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http", - "http-body", - "hyper", + "http 1.4.0", + "http-body 1.0.1", + "hyper 1.8.1", "ipnet", "libc", "percent-encoding", @@ -1633,6 +2165,7 @@ version = "0.1.0" dependencies = [ "aes-gcm", "argon2", + "aws-sdk-s3", "axum", "axum-extra", "base64 0.22.1", @@ -1663,6 +2196,7 @@ dependencies = [ "tower-http", "tracing", "tracing-subscriber", + "url", "uuid", "webp", "zip", @@ -1826,7 +2360,7 @@ dependencies = [ "rustls 0.23.35", "socket2 0.6.1", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.4", "url", "webpki-roots 1.0.4", ] @@ -1939,6 +2473,15 @@ dependencies = [ "imgref", ] +[[package]] +name = "lru" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +dependencies = [ + "hashbrown 0.15.5", +] + [[package]] name = "lru-slab" version = "0.1.2" @@ -2083,7 +2626,7 @@ dependencies = [ "bytes", "encoding_rs", "futures-util", - "http", + "http 1.4.0", "httparse", "memchr", "mime", @@ -2251,6 +2794,18 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" +[[package]] +name = "openssl-probe" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" + +[[package]] +name = "outref" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" + [[package]] name = "oxipng" version = "9.1.5" @@ -2763,6 +3318,18 @@ dependencies = [ "bitflags", ] +[[package]] +name = "regex" +version = "1.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + [[package]] name = "regex-automata" version = "0.4.13" @@ -2774,6 +3341,12 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "regex-lite" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" + [[package]] name = "regex-syntax" version = "0.8.8" @@ -2789,11 +3362,11 @@ dependencies = [ "base64 0.22.1", "bytes", "futures-core", - "http", - "http-body", + "http 1.4.0", + "http-body 1.0.1", "http-body-util", - "hyper", - "hyper-rustls", + "hyper 1.8.1", + "hyper-rustls 0.27.7", "hyper-util", "js-sys", "log", @@ -2807,7 +3380,7 @@ dependencies = [ "serde_urlencoded", "sync_wrapper", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.4", "tower", "tower-http", "tower-service", @@ -2867,6 +3440,15 @@ version = "2.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "357703d41365b4b27c590e3ed91eabb1b663f07c4c084095e60cbed4362dff0d" +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "rustix" version = "1.1.2" @@ -2886,6 +3468,7 @@ version = "0.21.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e" dependencies = [ + "log", "ring", "rustls-webpki 0.101.7", "sct", @@ -2897,6 +3480,7 @@ version = "0.23.35" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "533f54bc6a7d4f647e46ad909549eda97bf5afc1585190ef692b4286b198bd8f" dependencies = [ + "aws-lc-rs", "log", "once_cell", "ring", @@ -2906,6 +3490,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + [[package]] name = "rustls-pemfile" version = "1.0.4" @@ -2941,6 +3537,7 @@ version = "0.103.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2ffdfa2f5286e2247234e03f680868ac2815974dc39e00ea15adc445d0aafe52" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -2958,6 +3555,15 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f" +[[package]] +name = "schannel" +version = "0.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "scopeguard" version = "1.2.0" @@ -2974,6 +3580,35 @@ dependencies = [ "untrusted", ] +[[package]] +name = "security-framework" +version = "3.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d17b898a6d6948c3a8ee4372c17cb384f90d2e6e912ef00895b14fd7ab54ec38" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + [[package]] name = "serde" version = "1.0.228" @@ -3728,6 +4363,16 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-rustls" +version = "0.24.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081" +dependencies = [ + "rustls 0.21.12", + "tokio", +] + [[package]] name = "tokio-rustls" version = "0.26.4" @@ -3862,8 +4507,8 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http", - "http-body", + "http 1.4.0", + "http-body 1.0.1", "http-body-util", "http-range-header", "httpdate", @@ -4104,6 +4749,12 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "vsimd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" + [[package]] name = "want" version = "0.3.1" @@ -4594,6 +5245,12 @@ dependencies = [ "tap", ] +[[package]] +name = "xmlparser" +version = "0.13.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" + [[package]] name = "xz2" version = "0.1.7" diff --git a/Cargo.toml b/Cargo.toml index e1f0d06..389394b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ name = "imageforge" version = "0.1.0" edition = "2021" +rust-version = "1.92" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -38,6 +39,9 @@ rand = "0.8" sha2 = "0.10" aes-gcm = "0.10" +# S3-compatible object storage (Garage, MinIO, AWS S3, etc.) +aws-sdk-s3 = { version = "1", default-features = false, features = ["rustls", "default-https-client", "rt-tokio"] } + # Images # Keep image-rs limited to formats exposed by the API. AVIF decoding uses # libdav1d on Linux; encoding is handled by the direct ravif dependency. @@ -51,6 +55,7 @@ img-parts = "0.4" # HTTP client (Stripe API) reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } +url = "2" # Mail lettre = { version = "0.11", default-features = false, features = ["tokio1", "tokio1-rustls-tls", "builder", "smtp-transport"] } diff --git a/README.md b/README.md index dbac907..7202920 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,8 @@ - **压缩率**：1-100（JPEG/WebP/AVIF 以该比例为体积上限；无损格式按安全方式尽力优化） - **用户系统**：注册、登录、API Key 管理 - **计费与用量**：套餐/订阅/配额/发票 -- **管理员后台**：用户管理、系统监控、配置管理 +- **管理员后台**：用户管理、系统监控、配置管理、S3 多端点管理与连通性测试 +- **对象存储**：Garage/MinIO/AWS S3 兼容，私有 Bucket + 短期签名直连下载 ### 技术栈 @@ -23,7 +24,7 @@ | 缓存/队列 | Redis | 会话管理、限流、任务队列（Streams） | | 前端 | Vue3 + TypeScript | SPA 单页应用 | | 认证 | JWT + API Key | 双重认证机制 | -| 存储 | 本地文件系统 | S3/MinIO 尚在规划中 | +| 存储 | 本地文件系统 + AWS S3 SDK | 支持动态 S3 端点、分片上传、签名下载和本地回退 | | 计费 | Stripe | Checkout/Portal/Webhook | ## 目录结构 @@ -81,7 +82,7 @@ imageforge/ ### 环境要求 -- Rust（建议使用最新 stable；当前依赖链要求较新的 Rust，建议 `>= 1.85`） +- Rust（建议使用最新 stable；当前依赖链要求 `>= 1.91`，生产镜像使用 1.92） - PostgreSQL 16+ - Redis 7+ - Node.js 22+（前端构建） @@ -129,6 +130,7 @@ curl --fail http://127.0.0.1:8080/health - [邮件服务](./docs/email.md) - [可观测性](./docs/observability.md) - [部署指南](./docs/deployment.md) +- [双服务器 S3 方案](./docs/s3-storage-plan.md) - [服务条款（模板）](./docs/terms.md) - [隐私政策（模板）](./docs/privacy.md) diff --git a/docker/docker-compose.prod.yml b/docker/docker-compose.prod.yml index f55be87..9b0b3a1 100644 --- a/docker/docker-compose.prod.yml +++ b/docker/docker-compose.prod.yml @@ -8,7 +8,6 @@ x-imageforge-environment: &imageforge-environment JWT_EXPIRY_HOURS: ${JWT_EXPIRY_HOURS:-168} API_KEY_PEPPER: ${API_KEY_PEPPER:?API_KEY_PEPPER is required} BILLING_PROVIDER: ${BILLING_PROVIDER:-stripe} - STORAGE_TYPE: local STORAGE_PATH: /app/uploads PUBLIC_BASE_URL: ${PUBLIC_BASE_URL:-http://localhost:8080} WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-2} diff --git a/docker/storage/docker-compose.storage.yml.example b/docker/storage/docker-compose.storage.yml.example new file mode 100644 index 0000000..1ced95a --- /dev/null +++ b/docker/storage/docker-compose.storage.yml.example @@ -0,0 +1,56 @@ +name: imageforge-storage + +services: + garage: + image: dxflrs/garage:v2.3.0 + command: ["/garage", "server", "--single-node"] + restart: unless-stopped + read_only: true + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + pids_limit: 256 + mem_limit: 1200m + cpus: 1.75 + ports: + - "127.0.0.1:3900:3900" + - "10.70.0.2:3900:3900" + - "10.70.0.2:3901:3901" + - "127.0.0.1:3903:3903" + volumes: + - ./garage.toml:/etc/garage.toml:ro + - /srv/garage/meta:/var/lib/garage/meta + - /srv/garage/data:/var/lib/garage/data + - /srv/garage/snapshots:/var/lib/garage/snapshots + secrets: + - source: garage_rpc_secret + target: garage_rpc_secret + mode: 0400 + - source: garage_admin_token + target: garage_admin_token + mode: 0400 + - source: garage_metrics_token + target: garage_metrics_token + mode: 0400 + tmpfs: + - /tmp:size=64m,mode=1777 + healthcheck: + test: ["CMD", "/garage", "status"] + interval: 15s + timeout: 5s + retries: 10 + start_period: 20s + logging: + driver: json-file + options: + max-size: "20m" + max-file: "3" + +secrets: + garage_rpc_secret: + file: ./secrets/garage_rpc_secret + garage_admin_token: + file: ./secrets/garage_admin_token + garage_metrics_token: + file: ./secrets/garage_metrics_token diff --git a/docker/storage/garage.toml.example b/docker/storage/garage.toml.example new file mode 100644 index 0000000..5f9161e --- /dev/null +++ b/docker/storage/garage.toml.example @@ -0,0 +1,37 @@ +replication_factor = 1 +consistency_mode = "consistent" + +metadata_dir = "/var/lib/garage/meta" +data_dir = "/var/lib/garage/data" +metadata_snapshots_dir = "/var/lib/garage/snapshots" +metadata_fsync = true +data_fsync = false +metadata_auto_snapshot_interval = "12h" +use_local_tz = false + +# The first node is small and traffic is currently low. SQLite avoids a large +# memory map; a future multi-node rollout can use LMDB on the new nodes. +db_engine = "sqlite" +block_size = "1M" +block_ram_buffer_max = "64MiB" +block_max_concurrent_reads = 8 +block_max_concurrent_writes_per_request = 3 + +# JPEG, PNG, WebP, AVIF and ZIP outputs are already compressed. +compression_level = 'none' + +rpc_secret_file = "/run/secrets/garage_rpc_secret" +rpc_bind_addr = "[::]:3901" +rpc_public_addr = "10.70.0.2:3901" +bootstrap_peers = [] + +[s3_api] +api_bind_addr = "[::]:3900" +s3_region = "garage" +root_domain = ".s3.garage.internal" + +[admin] +api_bind_addr = "[::]:3903" +admin_token_file = "/run/secrets/garage_admin_token" +metrics_token_file = "/run/secrets/garage_metrics_token" +metrics_require_token = true diff --git a/docker/storage/lifecycle.json b/docker/storage/lifecycle.json new file mode 100644 index 0000000..a63b7a2 --- /dev/null +++ b/docker/storage/lifecycle.json @@ -0,0 +1,46 @@ +{ + "Rules": [ + { + "ID": "results-1d-safety", + "Status": "Enabled", + "Filter": { "Prefix": "results/1d/" }, + "Expiration": { "Days": 3 } + }, + { + "ID": "archives-1d-safety", + "Status": "Enabled", + "Filter": { "Prefix": "archives/1d/" }, + "Expiration": { "Days": 3 } + }, + { + "ID": "results-7d-safety", + "Status": "Enabled", + "Filter": { "Prefix": "results/7d/" }, + "Expiration": { "Days": 9 } + }, + { + "ID": "archives-7d-safety", + "Status": "Enabled", + "Filter": { "Prefix": "archives/7d/" }, + "Expiration": { "Days": 9 } + }, + { + "ID": "results-15d-safety", + "Status": "Enabled", + "Filter": { "Prefix": "results/15d/" }, + "Expiration": { "Days": 17 } + }, + { + "ID": "archives-15d-safety", + "Status": "Enabled", + "Filter": { "Prefix": "archives/15d/" }, + "Expiration": { "Days": 17 } + }, + { + "ID": "abort-incomplete-multipart", + "Status": "Enabled", + "Filter": { "Prefix": "" }, + "AbortIncompleteMultipartUpload": { "DaysAfterInitiation": 1 } + } + ] +} diff --git a/docker/storage/nginx-files.conf.example b/docker/storage/nginx-files.conf.example new file mode 100644 index 0000000..73588f5 --- /dev/null +++ b/docker/storage/nginx-files.conf.example @@ -0,0 +1,45 @@ +server { + listen 80; + listen [::]:80; + server_name files.example.com; + + location /.well-known/acme-challenge/ { + root /var/www/certbot; + } + + location / { + return 301 https://$host$request_uri; + } +} + +server { + listen 443 ssl http2; + listen [::]:443 ssl http2; + server_name files.example.com; + + ssl_certificate /etc/letsencrypt/live/files.example.com/fullchain.pem; + ssl_certificate_key /etc/letsencrypt/live/files.example.com/privkey.pem; + + # Signed URLs contain credentials and signatures. Do not write the query + # string to the standard access log. + access_log off; + + location / { + limit_except GET HEAD { + deny all; + } + + proxy_http_version 1.1; + proxy_set_header Host $http_host; + proxy_set_header X-Forwarded-Proto https; + proxy_set_header Connection ""; + proxy_request_buffering off; + proxy_buffering off; + proxy_max_temp_file_size 0; + proxy_read_timeout 300s; + proxy_send_timeout 300s; + proxy_pass http://127.0.0.1:3900; + } + + add_header X-Content-Type-Options nosniff always; +} diff --git a/docs/api.md b/docs/api.md index aa86ef5..552480f 100644 --- a/docs/api.md +++ b/docs/api.md @@ -678,6 +678,41 @@ Authorization: Bearer Content-Type: application/json ``` +### 11.6 S3 存储端点 + +```http +GET /admin/storage/endpoints +Authorization: Bearer +``` + +```http +POST /admin/storage/endpoints +Authorization: Bearer +Content-Type: application/json + +{ + "name": "119 高带宽 S3", + "internal_endpoint": "http://10.70.0.2:3900", + "public_endpoint": "https://files.example.com", + "bucket": "imageforge-results", + "region": "garage", + "access_key": "", + "secret_key": "", + "force_path_style": true, + "presign_ttl_seconds": 300 +} +``` + +```http +PUT /admin/storage/endpoints/{endpoint_id} +POST /admin/storage/endpoints/{endpoint_id}/test +POST /admin/storage/endpoints/{endpoint_id}/activate +DELETE /admin/storage/endpoints/{endpoint_id} +Authorization: Bearer +``` + +凭据加密保存且不通过 API 回传。测试接口执行 Bucket 检查、内部临时对象读写删和公网预签名下载；激活接口会再次测试并原子切换活动端点。活动端点不能直接编辑或删除。 + --- ## 12. WebSocket（网站任务进度） diff --git a/docs/database.md b/docs/database.md index 077c152..fd6b317 100644 --- a/docs/database.md +++ b/docs/database.md @@ -341,7 +341,14 @@ CREATE TABLE tasks ( completed_at TIMESTAMPTZ, -- 到期清理：匿名可默认 24h；登录用户应由应用按套餐写入更长 retention - expires_at TIMESTAMPTZ NOT NULL DEFAULT (NOW() + INTERVAL '24 hours') + expires_at TIMESTAMPTZ NOT NULL DEFAULT (NOW() + INTERVAL '24 hours'), + retention_hours INTEGER NOT NULL DEFAULT 24, + + zip_storage_backend VARCHAR(16), + zip_storage_endpoint_id UUID REFERENCES storage_endpoints(id) ON DELETE RESTRICT, + zip_storage_key TEXT, + zip_storage_etag TEXT, + zip_size BIGINT ); CREATE INDEX idx_tasks_user_id ON tasks(user_id); @@ -365,7 +372,12 @@ CREATE TABLE task_files ( compressed_size BIGINT, saved_percent DECIMAL(6, 2), - storage_path VARCHAR(500), -- S3 key 或本地路径 + storage_path VARCHAR(500), -- 兼容旧版本本地成品路径 + input_path TEXT, -- 批任务原图临时路径，处理后清空 + storage_backend VARCHAR(16) NOT NULL DEFAULT 'local', + storage_endpoint_id UUID REFERENCES storage_endpoints(id) ON DELETE RESTRICT, + storage_key TEXT, + storage_etag TEXT, status file_status NOT NULL DEFAULT 'pending', error_message TEXT, @@ -377,6 +389,12 @@ CREATE INDEX idx_task_files_task_id ON task_files(task_id); CREATE INDEX idx_task_files_status ON task_files(status); ``` +### 4.9.1 storage_endpoints - S3 端点 + +端点凭据使用 `API_KEY_PEPPER` 派生密钥并通过 AES-256-GCM 加密。每个对象保存自己的 `storage_endpoint_id`，因此切换活动端点不会影响历史对象。活动端点和仍有关联对象的端点不能删除；管理端删除采用软归档，待无引用 30 天后由 Worker 清理配置。 + +同一时间只允许一个 `is_active = true` 的端点接收新对象。没有配置活动端点时系统使用 `STORAGE_PATH` 本地回退。 + ### 4.10 idempotency_keys - 幂等记录 ```sql CREATE TABLE idempotency_keys ( diff --git a/docs/s3-storage-plan.md b/docs/s3-storage-plan.md new file mode 100644 index 0000000..a67acad --- /dev/null +++ b/docs/s3-storage-plan.md @@ -0,0 +1,131 @@ +# ImageForge 双服务器 S3 方案 + +> 本文是部署前方案与验收清单。本次开发未在 `118.145.177.79` 或 `119.29.142.248` 上安装、修改或启动任何服务。 + +## 1. 现状与结论 + +| 节点 | 实测资源 | 建议职责 | 关键约束 | +| --- | --- | --- | --- | +| `118.145.177.79` | 8 vCPU、31 GiB RAM、约 104 GiB 可用磁盘 | ImageForge API、Worker、PostgreSQL、Redis、ZIP 临时生成 | 已有较多容器，`9000` 端口已占用，后续建议应用只绑定 `127.0.0.1:18080` | +| `119.29.142.248` | 2 vCPU、1.9 GiB RAM、约 29 GiB 可用磁盘、200M 共享带宽 | Garage S3、Nginx/TLS、下载出口 | 只能作为首期临时对象存储；系统盘容量而不是带宽会最先成为瓶颈 | + +首期选择 Garage 2.3，而不是 MinIO。Garage 面向低资源和普通互联网环境，支持本项目所需的 SigV4、Path-style、预签名 URL、对象读写删除、分片上传和生命周期。MinIO 官方单节点说明包含 2 GiB 预分配并建议 32 GiB 内存，不适合 119 当前的 2 GiB 机器。 + +- Garage 目标与适用场景：[Goals and use cases](https://garagehq.deuxfleurs.fr/documentation/design/goals/) +- Garage 单节点与 2.3.0 镜像：[Quick Start](https://garagehq.deuxfleurs.fr/documentation/quick-start/) +- Garage S3 兼容性：[S3 Compatibility](https://garagehq.deuxfleurs.fr/documentation/reference-manual/s3-compatibility/) +- Garage 配置项：[Configuration](https://garagehq.deuxfleurs.fr/documentation/reference-manual/configuration/) +- MinIO 单节点资源说明：[Deploy MinIO Single-Node](https://min.io/docs/minio/container/operations/install-deploy-manage/deploy-minio-single-node-single-drive.html) + +## 2. 目标架构 + +```mermaid +flowchart LR + U["用户浏览器"] -->|"上传、鉴权、任务查询"| A["118: ImageForge API"] + A --> P["118: PostgreSQL"] + A --> R["118: Redis"] + A --> W["118: Image Worker"] + W -->|"WireGuard / S3 PUT"| S["119: Garage S3"] + A -->|"鉴权通过后返回 307"| U + U -->|"5 分钟预签名 GET"| N["119: Nginx TLS"] + N --> S +``` + +核心规则： + +1. Bucket 始终私有，不开放匿名读。 +2. 用户先访问 ImageForge 下载接口；后端检查用户、匿名会话和过期时间后，返回 `307` 到 5 分钟预签名 URL。 +3. 图片和 ZIP 的实际下载字节只经过 119，118 不再承担下载出口。 +4. 批处理原图只在 118 的共享 `uploads/orig` 临时保存，Worker 完成或失败后删除，不上传 S3。 +5. 批量 ZIP 在 118 临时生成，64 MiB 以上自动使用 S3 分片上传，上传后删除临时目录。 +6. 每个成品记录写入时的端点 ID。以后切换端点不会使旧文件失联；旧端点凭据会保留到关联对象清空。 + +`/api/v1/compress/direct` 保留原有“响应体直接返回图片”的 API 语义，避免破坏现有调用方；网站下载、历史记录下载、普通压缩结果和批量 ZIP 均走 S3。 + +## 3. 网络与端口 + +建议先在两台服务器之间建立 WireGuard： + +| 用途 | 118 | 119 | +| --- | --- | --- | +| WireGuard 地址 | `10.70.0.1/24` | `10.70.0.2/24` | +| S3 内部 Endpoint | 客户端 | `10.70.0.2:3900` | +| Garage RPC | 后续扩容节点 | `10.70.0.2:3901` | +| 公网下载 | 不开放 | `files.<你的域名>:443` | + +119 防火墙只允许：公网 `80/443`、受限来源的 `22`、WireGuard 对端访问 `3900/3901`。Garage 管理端口 `3903` 只绑定回环地址。不要把 `3900/3901/3903` 直接暴露到公网。 + +119 已有 Nginx 和 `wy.workyai.cn`，后续只新增独立的 `files.<你的域名>` server block，不能覆盖现有站点配置。公网 Nginx 只允许 `GET/HEAD`，应用上传走 WireGuard 内部 Endpoint。 + +## 4. 保留与清理 + +| 用户层级 | 数据库精确保留 | 对象前缀 | S3 兜底生命周期 | +| --- | --- | --- | --- | +| 未登录、免费用户 | 24 小时 | `results/1d/`、`archives/1d/` | 3 天 | +| 低级会员 Pro | 7 天 | `results/7d/`、`archives/7d/` | 9 天 | +| 高级会员 Business | 15 天 | `results/15d/`、`archives/15d/` | 17 天 | + +Worker 每 5 分钟按 `expires_at` 精确删除对象，删除成功后才删除数据库任务。S3 生命周期多保留 2 天，只负责处理数据库故障、进程崩溃或上传后未能落库的孤儿对象，不能作为精确会员权限判断。未完成的分片上传 1 天后由生命周期中止。 + +## 5. 119 首期容量 + +119 当前约 29 GiB 可用，建议至少给系统、Docker/Nginx 日志和 Garage 元数据保留 9 GiB，因此首期只按约 20 GiB 对象容量规划： + +- 70%（约 14 GiB）告警：检查日新增量和清理是否正常。 +- 85%（约 17 GiB）紧急告警：停止营销放量并准备扩盘或新端点。 +- 90%（约 18 GiB）写入保护：不要继续依赖该系统盘接收新对象。 + +粗略容量公式：`日均压缩后新增量 × 加权平均保留天数 × 1.2`。例如每天 1 GiB、平均保留 5 天，仅对象约 6 GiB；每天 4 GiB 时就会接近首期上限。200 Mbps 理论上约 25 MB/s，但共享带宽、磁盘随机读和 2 核 CPU 会使实际吞吐更低。 + +正式增长前优先给 119 挂载独立的 100 GiB 以上数据盘到 `/srv/garage/data`。单节点 `replication_factor = 1` 没有冗余，官方也不建议用于长期生产数据；本项目对象最长 15 天且可重新生成，低用户量阶段可以接受，但不能替代备份。后续可新增 Garage 节点并提升副本数，或在管理后台新增另一套 S3 端点并切换新对象。 + +## 6. 管理后台配置 + +部署 S3 并创建 Bucket/Key 后，在“管理后台 -> 对象存储”新增： + +| 字段 | 首期建议值 | +| --- | --- | +| 名称 | `119 高带宽 S3` | +| 内部 Endpoint | `http://10.70.0.2:3900` | +| 公网 Endpoint | `https://files.<你的域名>` | +| Bucket | `imageforge-results` | +| Region | `garage` | +| Force path style | 开启 | +| 签名有效期 | `300` 秒 | + +Access Key 和 Secret Key 使用项目现有 AES-256-GCM 机制加密入库，页面只显示掩码。保存后先执行“全链路测试”，再点“验证并启用”。后端激活前会再次执行 `HeadBucket + 内部 PutObject/GetObject + 公网预签名 GET + 内部 DeleteObject`。活动端点或仍有关联对象的端点不能直接编辑、归档，需先新增并启用替代端点，等待旧对象过期后再移除。 + +## 7. 后续部署顺序（本次不执行） + +1. 为下载域名添加 DNS，确认 119 的 80/443 可用，并建立 WireGuard。 +2. 在 119 创建 `/srv/garage/{meta,data,snapshots}`，生成三个权限为 `0600` 的 secret 文件。 +3. 使用 `docker/storage/garage.toml.example` 和 `docker-compose.storage.yml.example` 启动 Garage。 +4. 创建 `imageforge-results` Bucket 和仅限该 Bucket 的应用 Key，应用生命周期文件 `lifecycle.json`。 +5. 新增 `nginx-files.conf.example` 对应的 HTTPS 站点；确认访问日志不记录签名查询字符串。 +6. 在 118 备份数据库，部署新版本并执行迁移；首次启动没有活动 S3 时仍使用本地存储。 +7. 管理后台保存 119 端点，执行测试并启用。 +8. 使用 JPEG、PNG（透明/非透明）、WebP、AVIF、GIF、BMP、TIFF、ICO，以及中文名、同名文件、大图和批量 ZIP 做验收。 + +Garage 初始化命令示例（实际部署时执行，输出的 Secret Key 只录入管理后台，不提交仓库）： + +```bash +docker compose -f docker-compose.storage.yml exec garage /garage bucket create imageforge-results +docker compose -f docker-compose.storage.yml exec garage /garage key create imageforge-app +docker compose -f docker-compose.storage.yml exec garage /garage bucket allow \ + --read --write --owner imageforge-results --key imageforge-app +aws --endpoint-url http://127.0.0.1:3900 \ + s3api put-bucket-lifecycle-configuration \ + --bucket imageforge-results \ + --lifecycle-configuration file://lifecycle.json +``` + +## 8. 上线验收 + +- 管理端测试必须完成读、写、删，Bucket 内不能残留健康检查对象。 +- 普通下载接口先返回 `307`，`Location` 指向下载域名且有效期约 300 秒。 +- 未授权用户无法取得签名 URL；任务过期后应用下载接口返回 404。 +- 单文件与批任务数据库均记录正确的 `storage_endpoint_id` 和对象键。 +- 切换到第二端点后，新对象进入第二端点，第一端点历史对象仍可下载。 +- S3 停止时同步请求返回 503，批任务保留临时原图并重试，不静默写回本地。 +- 删除 S3 对象失败时，过期任务数据库记录保留并在下一轮重试。 +- 119 磁盘 70%/85% 告警、容器重启、NTP 时间同步和证书续期均验证通过。 diff --git a/frontend/src/app/layouts/AdminLayout.vue b/frontend/src/app/layouts/AdminLayout.vue index 2756560..d2c4189 100644 --- a/frontend/src/app/layouts/AdminLayout.vue +++ b/frontend/src/app/layouts/AdminLayout.vue @@ -51,6 +51,13 @@ > 支付与邮件 + + 对象存储 + import('@/pages/admin/AdminTasksPage.vue') }, { path: 'billing', name: 'admin-billing', component: () => import('@/pages/admin/AdminBillingPage.vue') }, { path: 'integrations', name: 'admin-integrations', component: () => import('@/pages/admin/AdminIntegrationsPage.vue') }, + { path: 'storage', name: 'admin-storage', component: () => import('@/pages/admin/AdminStoragePage.vue') }, { path: 'config', name: 'admin-config', component: () => import('@/pages/admin/AdminConfigPage.vue') }, ], }, diff --git a/frontend/src/pages/admin/AdminStoragePage.vue b/frontend/src/pages/admin/AdminStoragePage.vue new file mode 100644 index 0000000..88f60dd --- /dev/null +++ b/frontend/src/pages/admin/AdminStoragePage.vue @@ -0,0 +1,414 @@ + + + diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index 0de5731..a737b77 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -524,3 +524,87 @@ export async function updateMailConfig( export async function sendMailTest(token: string, to?: string): Promise<{ message: string }> { return apiJson<{ message: string }>('/api/v1/admin/mail/test', { to }, token) } + +export interface AdminStorageEndpoint { + id: string + name: string + internal_endpoint: string + public_endpoint: string + bucket: string + region: string + access_key_hint: string + credentials_configured: boolean + force_path_style: boolean + presign_ttl_seconds: number + is_active: boolean + last_test_at?: string | null + last_test_ok?: boolean | null + last_test_error?: string | null + object_count: number + stored_bytes: number + created_at: string + updated_at: string +} + +export interface AdminStorageEndpointsResponse { + active_backend: 'local' | 's3' + endpoints: AdminStorageEndpoint[] +} + +export interface AdminStorageEndpointPayload { + name: string + internal_endpoint: string + public_endpoint: string + bucket: string + region?: string + access_key?: string + secret_key?: string + force_path_style?: boolean + presign_ttl_seconds?: number +} + +export interface AdminStorageActionResponse { + message: string + endpoint: AdminStorageEndpoint +} + +export async function listStorageEndpoints(token: string): Promise { + return apiGet('/api/v1/admin/storage/endpoints', token) +} + +export async function createStorageEndpoint( + token: string, + payload: AdminStorageEndpointPayload & { access_key: string; secret_key: string }, +): Promise { + return apiJson('/api/v1/admin/storage/endpoints', payload, token) +} + +export async function updateStorageEndpoint( + token: string, + endpointId: string, + payload: Partial, +): Promise { + return apiJson(`/api/v1/admin/storage/endpoints/${endpointId}`, payload, token, { + method: 'PUT', + }) +} + +export async function testStorageEndpoint( + token: string, + endpointId: string, +): Promise { + return apiJson(`/api/v1/admin/storage/endpoints/${endpointId}/test`, {}, token) +} + +export async function activateStorageEndpoint( + token: string, + endpointId: string, +): Promise { + return apiJson(`/api/v1/admin/storage/endpoints/${endpointId}/activate`, {}, token) +} + +export async function deleteStorageEndpoint(token: string, endpointId: string): Promise<{ message: string }> { + return apiJson<{ message: string }>(`/api/v1/admin/storage/endpoints/${endpointId}`, undefined, token, { + method: 'DELETE', + }) +} diff --git a/migrations/005_s3_storage.sql b/migrations/005_s3_storage.sql new file mode 100644 index 0000000..6039843 --- /dev/null +++ b/migrations/005_s3_storage.sql @@ -0,0 +1,78 @@ +BEGIN; + +CREATE TABLE storage_endpoints ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + name VARCHAR(100) NOT NULL, + internal_endpoint TEXT NOT NULL, + public_endpoint TEXT NOT NULL, + bucket VARCHAR(100) NOT NULL, + region VARCHAR(100) NOT NULL DEFAULT 'garage', + access_key_encrypted TEXT NOT NULL, + secret_key_encrypted TEXT NOT NULL, + access_key_hint VARCHAR(32) NOT NULL, + force_path_style BOOLEAN NOT NULL DEFAULT true, + presign_ttl_seconds INTEGER NOT NULL DEFAULT 300, + is_active BOOLEAN NOT NULL DEFAULT false, + last_test_at TIMESTAMPTZ, + last_test_ok BOOLEAN, + last_test_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_by UUID REFERENCES users(id) ON DELETE SET NULL, + deleted_at TIMESTAMPTZ, + CONSTRAINT storage_endpoint_presign_ttl + CHECK (presign_ttl_seconds BETWEEN 60 AND 3600) +); + +CREATE UNIQUE INDEX storage_endpoints_one_active + ON storage_endpoints (is_active) + WHERE is_active = true; +CREATE UNIQUE INDEX storage_endpoints_name_active + ON storage_endpoints (lower(name)) + WHERE deleted_at IS NULL; +CREATE INDEX storage_endpoints_updated_at + ON storage_endpoints (updated_at DESC); + +ALTER TABLE task_files + ADD COLUMN input_path TEXT, + ADD COLUMN storage_backend VARCHAR(16) NOT NULL DEFAULT 'local', + ADD COLUMN storage_endpoint_id UUID REFERENCES storage_endpoints(id) ON DELETE RESTRICT, + ADD COLUMN storage_key TEXT, + ADD COLUMN storage_etag TEXT; + +UPDATE task_files +SET input_path = storage_path +WHERE status IN ('pending', 'processing') AND storage_path IS NOT NULL; + +UPDATE task_files +SET storage_key = storage_path +WHERE status = 'completed' AND storage_path IS NOT NULL; + +CREATE INDEX task_files_storage_endpoint + ON task_files (storage_endpoint_id) + WHERE storage_endpoint_id IS NOT NULL; + +ALTER TABLE tasks + ADD COLUMN retention_hours INTEGER NOT NULL DEFAULT 24, + ADD COLUMN zip_storage_backend VARCHAR(16), + ADD COLUMN zip_storage_endpoint_id UUID REFERENCES storage_endpoints(id) ON DELETE RESTRICT, + ADD COLUMN zip_storage_key TEXT, + ADD COLUMN zip_storage_etag TEXT, + ADD COLUMN zip_size BIGINT; + +UPDATE tasks +SET retention_hours = GREATEST( + 1, + CEIL(EXTRACT(EPOCH FROM (expires_at - created_at)) / 3600.0)::INTEGER +); + +CREATE INDEX tasks_zip_storage_endpoint + ON tasks (zip_storage_endpoint_id) + WHERE zip_storage_endpoint_id IS NOT NULL; + +UPDATE plans +SET retention_days = 15, + updated_at = NOW() +WHERE code = 'business_monthly'; + +COMMIT; diff --git a/src/api/admin.rs b/src/api/admin.rs index c43dc51..57d865f 100644 --- a/src/api/admin.rs +++ b/src/api/admin.rs @@ -40,7 +40,7 @@ pub fn router() -> Router { .route("/admin/config", put(update_config)) } -async fn require_admin( +pub(super) async fn require_admin( state: &AppState, jar: axum_extra::extract::cookie::CookieJar, headers: &HeaderMap, diff --git a/src/api/admin_storage.rs b/src/api/admin_storage.rs new file mode 100644 index 0000000..ce0229e --- /dev/null +++ b/src/api/admin_storage.rs @@ -0,0 +1,761 @@ +use crate::api::admin::require_admin; +use crate::api::context; +use crate::api::envelope::Envelope; +use crate::error::{AppError, ErrorCode}; +use crate::services::{settings, storage}; +use crate::state::AppState; + +use axum::extract::{ConnectInfo, Path, State}; +use axum::http::HeaderMap; +use axum::routing::{get, post, put}; +use axum::{Json, Router}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::net::{IpAddr, SocketAddr}; +use url::Url; +use uuid::Uuid; + +pub fn router() -> Router { + Router::new() + .route( + "/admin/storage/endpoints", + get(list_storage_endpoints).post(create_storage_endpoint), + ) + .route( + "/admin/storage/endpoints/{endpoint_id}", + put(update_storage_endpoint).delete(delete_storage_endpoint), + ) + .route( + "/admin/storage/endpoints/{endpoint_id}/test", + post(test_storage_endpoint), + ) + .route( + "/admin/storage/endpoints/{endpoint_id}/activate", + post(activate_storage_endpoint), + ) +} + +#[derive(Debug, Serialize)] +struct StorageEndpointView { + id: Uuid, + name: String, + internal_endpoint: String, + public_endpoint: String, + bucket: String, + region: String, + access_key_hint: String, + credentials_configured: bool, + force_path_style: bool, + presign_ttl_seconds: i32, + is_active: bool, + last_test_at: Option>, + last_test_ok: Option, + last_test_error: Option, + object_count: i64, + stored_bytes: i64, + created_at: DateTime, + updated_at: DateTime, +} + +#[derive(Debug, Serialize)] +struct StorageEndpointsResponse { + active_backend: String, + endpoints: Vec, +} + +#[derive(Debug, Deserialize)] +struct CreateStorageEndpointRequest { + name: String, + internal_endpoint: String, + public_endpoint: String, + bucket: String, + region: Option, + access_key: String, + secret_key: String, + force_path_style: Option, + presign_ttl_seconds: Option, +} + +#[derive(Debug, Deserialize)] +struct UpdateStorageEndpointRequest { + name: Option, + internal_endpoint: Option, + public_endpoint: Option, + bucket: Option, + region: Option, + access_key: Option, + secret_key: Option, + force_path_style: Option, + presign_ttl_seconds: Option, +} + +#[derive(Debug, Serialize)] +struct StorageActionResponse { + message: String, + endpoint: StorageEndpointView, +} + +async fn list_storage_endpoints( + State(state): State, + jar: axum_extra::extract::cookie::CookieJar, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, +) -> Result>, AppError> { + let ip = context::client_ip(&headers, addr.ip()); + let (_jar, _admin_id) = require_admin(&state, jar, &headers, ip).await?; + let endpoints = storage::list_endpoints(&state).await?; + let active_backend = if endpoints.iter().any(|endpoint| endpoint.is_active) { + "s3" + } else { + "local" + }; + let mut views = Vec::with_capacity(endpoints.len()); + for endpoint in endpoints { + views.push(endpoint_view(&state, endpoint).await?); + } + + Ok(Json(Envelope { + success: true, + data: StorageEndpointsResponse { + active_backend: active_backend.to_string(), + endpoints: views, + }, + })) +} + +async fn create_storage_endpoint( + State(state): State, + jar: axum_extra::extract::cookie::CookieJar, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + Json(req): Json, +) -> Result>, AppError> { + let ip = context::client_ip(&headers, addr.ip()); + let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?; + + let name = validate_name(&req.name)?; + ensure_name_available(&state, &name, None).await?; + let internal_endpoint = validate_endpoint_url(&req.internal_endpoint, "内部 Endpoint")?; + let public_endpoint = validate_endpoint_url(&req.public_endpoint, "公网 Endpoint")?; + let bucket = validate_bucket(&req.bucket)?; + let region = validate_region(req.region.as_deref().unwrap_or("garage"))?; + let access_key = validate_credential(&req.access_key, "Access Key")?; + let secret_key = validate_credential(&req.secret_key, "Secret Key")?; + let presign_ttl_seconds = validate_presign_ttl(req.presign_ttl_seconds.unwrap_or(300))?; + + let endpoint = sqlx::query_as::<_, storage::StorageEndpoint>( + r#" + INSERT INTO storage_endpoints ( + name, internal_endpoint, public_endpoint, bucket, region, + access_key_encrypted, secret_key_encrypted, access_key_hint, + force_path_style, presign_ttl_seconds, updated_by + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + RETURNING id, name, internal_endpoint, public_endpoint, bucket, region, + access_key_encrypted, secret_key_encrypted, access_key_hint, + force_path_style, presign_ttl_seconds, is_active, + last_test_at, last_test_ok, last_test_error, + created_at, updated_at, deleted_at + "#, + ) + .bind(&name) + .bind(&internal_endpoint) + .bind(&public_endpoint) + .bind(&bucket) + .bind(®ion) + .bind(settings::encrypt_secret(&state, &access_key)?) + .bind(settings::encrypt_secret(&state, &secret_key)?) + .bind(access_key_hint(&access_key)) + .bind(req.force_path_style.unwrap_or(true)) + .bind(presign_ttl_seconds) + .bind(admin_id) + .fetch_one(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "创建存储端点失败").with_source(err))?; + + audit_storage_action( + &state, + admin_id, + ip, + "storage_endpoint.create", + endpoint.id, + serde_json::json!({ "name": endpoint.name }), + ) + .await; + + Ok(Json(Envelope { + success: true, + data: endpoint_view(&state, endpoint).await?, + })) +} + +async fn update_storage_endpoint( + State(state): State, + jar: axum_extra::extract::cookie::CookieJar, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + Path(endpoint_id): Path, + Json(req): Json, +) -> Result>, AppError> { + let ip = context::client_ip(&headers, addr.ip()); + let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?; + let existing = storage::get_endpoint(&state, endpoint_id).await?; + ensure_configurable(&existing)?; + if existing.is_active { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "活动端点不能直接编辑，请先新增并启用替代端点", + )); + } + let (object_count, _) = endpoint_usage(&state, endpoint_id).await?; + if object_count > 0 { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "端点仍有关联对象，不能修改；请新增替代端点并等待旧对象过期", + )); + } + + let name = match req.name.as_deref() { + Some(value) => validate_name(value)?, + None => existing.name.clone(), + }; + ensure_name_available(&state, &name, Some(endpoint_id)).await?; + let internal_endpoint = match req.internal_endpoint.as_deref() { + Some(value) => validate_endpoint_url(value, "内部 Endpoint")?, + None => existing.internal_endpoint.clone(), + }; + let public_endpoint = match req.public_endpoint.as_deref() { + Some(value) => validate_endpoint_url(value, "公网 Endpoint")?, + None => existing.public_endpoint.clone(), + }; + let bucket = match req.bucket.as_deref() { + Some(value) => validate_bucket(value)?, + None => existing.bucket.clone(), + }; + let region = match req.region.as_deref() { + Some(value) => validate_region(value)?, + None => existing.region.clone(), + }; + let access_key_encrypted = match req.access_key.as_deref() { + Some(value) if !value.trim().is_empty() => Some(settings::encrypt_secret( + &state, + &validate_credential(value, "Access Key")?, + )?), + _ => None, + }; + let secret_key_encrypted = match req.secret_key.as_deref() { + Some(value) if !value.trim().is_empty() => Some(settings::encrypt_secret( + &state, + &validate_credential(value, "Secret Key")?, + )?), + _ => None, + }; + let access_key_hint_value = req + .access_key + .as_deref() + .filter(|value| !value.trim().is_empty()) + .map(access_key_hint); + let presign_ttl_seconds = validate_presign_ttl( + req.presign_ttl_seconds + .unwrap_or(existing.presign_ttl_seconds), + )?; + + let endpoint = sqlx::query_as::<_, storage::StorageEndpoint>( + r#" + UPDATE storage_endpoints + SET name = $2, + internal_endpoint = $3, + public_endpoint = $4, + bucket = $5, + region = $6, + access_key_encrypted = COALESCE($7, access_key_encrypted), + secret_key_encrypted = COALESCE($8, secret_key_encrypted), + access_key_hint = COALESCE($9, access_key_hint), + force_path_style = $10, + presign_ttl_seconds = $11, + is_active = false, + last_test_at = NULL, + last_test_ok = NULL, + last_test_error = NULL, + updated_at = NOW(), + updated_by = $12 + WHERE id = $1 + RETURNING id, name, internal_endpoint, public_endpoint, bucket, region, + access_key_encrypted, secret_key_encrypted, access_key_hint, + force_path_style, presign_ttl_seconds, is_active, + last_test_at, last_test_ok, last_test_error, + created_at, updated_at, deleted_at + "#, + ) + .bind(endpoint_id) + .bind(&name) + .bind(&internal_endpoint) + .bind(&public_endpoint) + .bind(&bucket) + .bind(®ion) + .bind(access_key_encrypted) + .bind(secret_key_encrypted) + .bind(access_key_hint_value) + .bind(req.force_path_style.unwrap_or(existing.force_path_style)) + .bind(presign_ttl_seconds) + .bind(admin_id) + .fetch_one(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "更新存储端点失败").with_source(err))?; + + audit_storage_action( + &state, + admin_id, + ip, + "storage_endpoint.update", + endpoint.id, + serde_json::json!({ "name": endpoint.name, "requires_reactivation": true }), + ) + .await; + + Ok(Json(Envelope { + success: true, + data: endpoint_view(&state, endpoint).await?, + })) +} + +async fn test_storage_endpoint( + State(state): State, + jar: axum_extra::extract::cookie::CookieJar, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + Path(endpoint_id): Path, +) -> Result>, AppError> { + let ip = context::client_ip(&headers, addr.ip()); + let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?; + let endpoint = storage::get_endpoint(&state, endpoint_id).await?; + ensure_configurable(&endpoint)?; + if let Err(err) = storage::test_endpoint(&state, &endpoint).await { + let detail = storage_test_message(&err); + record_test_result(&state, endpoint_id, false, Some(&detail), admin_id).await?; + audit_storage_action( + &state, + admin_id, + ip, + "storage_endpoint.test_failed", + endpoint_id, + serde_json::json!({ "message": detail }), + ) + .await; + return Err(AppError::new(ErrorCode::StorageUnavailable, detail)); + } + + record_test_result(&state, endpoint_id, true, None, admin_id).await?; + let endpoint = storage::get_endpoint(&state, endpoint_id).await?; + ensure_configurable(&endpoint)?; + audit_storage_action( + &state, + admin_id, + ip, + "storage_endpoint.test_succeeded", + endpoint_id, + serde_json::json!({}), + ) + .await; + + Ok(Json(Envelope { + success: true, + data: StorageActionResponse { + message: "S3 内部读写删与公网签名下载测试通过".to_string(), + endpoint: endpoint_view(&state, endpoint).await?, + }, + })) +} + +async fn activate_storage_endpoint( + State(state): State, + jar: axum_extra::extract::cookie::CookieJar, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + Path(endpoint_id): Path, +) -> Result>, AppError> { + let ip = context::client_ip(&headers, addr.ip()); + let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?; + let endpoint = storage::get_endpoint(&state, endpoint_id).await?; + ensure_configurable(&endpoint)?; + let tested_config_updated_at = endpoint.updated_at; + + if let Err(err) = storage::test_endpoint(&state, &endpoint).await { + let detail = storage_test_message(&err); + record_test_result(&state, endpoint_id, false, Some(&detail), admin_id).await?; + return Err(AppError::new(ErrorCode::StorageUnavailable, detail)); + } + + let mut tx = state.db.begin().await.map_err(|err| { + AppError::new(ErrorCode::Internal, "开启存储切换事务失败").with_source(err) + })?; + sqlx::query( + "UPDATE storage_endpoints SET is_active = false, updated_at = NOW() WHERE is_active = true", + ) + .execute(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "停用旧存储端点失败").with_source(err))?; + let activated = sqlx::query( + r#" + UPDATE storage_endpoints + SET is_active = true, + last_test_at = NOW(), + last_test_ok = true, + last_test_error = NULL, + updated_at = NOW(), + updated_by = $2 + WHERE id = $1 AND updated_at = $3 AND deleted_at IS NULL + "#, + ) + .bind(endpoint_id) + .bind(admin_id) + .bind(tested_config_updated_at) + .execute(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "启用存储端点失败").with_source(err))?; + if activated.rows_affected() == 0 { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "端点配置在测试期间发生变化，请重新测试", + )); + } + tx.commit().await.map_err(|err| { + AppError::new(ErrorCode::Internal, "提交存储切换事务失败").with_source(err) + })?; + + let endpoint = storage::get_endpoint(&state, endpoint_id).await?; + audit_storage_action( + &state, + admin_id, + ip, + "storage_endpoint.activate", + endpoint_id, + serde_json::json!({ "name": endpoint.name }), + ) + .await; + + Ok(Json(Envelope { + success: true, + data: StorageActionResponse { + message: "S3 端点已启用，新生成文件将写入该端点".to_string(), + endpoint: endpoint_view(&state, endpoint).await?, + }, + })) +} + +async fn delete_storage_endpoint( + State(state): State, + jar: axum_extra::extract::cookie::CookieJar, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + Path(endpoint_id): Path, +) -> Result>, AppError> { + let ip = context::client_ip(&headers, addr.ip()); + let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?; + let endpoint = storage::get_endpoint(&state, endpoint_id).await?; + ensure_configurable(&endpoint)?; + if endpoint.is_active { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "活动端点不能删除，请先启用其他端点", + )); + } + + let (object_count, _): (i64, i64) = endpoint_usage(&state, endpoint_id).await?; + if object_count > 0 { + return Err(AppError::new( + ErrorCode::InvalidRequest, + format!("端点仍关联 {object_count} 个对象，需等待对象过期清理后再删除"), + )); + } + + sqlx::query( + "UPDATE storage_endpoints SET is_active = false, deleted_at = NOW(), updated_at = NOW(), updated_by = $2 WHERE id = $1", + ) + .bind(endpoint_id) + .bind(admin_id) + .execute(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "归档存储端点失败").with_source(err))?; + audit_storage_action( + &state, + admin_id, + ip, + "storage_endpoint.delete", + endpoint_id, + serde_json::json!({ "name": endpoint.name }), + ) + .await; + + Ok(Json(Envelope { + success: true, + data: serde_json::json!({ "message": "存储端点已删除" }), + })) +} + +async fn endpoint_view( + state: &AppState, + endpoint: storage::StorageEndpoint, +) -> Result { + let (object_count, stored_bytes) = endpoint_usage(state, endpoint.id).await?; + Ok(StorageEndpointView { + id: endpoint.id, + name: endpoint.name, + internal_endpoint: endpoint.internal_endpoint, + public_endpoint: endpoint.public_endpoint, + bucket: endpoint.bucket, + region: endpoint.region, + access_key_hint: endpoint.access_key_hint, + credentials_configured: !endpoint.access_key_encrypted.is_empty() + && !endpoint.secret_key_encrypted.is_empty(), + force_path_style: endpoint.force_path_style, + presign_ttl_seconds: endpoint.presign_ttl_seconds, + is_active: endpoint.is_active, + last_test_at: endpoint.last_test_at, + last_test_ok: endpoint.last_test_ok, + last_test_error: endpoint.last_test_error, + object_count, + stored_bytes, + created_at: endpoint.created_at, + updated_at: endpoint.updated_at, + }) +} + +async fn endpoint_usage(state: &AppState, endpoint_id: Uuid) -> Result<(i64, i64), AppError> { + sqlx::query_as::<_, (i64, i64)>( + r#" + SELECT + (SELECT COUNT(*) FROM task_files WHERE storage_endpoint_id = $1) + + (SELECT COUNT(*) FROM tasks WHERE zip_storage_endpoint_id = $1) AS object_count, + COALESCE((SELECT SUM(compressed_size)::BIGINT FROM task_files WHERE storage_endpoint_id = $1), 0::BIGINT) + + COALESCE((SELECT SUM(zip_size)::BIGINT FROM tasks WHERE zip_storage_endpoint_id = $1), 0::BIGINT) AS stored_bytes + "#, + ) + .bind(endpoint_id) + .fetch_one(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "统计存储使用量失败").with_source(err)) +} + +async fn record_test_result( + state: &AppState, + endpoint_id: Uuid, + ok: bool, + error: Option<&str>, + admin_id: Uuid, +) -> Result<(), AppError> { + sqlx::query( + r#" + UPDATE storage_endpoints + SET last_test_at = NOW(), + last_test_ok = $2, + last_test_error = $3, + updated_at = NOW(), + updated_by = $4 + WHERE id = $1 + "#, + ) + .bind(endpoint_id) + .bind(ok) + .bind(error) + .bind(admin_id) + .execute(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "记录存储测试结果失败").with_source(err))?; + Ok(()) +} + +async fn ensure_name_available( + state: &AppState, + name: &str, + except_id: Option, +) -> Result<(), AppError> { + let exists: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM storage_endpoints WHERE deleted_at IS NULL AND lower(name) = lower($1) AND ($2::uuid IS NULL OR id <> $2))", + ) + .bind(name) + .bind(except_id) + .fetch_one(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "校验存储端点名称失败").with_source(err))?; + if exists { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "存储端点名称已存在", + )); + } + Ok(()) +} + +fn validate_name(value: &str) -> Result { + let value = value.trim(); + if value.is_empty() || value.chars().count() > 100 { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "端点名称长度必须为 1-100 个字符", + )); + } + Ok(value.to_string()) +} + +fn ensure_configurable(endpoint: &storage::StorageEndpoint) -> Result<(), AppError> { + if endpoint.deleted_at.is_some() { + return Err(AppError::new(ErrorCode::NotFound, "存储端点不存在")); + } + Ok(()) +} + +fn validate_endpoint_url(value: &str, label: &str) -> Result { + let value = value.trim().trim_end_matches('/'); + let parsed = Url::parse(value).map_err(|err| { + AppError::new(ErrorCode::InvalidRequest, format!("{label} 格式错误")).with_source(err) + })?; + if !matches!(parsed.scheme(), "http" | "https") + || !parsed.username().is_empty() + || parsed.password().is_some() + || parsed.query().is_some() + || parsed.fragment().is_some() + || parsed.host_str().is_none() + { + return Err(AppError::new( + ErrorCode::InvalidRequest, + format!("{label} 必须是无凭据、无查询参数的 HTTP(S) 地址"), + )); + } + Ok(value.to_string()) +} + +fn validate_bucket(value: &str) -> Result { + let value = value.trim(); + let valid = (3..=63).contains(&value.len()) + && value + .bytes() + .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || matches!(ch, b'.' | b'-')) + && value + .as_bytes() + .first() + .is_some_and(u8::is_ascii_alphanumeric) + && value + .as_bytes() + .last() + .is_some_and(u8::is_ascii_alphanumeric) + && !value.contains("..") + && !value.contains(".-") + && !value.contains("-."); + if !valid { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "Bucket 名称不符合 S3 命名规则", + )); + } + Ok(value.to_string()) +} + +fn validate_region(value: &str) -> Result { + let value = value.trim(); + if value.is_empty() + || value.len() > 100 + || !value + .bytes() + .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, b'-' | b'_')) + { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "Region 格式不正确", + )); + } + Ok(value.to_string()) +} + +fn validate_credential(value: &str, label: &str) -> Result { + let value = value.trim(); + if value.len() < 3 || value.len() > 512 || value.chars().any(char::is_control) { + return Err(AppError::new( + ErrorCode::InvalidRequest, + format!("{label} 长度或格式不正确"), + )); + } + Ok(value.to_string()) +} + +fn validate_presign_ttl(value: i32) -> Result { + if !(60..=3600).contains(&value) { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "签名有效期必须为 60-3600 秒", + )); + } + Ok(value) +} + +fn access_key_hint(value: &str) -> String { + let chars = value.chars().collect::>(); + if chars.len() <= 8 { + return "********".to_string(); + } + let start = chars.iter().take(4).collect::(); + let end = chars.iter().rev().take(4).rev().collect::(); + format!("{start}...{end}") +} + +fn storage_test_message(error: &AppError) -> String { + let Some(source) = error.source.as_deref() else { + return error.message.clone(); + }; + let source = source.replace(['\r', '\n'], " "); + let mut detail = format!("{}：{}", error.message, source); + if detail.len() > 500 { + let mut end = 500; + while !detail.is_char_boundary(end) { + end -= 1; + } + detail.truncate(end); + } + detail +} + +async fn audit_storage_action( + state: &AppState, + admin_id: Uuid, + ip: IpAddr, + action: &str, + endpoint_id: Uuid, + details: serde_json::Value, +) { + let _ = sqlx::query( + r#" + INSERT INTO audit_logs (user_id, action, resource_type, resource_id, details, ip_address) + VALUES ($1, $2, 'storage_endpoint', $3, $4, $5::inet) + "#, + ) + .bind(admin_id) + .bind(action) + .bind(endpoint_id) + .bind(details) + .bind(ip.to_string()) + .execute(&state.db) + .await; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn endpoint_url_rejects_embedded_credentials() { + assert!(validate_endpoint_url("https://user:pass@example.com", "Endpoint").is_err()); + assert!(validate_endpoint_url("https://files.example.com/", "Endpoint").is_ok()); + } + + #[test] + fn bucket_validation_matches_s3_basics() { + assert!(validate_bucket("imageforge-results").is_ok()); + assert!(validate_bucket("Bad_Bucket").is_err()); + assert!(validate_bucket("ab").is_err()); + } + + #[test] + fn credentials_are_masked() { + assert_eq!(access_key_hint("ABCD12345678WXYZ"), "ABCD...WXYZ"); + assert_eq!(access_key_hint("short"), "********"); + } +} diff --git a/src/api/compress.rs b/src/api/compress.rs index e18275a..4512dfb 100644 --- a/src/api/compress.rs +++ b/src/api/compress.rs @@ -7,6 +7,7 @@ use crate::services::compress; use crate::services::compress::{CompressionLevel, ImageFmt}; use crate::services::idempotency; use crate::services::quota; +use crate::services::storage; use crate::state::AppState; use axum::extract::{ConnectInfo, Multipart, State}; @@ -19,7 +20,6 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use sqlx::FromRow; use std::net::{IpAddr, SocketAddr}; -use tokio::io::AsyncWriteExt; use uuid::Uuid; pub fn router() -> Router { @@ -300,35 +300,31 @@ async fn compress_json( && req.max_height.is_none(); let charge_units = !skip_charge && compressed_size < original_size; - if charge_units { - if let QuotaContext::Anonymous { session_id, ip } = "a_ctx { - quota::consume_anonymous_units(&state, session_id, *ip, 1).await?; - } - } - - if !state.config.storage_type.eq_ignore_ascii_case("local") { - return Err(AppError::new( - ErrorCode::StorageUnavailable, - "当前仅支持本地存储（STORAGE_TYPE=local）", - )); - } - - tokio::fs::create_dir_all(&state.config.storage_path) - .await - .map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "创建存储目录失败").with_source(err) - })?; - let task_id = Uuid::new_v4(); let file_id = Uuid::new_v4(); - let file_path = format!( - "{}/{}.{}", - state.config.storage_path, - file_id, - format_out.extension() - ); + let retention_hours = retention.num_hours(); + let object_key = + storage::result_key(retention_hours, task_id, file_id, format_out.extension()); + let stored = + storage::store_bytes(&state, &object_key, compressed, format_out.content_type()) + .await?; - write_file(&file_path, &compressed).await?; + if charge_units { + if let QuotaContext::Anonymous { session_id, ip } = "a_ctx { + if let Err(err) = quota::consume_anonymous_units(&state, session_id, *ip, 1).await { + let _ = storage::delete_object( + &state, + &storage::ObjectLocator { + backend: stored.backend.clone(), + endpoint_id: stored.endpoint_id, + key: stored.key.clone(), + }, + ) + .await; + return Err(err); + } + } + } let expires_at = Utc::now() + retention; @@ -338,7 +334,7 @@ async fn compress_json( ip, task_id, file_id, - &file_path, + &stored, &req.file_name, req.max_width, req.max_height, @@ -350,12 +346,21 @@ async fn compress_json( compressed_size, saved_percent, expires_at, + retention_hours, "a_ctx, charge_units, ) .await { - let _ = tokio::fs::remove_file(&file_path).await; + let _ = storage::delete_object( + &state, + &storage::ObjectLocator { + backend: stored.backend.clone(), + endpoint_id: stored.endpoint_id, + key: stored.key.clone(), + }, + ) + .await; return Err(err); } @@ -671,27 +676,18 @@ async fn compress_direct( && req.max_height.is_none(); let charge_units = !skip_charge && compressed_size < original_size; - if !state.config.storage_type.eq_ignore_ascii_case("local") { - return Err(AppError::new( - ErrorCode::StorageUnavailable, - "当前仅支持本地存储（STORAGE_TYPE=local）", - )); - } - tokio::fs::create_dir_all(&state.config.storage_path) - .await - .map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "创建存储目录失败").with_source(err) - })?; - let task_id = Uuid::new_v4(); let file_id = Uuid::new_v4(); - let file_path = format!( - "{}/{}.{}", - state.config.storage_path, - file_id, - format_out.extension() - ); - write_file(&file_path, &compressed).await?; + let retention_hours = retention.num_hours(); + let object_key = + storage::result_key(retention_hours, task_id, file_id, format_out.extension()); + let stored = storage::store_bytes( + &state, + &object_key, + compressed.clone(), + format_out.content_type(), + ) + .await?; let expires_at = Utc::now() + retention; @@ -701,7 +697,7 @@ async fn compress_direct( ip, task_id, file_id, - &file_path, + &stored, &req.file_name, req.max_width, req.max_height, @@ -713,12 +709,21 @@ async fn compress_direct( compressed_size, saved_percent, expires_at, + retention_hours, "a_ctx, charge_units, ) .await { - let _ = tokio::fs::remove_file(&file_path).await; + let _ = storage::delete_object( + &state, + &storage::ObjectLocator { + backend: stored.backend.clone(), + endpoint_id: stored.endpoint_id, + key: stored.key.clone(), + }, + ) + .await; return Err(err); } @@ -800,19 +805,11 @@ async fn compress_direct( } } -async fn write_file(path: &str, bytes: &[u8]) -> Result<(), AppError> { - let mut file = tokio::fs::File::create(path).await.map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "写入文件失败").with_source(err) - })?; - file.write_all(bytes).await.map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "写入文件失败").with_source(err) - })?; - Ok(()) -} - #[derive(Debug, FromRow)] struct DirectReplayRow { - storage_path: Option, + storage_backend: String, + storage_endpoint_id: Option, + storage_key: Option, output_format: String, file_status: String, expires_at: DateTime, @@ -828,7 +825,9 @@ async fn load_direct_replay_bytes( sqlx::query_as::<_, DirectReplayRow>( r#" SELECT - f.storage_path, + f.storage_backend, + f.storage_endpoint_id, + COALESCE(f.storage_key, f.storage_path) AS storage_key, f.output_format, f.status::text AS file_status, t.expires_at @@ -846,7 +845,9 @@ async fn load_direct_replay_bytes( sqlx::query_as::<_, DirectReplayRow>( r#" SELECT - f.storage_path, + f.storage_backend, + f.storage_endpoint_id, + COALESCE(f.storage_key, f.storage_path) AS storage_key, f.output_format, f.status::text AS file_status, t.expires_at @@ -873,13 +874,18 @@ async fn load_direct_replay_bytes( if row.file_status != "completed" { return Err(AppError::new(ErrorCode::NotFound, "文件不存在")); } - let Some(path) = row.storage_path else { + let Some(key) = row.storage_key else { return Err(AppError::new(ErrorCode::NotFound, "文件不存在")); }; - - let bytes = tokio::fs::read(&path).await.map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "读取文件失败").with_source(err) - })?; + let bytes = storage::read_bytes( + state, + &storage::ObjectLocator { + backend: row.storage_backend, + endpoint_id: row.storage_endpoint_id, + key, + }, + ) + .await?; let fmt = compress::parse_output_format(&row.output_format)?; Ok((bytes, fmt)) @@ -1082,7 +1088,7 @@ async fn record_task_and_metering( client_ip: IpAddr, task_id: Uuid, file_id: Uuid, - file_path: &str, + stored: &storage::StoredObject, original_name: &str, max_width: Option, max_height: Option, @@ -1094,6 +1100,7 @@ async fn record_task_and_metering( compressed_size: u64, saved_percent: f64, expires_at: DateTime, + retention_hours: i64, quota_ctx: &QuotaContext, charge_units: bool, ) -> Result<(), AppError> { @@ -1123,13 +1130,13 @@ async fn record_task_and_metering( compression_rate, total_files, completed_files, failed_files, total_original_size, total_compressed_size, - started_at, completed_at, expires_at + started_at, completed_at, expires_at, retention_hours ) VALUES ( $1, $2, $3, $4, $5::inet, $6::task_source, 'completed', $7::compression_level, $8, $9, $10, $11, $12, 1, 1, 0, $13, $14, - NOW(), NOW(), $15 + NOW(), NOW(), $15, $16 ) "#, ) @@ -1148,6 +1155,7 @@ async fn record_task_and_metering( .bind(original_size as i64) .bind(compressed_size as i64) .bind(expires_at) + .bind(retention_hours as i32) .execute(&mut *tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "创建任务失败").with_source(err))?; @@ -1158,12 +1166,13 @@ async fn record_task_and_metering( id, task_id, original_name, original_format, output_format, original_size, compressed_size, saved_percent, - storage_path, status, completed_at + storage_path, storage_backend, storage_endpoint_id, storage_key, storage_etag, + status, completed_at ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, - $9, 'completed', NOW() + $9, $10, $11, $12, $13, 'completed', NOW() ) "#, ) @@ -1175,7 +1184,15 @@ async fn record_task_and_metering( .bind(original_size as i64) .bind(compressed_size as i64) .bind(saved_percent) - .bind(file_path) + .bind(if stored.backend == "local" { + Some(stored.key.as_str()) + } else { + None + }) + .bind(&stored.backend) + .bind(stored.endpoint_id) + .bind(&stored.key) + .bind(&stored.etag) .execute(&mut *tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "创建文件记录失败").with_source(err))?; diff --git a/src/api/downloads.rs b/src/api/downloads.rs index 32126ac..7b68cbe 100644 --- a/src/api/downloads.rs +++ b/src/api/downloads.rs @@ -1,11 +1,12 @@ use crate::api::context; use crate::error::{AppError, ErrorCode}; +use crate::services::storage; use crate::state::AppState; use axum::body::Body; use axum::extract::{ConnectInfo, Path, State}; use axum::http::{header, HeaderMap}; -use axum::response::{IntoResponse, Response}; +use axum::response::{IntoResponse, Redirect, Response}; use axum::routing::get; use axum::Router; use chrono::{DateTime, Utc}; @@ -25,7 +26,9 @@ pub fn router() -> Router { #[derive(Debug, FromRow)] struct DownloadRow { - storage_path: Option, + storage_backend: String, + storage_endpoint_id: Option, + storage_key: Option, output_format: String, original_name: String, file_status: String, @@ -47,7 +50,9 @@ async fn download_file( let row = sqlx::query_as::<_, DownloadRow>( r#" SELECT - f.storage_path, + f.storage_backend, + f.storage_endpoint_id, + COALESCE(f.storage_key, f.storage_path) AS storage_key, f.output_format, f.original_name, f.status::text AS file_status, @@ -75,32 +80,51 @@ async fn download_file( authorize_download(&principal, &row)?; - let Some(path) = &row.storage_path else { + let Some(key) = row.storage_key else { return Err(AppError::new(ErrorCode::NotFound, "文件不存在")); }; + let object = storage::ObjectLocator { + backend: row.storage_backend, + endpoint_id: row.storage_endpoint_id, + key, + }; + respond_object( + &state, + jar, + &object, + &output_file_name(&row.original_name, &row.output_format), + content_type(&row.output_format), + ) + .await +} - let file = tokio::fs::File::open(path).await.map_err(|err| { +async fn respond_object( + state: &AppState, + jar: axum_extra::extract::cookie::CookieJar, + object: &storage::ObjectLocator, + file_name: &str, + content_type_value: &str, +) -> Result<(axum_extra::extract::cookie::CookieJar, Response), AppError> { + if let Some(url) = + storage::presign_download(state, object, file_name, content_type_value).await? + { + return Ok((jar, Redirect::temporary(&url).into_response())); + } + + let file = tokio::fs::File::open(&object.key).await.map_err(|err| { AppError::new(ErrorCode::StorageUnavailable, "读取文件失败").with_source(err) })?; let content_length = file.metadata().await.ok().map(|metadata| metadata.len()); let body = Body::from_stream(ReaderStream::new(file)); - let mut resp_headers = HeaderMap::new(); - resp_headers.insert( - header::CONTENT_TYPE, - content_type(&row.output_format).parse().unwrap(), - ); - resp_headers.insert( - header::CONTENT_DISPOSITION, - content_disposition(&row.original_name)?, - ); + resp_headers.insert(header::CONTENT_TYPE, content_type_value.parse().unwrap()); + resp_headers.insert(header::CONTENT_DISPOSITION, content_disposition(file_name)?); if let Some(content_length) = content_length { resp_headers.insert( header::CONTENT_LENGTH, content_length.to_string().parse().unwrap(), ); } - Ok((jar, (resp_headers, body).into_response())) } @@ -126,10 +150,36 @@ fn content_type(format: &str) -> &'static str { "jpeg" | "jpg" => "image/jpeg", "webp" => "image/webp", "avif" => "image/avif", + "gif" => "image/gif", + "bmp" => "image/bmp", + "tif" | "tiff" => "image/tiff", + "ico" => "image/x-icon", _ => "application/octet-stream", } } +fn output_file_name(original_name: &str, output_format: &str) -> String { + let sanitized = sanitize_filename(original_name); + let base = sanitized + .rsplit_once('.') + .map(|(value, _)| value) + .unwrap_or(&sanitized) + .trim_end_matches('.'); + let base = if base.is_empty() { "download" } else { base }; + let extension = match output_format.trim().to_ascii_lowercase().as_str() { + "jpeg" | "jpg" => "jpg", + "png" => "png", + "webp" => "webp", + "avif" => "avif", + "gif" => "gif", + "bmp" => "bmp", + "tif" | "tiff" => "tiff", + "ico" => "ico", + _ => "bin", + }; + format!("{base}.{extension}") +} + fn sanitize_filename(name: &str) -> String { let mut out = name.trim().to_string(); if out.is_empty() { @@ -177,11 +227,17 @@ struct TaskZipRow { status: String, completed_at: Option>, expires_at: DateTime, + retention_hours: i32, + zip_storage_backend: Option, + zip_storage_endpoint_id: Option, + zip_storage_key: Option, } #[derive(Debug, FromRow)] struct TaskZipFileRow { - storage_path: Option, + storage_backend: String, + storage_endpoint_id: Option, + storage_key: Option, original_name: String, output_format: String, } @@ -203,7 +259,11 @@ async fn download_task_zip( session_id, status::text AS status, completed_at, - expires_at + expires_at, + retention_hours, + zip_storage_backend, + zip_storage_endpoint_id, + zip_storage_key FROM tasks WHERE id = $1 "#, @@ -235,26 +295,29 @@ async fn download_task_zip( } } - if !state.config.storage_type.eq_ignore_ascii_case("local") { - return Err(AppError::new( - ErrorCode::StorageUnavailable, - "当前仅支持本地存储（STORAGE_TYPE=local）", - )); - } - - let zip_dir = format!("{}/zips", state.config.storage_path); - tokio::fs::create_dir_all(&zip_dir).await.map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "创建存储目录失败").with_source(err) - })?; - let zip_path = PathBuf::from(format!("{zip_dir}/{task_id}.zip")); - - if tokio::fs::try_exists(&zip_path).await.unwrap_or(false) { - return stream_zip(jar, zip_path, task_id).await; + if let (Some(backend), Some(key)) = ( + task.zip_storage_backend.clone(), + task.zip_storage_key.clone(), + ) { + return respond_object( + &state, + jar, + &storage::ObjectLocator { + backend, + endpoint_id: task.zip_storage_endpoint_id, + key, + }, + &format!("task_{task_id}.zip"), + "application/zip", + ) + .await; } let rows = sqlx::query_as::<_, TaskZipFileRow>( r#" - SELECT storage_path, original_name, output_format + SELECT storage_backend, storage_endpoint_id, + COALESCE(storage_key, storage_path) AS storage_key, + original_name, output_format FROM task_files WHERE task_id = $1 AND status = 'completed' ORDER BY created_at ASC @@ -269,53 +332,92 @@ async fn download_task_zip( return Err(AppError::new(ErrorCode::NotFound, "没有可打包的文件")); } - let mut used_names: HashMap = HashMap::new(); - let mut entries: Vec<(String, String)> = Vec::new(); - for row in rows { - let Some(path) = row.storage_path else { - continue; - }; - let name = build_zip_entry_name(&row.original_name, &row.output_format, &mut used_names); - entries.push((name, path)); - } - if entries.is_empty() { - return Err(AppError::new(ErrorCode::NotFound, "没有可打包的文件")); - } - - let zip_path_cloned = zip_path.clone(); - let task_id_str = task_id.to_string(); - tokio::task::spawn_blocking(move || { - generate_zip_file(&zip_path_cloned, &task_id_str, &entries) - }) - .await - .map_err(|err| AppError::new(ErrorCode::Internal, "生成 ZIP 失败").with_source(err))? - .map_err(|err| AppError::new(ErrorCode::Internal, "生成 ZIP 失败").with_source(err))?; - - stream_zip(jar, zip_path, task_id).await -} - -async fn stream_zip( - jar: axum_extra::extract::cookie::CookieJar, - zip_path: PathBuf, - task_id: Uuid, -) -> Result<(axum_extra::extract::cookie::CookieJar, Response), AppError> { - let file = tokio::fs::File::open(&zip_path).await.map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "读取 ZIP 失败").with_source(err) + let temp_dir = PathBuf::from(format!( + "{}/tmp/zips/{task_id}-{}", + state.config.storage_path, + Uuid::new_v4() + )); + tokio::fs::create_dir_all(&temp_dir).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "创建 ZIP 临时目录失败").with_source(err) })?; + let zip_path = temp_dir.join(format!("task_{task_id}.zip")); - let stream = ReaderStream::new(file); - let body = Body::from_stream(stream); + let build_result: Result = async { + let mut used_names: HashMap = HashMap::new(); + let mut entries: Vec<(String, String)> = Vec::new(); + for (index, row) in rows.into_iter().enumerate() { + let Some(key) = row.storage_key else { + continue; + }; + let path = temp_dir.join(format!("entry-{index}")); + storage::download_to_file( + &state, + &storage::ObjectLocator { + backend: row.storage_backend, + endpoint_id: row.storage_endpoint_id, + key, + }, + &path, + ) + .await?; + let name = + build_zip_entry_name(&row.original_name, &row.output_format, &mut used_names); + entries.push((name, path.to_string_lossy().to_string())); + } + if entries.is_empty() { + return Err(AppError::new(ErrorCode::NotFound, "没有可打包的文件")); + } - let mut resp_headers = HeaderMap::new(); - resp_headers.insert(header::CONTENT_TYPE, "application/zip".parse().unwrap()); - resp_headers.insert( - header::CONTENT_DISPOSITION, - format!("attachment; filename=\"task_{task_id}.zip\"") - .parse() - .unwrap(), - ); + let zip_path_cloned = zip_path.clone(); + let task_id_str = task_id.to_string(); + tokio::task::spawn_blocking(move || { + generate_zip_file(&zip_path_cloned, &task_id_str, &entries) + }) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "生成 ZIP 失败").with_source(err))? + .map_err(|err| AppError::new(ErrorCode::Internal, "生成 ZIP 失败").with_source(err))?; - Ok((jar, (resp_headers, body).into_response())) + let object_key = storage::archive_key(task.retention_hours as i64, task_id); + storage::store_file(&state, &object_key, &zip_path, "application/zip").await + } + .await; + + let _ = tokio::fs::remove_dir_all(&temp_dir).await; + let stored = build_result?; + + sqlx::query( + r#" + UPDATE tasks + SET zip_storage_backend = $2, + zip_storage_endpoint_id = $3, + zip_storage_key = $4, + zip_storage_etag = $5, + zip_size = $6 + WHERE id = $1 AND zip_storage_key IS NULL + "#, + ) + .bind(task_id) + .bind(&stored.backend) + .bind(stored.endpoint_id) + .bind(&stored.key) + .bind(&stored.etag) + .bind(stored.size as i64) + .execute(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "记录 ZIP 对象失败").with_source(err))?; + + respond_object( + &state, + jar, + &storage::ObjectLocator { + backend: stored.backend, + endpoint_id: stored.endpoint_id, + key: stored.key, + }, + &format!("task_{task_id}.zip"), + "application/zip", + ) + .await } fn build_zip_entry_name( @@ -333,6 +435,10 @@ fn build_zip_entry_name( "png" => "png", "webp" => "webp", "avif" => "avif", + "gif" => "gif", + "bmp" => "bmp", + "tif" | "tiff" => "tiff", + "ico" => "ico", _ => "bin", }; @@ -411,4 +517,10 @@ mod tests { assert_eq!(sanitized, "a".repeat(119)); assert!(sanitized.is_char_boundary(sanitized.len())); } + + #[test] + fn output_file_name_matches_converted_format() { + assert_eq!(output_file_name("photo.png", "webp"), "photo.webp"); + assert_eq!(output_file_name("没有扩展名", "jpeg"), "没有扩展名.jpg"); + } } diff --git a/src/api/mod.rs b/src/api/mod.rs index 8f290a7..284016d 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,4 +1,5 @@ mod admin; +mod admin_storage; mod auth; mod billing; mod compress; @@ -62,5 +63,6 @@ fn v1_router() -> Router { .merge(webhooks::router()) .merge(user::router()) .merge(admin::router()) + .merge(admin_storage::router()) .fallback(response::not_found) } diff --git a/src/api/tasks.rs b/src/api/tasks.rs index 114f3b0..499895c 100644 --- a/src/api/tasks.rs +++ b/src/api/tasks.rs @@ -6,6 +6,7 @@ use crate::services::billing::{BillingContext, Plan}; use crate::services::compress; use crate::services::compress::{CompressionLevel, ImageFmt}; use crate::services::idempotency; +use crate::services::storage; use crate::state::AppState; use axum::extract::{ConnectInfo, Multipart, Path, State}; @@ -72,13 +73,6 @@ async fn create_batch_task( let ip = context::client_ip(&headers, addr.ip()); let (jar, principal) = context::authenticate(&state, jar, &headers, ip).await?; - if !state.config.storage_type.eq_ignore_ascii_case("local") { - return Err(AppError::new( - ErrorCode::StorageUnavailable, - "当前仅支持本地存储（STORAGE_TYPE=local）", - )); - } - let idempotency_key = headers .get("idempotency-key") .and_then(|v| v.to_str().ok()) @@ -217,13 +211,8 @@ async fn create_batch_task( } }?; - tokio::fs::create_dir_all(&state.config.storage_path) - .await - .map_err(|err| { - AppError::new(ErrorCode::StorageUnavailable, "创建存储目录失败").with_source(err) - })?; - let expires_at = Utc::now() + retention; + let retention_hours = retention.num_hours(); let (user_id, session_id, api_key_id) = match &task_owner { TaskOwner::Anonymous { session_id } => (None, Some(session_id.clone()), None), TaskOwner::User { user_id } => (Some(*user_id), None, None), @@ -248,13 +237,13 @@ async fn create_batch_task( compression_rate, total_files, completed_files, failed_files, total_original_size, total_compressed_size, - expires_at + expires_at, retention_hours ) VALUES ( $1, $2, $3, $4, $5::inet, $6::task_source, 'pending', $7::compression_level, $8, $9, $10, $11, $12, $13, 0, 0, $14, 0, - $15 + $15, $16 ) "#, ) @@ -273,6 +262,7 @@ async fn create_batch_task( .bind(files.len() as i32) .bind(total_original_size) .bind(expires_at) + .bind(retention_hours as i32) .execute(&mut *tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "创建任务失败").with_source(err))?; @@ -284,7 +274,7 @@ async fn create_batch_task( id, task_id, original_name, original_format, output_format, original_size, - storage_path, status + input_path, status ) VALUES ( $1, $2, $3, $4, $5, @@ -990,24 +980,61 @@ async fn delete_task( )); } - let paths: Vec> = - sqlx::query_scalar("SELECT storage_path FROM task_files WHERE task_id = $1") - .bind(task_id) - .fetch_all(&state.db) - .await - .map_err(|err| AppError::new(ErrorCode::Internal, "查询文件失败").with_source(err))?; + let files = sqlx::query_as::<_, TaskStorageRow>( + r#" + SELECT storage_backend, storage_endpoint_id, + COALESCE(storage_key, storage_path) AS storage_key, + input_path + FROM task_files + WHERE task_id = $1 + "#, + ) + .bind(task_id) + .fetch_all(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询文件失败").with_source(err))?; - for p in paths.into_iter().flatten() { - let _ = tokio::fs::remove_file(p).await; + for file in files { + if let Some(key) = file.storage_key { + storage::delete_object( + &state, + &storage::ObjectLocator { + backend: file.storage_backend, + endpoint_id: file.storage_endpoint_id, + key, + }, + ) + .await?; + } + if let Some(input_path) = file.input_path { + let _ = tokio::fs::remove_file(input_path).await; + } } - if state.config.storage_type.eq_ignore_ascii_case("local") { - let zip_path = format!("{}/zips/{task_id}.zip", state.config.storage_path); - let _ = tokio::fs::remove_file(zip_path).await; - let orig_dir = format!("{}/orig/{task_id}", state.config.storage_path); - let _ = tokio::fs::remove_dir_all(orig_dir).await; + let zip = sqlx::query_as::<_, TaskZipStorageRow>( + "SELECT zip_storage_backend, zip_storage_endpoint_id, zip_storage_key FROM tasks WHERE id = $1", + ) + .bind(task_id) + .fetch_one(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询 ZIP 文件失败").with_source(err))?; + if let (Some(backend), Some(key)) = (zip.zip_storage_backend, zip.zip_storage_key) { + storage::delete_object( + &state, + &storage::ObjectLocator { + backend, + endpoint_id: zip.zip_storage_endpoint_id, + key, + }, + ) + .await?; } + let legacy_zip_path = format!("{}/zips/{task_id}.zip", state.config.storage_path); + let _ = tokio::fs::remove_file(legacy_zip_path).await; + let orig_dir = format!("{}/orig/{task_id}", state.config.storage_path); + let _ = tokio::fs::remove_dir_all(orig_dir).await; + let deleted = sqlx::query("DELETE FROM tasks WHERE id = $1") .bind(task_id) .execute(&state.db) @@ -1027,6 +1054,21 @@ async fn delete_task( )) } +#[derive(Debug, FromRow)] +struct TaskStorageRow { + storage_backend: String, + storage_endpoint_id: Option, + storage_key: Option, + input_path: Option, +} + +#[derive(Debug, FromRow)] +struct TaskZipStorageRow { + zip_storage_backend: Option, + zip_storage_endpoint_id: Option, + zip_storage_key: Option, +} + fn authorize_task( principal: &context::Principal, user_id: Option, diff --git a/src/api/user.rs b/src/api/user.rs index 72a13c3..e452547 100644 --- a/src/api/user.rs +++ b/src/api/user.rs @@ -500,7 +500,7 @@ async fn list_history( status: String, output_format: String, error_message: Option, - storage_path: Option, + has_storage: bool, } let now = Utc::now(); @@ -517,7 +517,7 @@ async fn list_history( status::text AS status, output_format, error_message, - storage_path + COALESCE(storage_key, storage_path) IS NOT NULL AS has_storage FROM task_files WHERE task_id = $1 ORDER BY created_at ASC @@ -540,7 +540,7 @@ async fn list_history( output_format: file.output_format, error_message: file.error_message, download_url: if file.status == "completed" - && file.storage_path.is_some() + && file.has_storage && task.expires_at > now { Some(format!("/downloads/{}", file.id)) diff --git a/src/config.rs b/src/config.rs index 0c6ee9e..510625d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -25,7 +25,6 @@ pub struct Config { pub stripe_secret_key: Option, pub stripe_webhook_secret: Option, - pub storage_type: String, pub storage_path: String, pub allow_anonymous_upload: bool, @@ -95,13 +94,6 @@ impl Config { let stripe_secret_key = env_string("STRIPE_SECRET_KEY"); let stripe_webhook_secret = env_string("STRIPE_WEBHOOK_SECRET"); - let storage_type = env_string("STORAGE_TYPE").unwrap_or_else(|| "local".to_string()); - if !storage_type.eq_ignore_ascii_case("local") { - return Err(AppError::new( - ErrorCode::InvalidRequest, - "STORAGE_TYPE 目前仅支持 local", - )); - } let storage_path = env_string("STORAGE_PATH").unwrap_or_else(|| "./uploads".to_string()); let allow_anonymous_upload = env_bool("ALLOW_ANONYMOUS_UPLOAD").unwrap_or(true); @@ -141,7 +133,6 @@ impl Config { api_key_pepper, stripe_secret_key, stripe_webhook_secret, - storage_type, storage_path, allow_anonymous_upload, anon_max_file_size_mb, diff --git a/src/services/mod.rs b/src/services/mod.rs index aafdc10..9a21c3d 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -5,3 +5,4 @@ pub mod idempotency; pub mod mail; pub mod quota; pub mod settings; +pub mod storage; diff --git a/src/services/storage.rs b/src/services/storage.rs new file mode 100644 index 0000000..f0bd1c9 --- /dev/null +++ b/src/services/storage.rs @@ -0,0 +1,718 @@ +use crate::error::{AppError, ErrorCode}; +use crate::services::settings; +use crate::state::AppState; + +use aws_sdk_s3::config::{ + BehaviorVersion, Credentials, Region, RequestChecksumCalculation, ResponseChecksumValidation, +}; +use aws_sdk_s3::presigning::PresigningConfig; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart}; +use aws_sdk_s3::Client; +use chrono::{DateTime, Datelike, Utc}; +use sqlx::FromRow; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use uuid::Uuid; + +const MULTIPART_THRESHOLD: u64 = 64 * 1024 * 1024; +const MULTIPART_PART_SIZE: usize = 16 * 1024 * 1024; + +#[derive(Debug, Clone, FromRow)] +pub struct StorageEndpoint { + pub id: Uuid, + pub name: String, + pub internal_endpoint: String, + pub public_endpoint: String, + pub bucket: String, + pub region: String, + pub access_key_encrypted: String, + pub secret_key_encrypted: String, + pub access_key_hint: String, + pub force_path_style: bool, + pub presign_ttl_seconds: i32, + pub is_active: bool, + pub last_test_at: Option>, + pub last_test_ok: Option, + pub last_test_error: Option, + pub created_at: DateTime, + pub updated_at: DateTime, + pub deleted_at: Option>, +} + +#[derive(Debug, Clone)] +pub struct StoredObject { + pub backend: String, + pub endpoint_id: Option, + pub key: String, + pub etag: Option, + pub size: u64, +} + +#[derive(Debug, Clone)] +pub struct ObjectLocator { + pub backend: String, + pub endpoint_id: Option, + pub key: String, +} + +pub async fn list_endpoints(state: &AppState) -> Result, AppError> { + sqlx::query_as::<_, StorageEndpoint>( + r#" + SELECT id, name, internal_endpoint, public_endpoint, bucket, region, + access_key_encrypted, secret_key_encrypted, access_key_hint, + force_path_style, presign_ttl_seconds, is_active, + last_test_at, last_test_ok, last_test_error, + created_at, updated_at, deleted_at + FROM storage_endpoints + WHERE deleted_at IS NULL + ORDER BY is_active DESC, updated_at DESC + "#, + ) + .fetch_all(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询存储端点失败").with_source(err)) +} + +pub async fn get_endpoint( + state: &AppState, + endpoint_id: Uuid, +) -> Result { + sqlx::query_as::<_, StorageEndpoint>( + r#" + SELECT id, name, internal_endpoint, public_endpoint, bucket, region, + access_key_encrypted, secret_key_encrypted, access_key_hint, + force_path_style, presign_ttl_seconds, is_active, + last_test_at, last_test_ok, last_test_error, + created_at, updated_at, deleted_at + FROM storage_endpoints + WHERE id = $1 + "#, + ) + .bind(endpoint_id) + .fetch_optional(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询存储端点失败").with_source(err))? + .ok_or_else(|| AppError::new(ErrorCode::NotFound, "存储端点不存在")) +} + +pub async fn active_endpoint(state: &AppState) -> Result, AppError> { + sqlx::query_as::<_, StorageEndpoint>( + r#" + SELECT id, name, internal_endpoint, public_endpoint, bucket, region, + access_key_encrypted, secret_key_encrypted, access_key_hint, + force_path_style, presign_ttl_seconds, is_active, + last_test_at, last_test_ok, last_test_error, + created_at, updated_at, deleted_at + FROM storage_endpoints + WHERE is_active = true AND deleted_at IS NULL + LIMIT 1 + "#, + ) + .fetch_optional(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询活动存储端点失败").with_source(err)) +} + +pub fn result_key(retention_hours: i64, task_id: Uuid, file_id: Uuid, extension: &str) -> String { + let now = Utc::now(); + format!( + "results/{}/{:04}/{:02}/{task_id}/{file_id}.{}", + retention_prefix(retention_hours), + now.year(), + now.month(), + extension.trim_start_matches('.') + ) +} + +pub fn archive_key(retention_hours: i64, task_id: Uuid) -> String { + let now = Utc::now(); + format!( + "archives/{}/{:04}/{:02}/{task_id}.zip", + retention_prefix(retention_hours), + now.year(), + now.month() + ) +} + +fn retention_prefix(hours: i64) -> String { + match hours { + 0..=24 => "1d".to_string(), + 25..=168 => "7d".to_string(), + 169..=360 => "15d".to_string(), + value => format!("custom-{}h", value.max(1)), + } +} + +pub async fn store_bytes( + state: &AppState, + key: &str, + bytes: Vec, + content_type: &str, +) -> Result { + if let Some(endpoint) = active_endpoint(state).await? { + let client = client_for(state, &endpoint, EndpointKind::Internal)?; + let size = bytes.len() as u64; + let output = client + .put_object() + .bucket(&endpoint.bucket) + .key(key) + .content_type(content_type) + .body(ByteStream::from(bytes)) + .send() + .await + .map_err(|err| storage_error("上传 S3 对象失败", err))?; + + return Ok(StoredObject { + backend: "s3".to_string(), + endpoint_id: Some(endpoint.id), + key: key.to_string(), + etag: output.e_tag().map(ToOwned::to_owned), + size, + }); + } + + let path = local_path(state, key)?; + if let Some(parent) = path.parent() { + tokio::fs::create_dir_all(parent).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "创建本地存储目录失败").with_source(err) + })?; + } + tokio::fs::write(&path, &bytes).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "写入本地存储失败").with_source(err) + })?; + + Ok(StoredObject { + backend: "local".to_string(), + endpoint_id: None, + key: path.to_string_lossy().to_string(), + etag: None, + size: bytes.len() as u64, + }) +} + +pub async fn store_file( + state: &AppState, + key: &str, + path: &Path, + content_type: &str, +) -> Result { + let metadata = tokio::fs::metadata(path).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "读取待上传文件失败").with_source(err) + })?; + + if let Some(endpoint) = active_endpoint(state).await? { + let client = client_for(state, &endpoint, EndpointKind::Internal)?; + let etag = if metadata.len() >= MULTIPART_THRESHOLD { + multipart_upload(&client, &endpoint.bucket, key, path, content_type).await? + } else { + let body = ByteStream::from_path(path).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "读取待上传文件失败").with_source(err) + })?; + client + .put_object() + .bucket(&endpoint.bucket) + .key(key) + .content_type(content_type) + .body(body) + .send() + .await + .map_err(|err| storage_error("上传 S3 对象失败", err))? + .e_tag() + .map(ToOwned::to_owned) + }; + + return Ok(StoredObject { + backend: "s3".to_string(), + endpoint_id: Some(endpoint.id), + key: key.to_string(), + etag, + size: metadata.len(), + }); + } + + let destination = local_path(state, key)?; + if let Some(parent) = destination.parent() { + tokio::fs::create_dir_all(parent).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "创建本地存储目录失败").with_source(err) + })?; + } + tokio::fs::copy(path, &destination).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "保存本地文件失败").with_source(err) + })?; + + Ok(StoredObject { + backend: "local".to_string(), + endpoint_id: None, + key: destination.to_string_lossy().to_string(), + etag: None, + size: metadata.len(), + }) +} + +pub async fn read_bytes(state: &AppState, object: &ObjectLocator) -> Result, AppError> { + if object.backend == "local" { + return tokio::fs::read(&object.key).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "读取本地文件失败").with_source(err) + }); + } + + let endpoint = endpoint_for_object(state, object).await?; + let client = client_for(state, &endpoint, EndpointKind::Internal)?; + let response = client + .get_object() + .bucket(&endpoint.bucket) + .key(&object.key) + .send() + .await + .map_err(|err| storage_error("读取 S3 对象失败", err))?; + let bytes = response + .body + .collect() + .await + .map_err(|err| storage_error("接收 S3 对象失败", err))?; + Ok(bytes.into_bytes().to_vec()) +} + +pub async fn download_to_file( + state: &AppState, + object: &ObjectLocator, + destination: &Path, +) -> Result<(), AppError> { + if object.backend == "local" { + tokio::fs::copy(&object.key, destination) + .await + .map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "复制本地文件失败").with_source(err) + })?; + return Ok(()); + } + + let endpoint = endpoint_for_object(state, object).await?; + let client = client_for(state, &endpoint, EndpointKind::Internal)?; + let response = client + .get_object() + .bucket(&endpoint.bucket) + .key(&object.key) + .send() + .await + .map_err(|err| storage_error("读取 S3 对象失败", err))?; + let mut source = response.body.into_async_read(); + let mut output = tokio::fs::File::create(destination).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "创建临时文件失败").with_source(err) + })?; + tokio::io::copy(&mut source, &mut output) + .await + .map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "保存临时文件失败").with_source(err) + })?; + output.flush().await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "刷新临时文件失败").with_source(err) + })?; + Ok(()) +} + +pub async fn presign_download( + state: &AppState, + object: &ObjectLocator, + file_name: &str, + content_type: &str, +) -> Result, AppError> { + if object.backend == "local" { + return Ok(None); + } + + let endpoint = endpoint_for_object(state, object).await?; + let client = client_for(state, &endpoint, EndpointKind::Public)?; + let ttl = Duration::from_secs(endpoint.presign_ttl_seconds as u64); + let config = PresigningConfig::expires_in(ttl).map_err(|err| { + AppError::new(ErrorCode::Internal, "生成下载签名配置失败").with_source(err) + })?; + let disposition = content_disposition(file_name); + let request = client + .get_object() + .bucket(&endpoint.bucket) + .key(&object.key) + .response_content_disposition(disposition) + .response_content_type(content_type) + .presigned(config) + .await + .map_err(|err| storage_error("生成 S3 下载地址失败", err))?; + Ok(Some(request.uri().to_string())) +} + +pub async fn delete_object(state: &AppState, object: &ObjectLocator) -> Result<(), AppError> { + if object.backend == "local" { + match tokio::fs::remove_file(&object.key).await { + Ok(()) => return Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(err) => { + return Err( + AppError::new(ErrorCode::StorageUnavailable, "删除本地文件失败") + .with_source(err), + ) + } + } + } + + let endpoint = endpoint_for_object(state, object).await?; + let client = client_for(state, &endpoint, EndpointKind::Internal)?; + client + .delete_object() + .bucket(&endpoint.bucket) + .key(&object.key) + .send() + .await + .map_err(|err| storage_error("删除 S3 对象失败", err))?; + Ok(()) +} + +pub async fn test_endpoint(state: &AppState, endpoint: &StorageEndpoint) -> Result<(), AppError> { + let client = client_for(state, endpoint, EndpointKind::Internal)?; + client + .head_bucket() + .bucket(&endpoint.bucket) + .send() + .await + .map_err(|err| storage_error("S3 Bucket 访问失败", err))?; + + let key = format!(".imageforge-health/{}.txt", Uuid::new_v4()); + let expected = b"imageforge-storage-check".to_vec(); + client + .put_object() + .bucket(&endpoint.bucket) + .key(&key) + .content_type("text/plain") + .body(ByteStream::from(expected.clone())) + .send() + .await + .map_err(|err| storage_error("S3 写入测试失败", err))?; + + let result = async { + let response = client + .get_object() + .bucket(&endpoint.bucket) + .key(&key) + .send() + .await + .map_err(|err| storage_error("S3 读取测试失败", err))?; + let actual = response + .body + .collect() + .await + .map_err(|err| storage_error("S3 测试对象接收失败", err))? + .into_bytes(); + if actual.as_ref() != expected.as_slice() { + return Err(AppError::new( + ErrorCode::StorageUnavailable, + "S3 读写校验内容不一致", + )); + } + + let public_client = client_for(state, endpoint, EndpointKind::Public)?; + let signed = public_client + .get_object() + .bucket(&endpoint.bucket) + .key(&key) + .presigned( + PresigningConfig::expires_in(Duration::from_secs(60)).map_err(|err| { + AppError::new(ErrorCode::Internal, "生成公网下载测试签名失败").with_source(err) + })?, + ) + .await + .map_err(|err| storage_error("生成公网下载测试地址失败", err))?; + let response = reqwest::Client::builder() + .timeout(Duration::from_secs(15)) + .build() + .map_err(|err| storage_error("创建公网下载测试客户端失败", err))? + .get(signed.uri().to_string()) + .header(reqwest::header::USER_AGENT, "ImageForge-Storage-Check/1.0") + .send() + .await + .map_err(|err| public_test_request_error(endpoint, err))?; + let status = response.status(); + if !status.is_success() { + return Err(AppError::new( + ErrorCode::StorageUnavailable, + format!("公网 Endpoint 下载测试返回 HTTP {status}"), + )); + } + let public_bytes = response.bytes().await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "读取公网下载测试响应失败") + .with_source(format!("response body error: {}", err.is_timeout())) + })?; + if public_bytes.as_ref() != expected.as_slice() { + return Err(AppError::new( + ErrorCode::StorageUnavailable, + "公网 Endpoint 下载内容校验不一致", + )); + } + Ok(()) + } + .await; + + let delete_result = client + .delete_object() + .bucket(&endpoint.bucket) + .key(&key) + .send() + .await; + + result?; + delete_result.map_err(|err| storage_error("S3 删除测试失败", err))?; + Ok(()) +} + +fn client_for( + state: &AppState, + endpoint: &StorageEndpoint, + kind: EndpointKind, +) -> Result { + let access_key = + settings::decrypt_secret(state, &endpoint.access_key_encrypted).map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "存储端点凭据不可用").with_source(err) + })?; + let secret_key = + settings::decrypt_secret(state, &endpoint.secret_key_encrypted).map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "存储端点凭据不可用").with_source(err) + })?; + let endpoint_url = match kind { + EndpointKind::Internal => &endpoint.internal_endpoint, + EndpointKind::Public => &endpoint.public_endpoint, + }; + Ok(build_client( + access_key, + secret_key, + endpoint_url, + &endpoint.region, + endpoint.force_path_style, + )) +} + +fn build_client( + access_key: String, + secret_key: String, + endpoint_url: &str, + region: &str, + force_path_style: bool, +) -> Client { + let config = aws_sdk_s3::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .credentials_provider(Credentials::new( + access_key, + secret_key, + None, + None, + "imageforge-admin", + )) + .region(Region::new(region.to_string())) + .endpoint_url(endpoint_url) + .force_path_style(force_path_style) + .request_checksum_calculation(RequestChecksumCalculation::WhenRequired) + .response_checksum_validation(ResponseChecksumValidation::WhenRequired) + .build(); + Client::from_conf(config) +} + +async fn endpoint_for_object( + state: &AppState, + object: &ObjectLocator, +) -> Result { + let endpoint_id = object + .endpoint_id + .ok_or_else(|| AppError::new(ErrorCode::StorageUnavailable, "S3 对象缺少存储端点标识"))?; + get_endpoint(state, endpoint_id).await +} + +fn local_path(state: &AppState, key: &str) -> Result { + if key.is_empty() + || key.starts_with('/') + || key.starts_with('\\') + || key.split('/').any(|part| part == "..") + { + return Err(AppError::new(ErrorCode::InvalidRequest, "非法存储对象键")); + } + Ok(Path::new(&state.config.storage_path).join(key)) +} + +fn content_disposition(file_name: &str) -> String { + let ascii = file_name + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '-' | '_') { + ch + } else { + '_' + } + }) + .collect::(); + let ascii = if ascii.is_empty() { "download" } else { &ascii }; + let encoded = + percent_encoding::utf8_percent_encode(file_name, percent_encoding::NON_ALPHANUMERIC); + format!("attachment; filename=\"{ascii}\"; filename*=UTF-8''{encoded}") +} + +async fn multipart_upload( + client: &Client, + bucket: &str, + key: &str, + path: &Path, + content_type: &str, +) -> Result, AppError> { + let created = client + .create_multipart_upload() + .bucket(bucket) + .key(key) + .content_type(content_type) + .send() + .await + .map_err(|err| storage_error("创建 S3 分片上传失败", err))?; + let upload_id = created + .upload_id() + .ok_or_else(|| AppError::new(ErrorCode::StorageUnavailable, "S3 未返回分片上传标识"))?; + + let result = async { + let mut file = tokio::fs::File::open(path).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "读取待上传文件失败").with_source(err) + })?; + let mut parts = Vec::new(); + let mut part_number = 1; + + loop { + let mut buffer = vec![0u8; MULTIPART_PART_SIZE]; + let mut filled = 0; + while filled < buffer.len() { + let read = file.read(&mut buffer[filled..]).await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "读取上传分片失败") + .with_source(err) + })?; + if read == 0 { + break; + } + filled += read; + } + if filled == 0 { + break; + } + buffer.truncate(filled); + let uploaded = client + .upload_part() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .part_number(part_number) + .body(ByteStream::from(buffer)) + .send() + .await + .map_err(|err| storage_error("上传 S3 分片失败", err))?; + let part = CompletedPart::builder() + .part_number(part_number) + .set_e_tag(uploaded.e_tag().map(ToOwned::to_owned)) + .build(); + parts.push(part); + part_number += 1; + } + + let completed = CompletedMultipartUpload::builder() + .set_parts(Some(parts)) + .build(); + let response = client + .complete_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .multipart_upload(completed) + .send() + .await + .map_err(|err| storage_error("完成 S3 分片上传失败", err))?; + Ok(response.e_tag().map(ToOwned::to_owned)) + } + .await; + + if result.is_err() { + let _ = client + .abort_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .send() + .await; + } + result +} + +fn storage_error(message: &'static str, err: impl std::fmt::Display) -> AppError { + AppError::new(ErrorCode::StorageUnavailable, message).with_source(err) +} + +fn public_test_request_error(endpoint: &StorageEndpoint, err: reqwest::Error) -> AppError { + AppError::new(ErrorCode::StorageUnavailable, "公网 Endpoint 下载测试失败").with_source(format!( + "endpoint={}, connect={}, timeout={}", + endpoint.public_endpoint, + err.is_connect(), + err.is_timeout() + )) +} + +#[derive(Clone, Copy)] +enum EndpointKind { + Internal, + Public, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retention_prefixes_match_product_policy() { + assert_eq!(retention_prefix(24), "1d"); + assert_eq!(retention_prefix(168), "7d"); + assert_eq!(retention_prefix(360), "15d"); + assert_eq!(retention_prefix(720), "custom-720h"); + } + + #[test] + fn generated_keys_are_lifecycle_scoped() { + let task_id = Uuid::nil(); + let file_id = Uuid::from_u128(1); + let key = result_key(168, task_id, file_id, "webp"); + assert!(key.starts_with("results/7d/")); + assert!(key.ends_with("/00000000-0000-0000-0000-000000000001.webp")); + assert!(archive_key(360, task_id).starts_with("archives/15d/")); + } + + #[test] + fn content_disposition_never_injects_headers() { + let value = content_disposition("a\r\nX-Test: yes.png"); + assert!(!value.contains('\r')); + assert!(!value.contains('\n')); + assert!(value.contains("filename*=UTF-8''")); + } + + #[tokio::test] + async fn presigning_uses_public_path_style_endpoint() { + let client = build_client( + "GKTESTACCESSKEY".to_string(), + "test-secret-key".to_string(), + "https://files.example.com", + "garage", + true, + ); + let request = client + .get_object() + .bucket("imageforge-results") + .key("results/1d/test.webp") + .presigned(PresigningConfig::expires_in(Duration::from_secs(300)).unwrap()) + .await + .unwrap(); + let uri = request.uri().to_string(); + assert!( + uri.starts_with("https://files.example.com/imageforge-results/results/1d/test.webp?") + ); + assert!(uri.contains("X-Amz-Algorithm=AWS4-HMAC-SHA256")); + assert!(uri.contains("X-Amz-Expires=300")); + assert!(uri.contains("X-Amz-Signature=")); + } +} diff --git a/src/worker/mod.rs b/src/worker/mod.rs index 430d7a1..371fd9a 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -2,6 +2,7 @@ use crate::error::{AppError, ErrorCode}; use crate::services::billing; use crate::services::compress; use crate::services::quota; +use crate::services::storage; use crate::state::AppState; use redis::streams::StreamReadOptions; @@ -148,17 +149,33 @@ struct TaskProcRow { api_key_id: Option, source: String, client_ip: Option, + retention_hours: i32, } #[derive(Debug, FromRow)] struct TaskFileProcRow { id: Uuid, - storage_path: Option, + input_path: Option, original_format: String, output_format: String, status: String, } +#[derive(Debug, FromRow)] +struct CleanupFileRow { + storage_backend: String, + storage_endpoint_id: Option, + storage_key: Option, + input_path: Option, +} + +#[derive(Debug, FromRow)] +struct CleanupZipRow { + zip_storage_backend: Option, + zip_storage_endpoint_id: Option, + zip_storage_key: Option, +} + #[derive(Clone)] struct TaskContext { api_key_id: Option, @@ -167,6 +184,7 @@ struct TaskContext { session_id: Option, anon_ip: Option, is_anonymous: bool, + retention_hours: i32, } async fn process_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> { @@ -183,7 +201,8 @@ async fn process_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> { session_id, api_key_id, source::text AS source, - host(client_ip) AS client_ip + host(client_ip) AS client_ip, + retention_hours FROM tasks WHERE id = $1 "#, @@ -241,7 +260,7 @@ async fn process_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> { r#" SELECT id, - storage_path, + input_path, original_format, output_format, status::text AS status @@ -273,6 +292,7 @@ async fn process_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> { session_id: task.session_id.clone(), anon_ip, is_anonymous: task.user_id.is_none(), + retention_hours: task.retention_hours, }; let concurrency = state.config.worker_concurrency.max(1) as usize; @@ -391,7 +411,7 @@ async fn process_task_file( return Ok(()); } - let Some(input_path) = file.storage_path.clone() else { + let Some(input_path) = file.input_path.clone() else { mark_file_failed(&state, task_id, file.id, "原文件不存在").await?; return Ok(()); }; @@ -462,40 +482,50 @@ async fn process_task_file( && max_height.is_none(); let charge_units = !skip_charge && compressed_size < original_size; + let object_key = storage::result_key( + ctx.retention_hours as i64, + task_id, + file.id, + format_out.extension(), + ); + let stored = match storage::store_bytes( + &state, + &object_key, + compressed, + format_out.content_type(), + ) + .await + { + Ok(value) => value, + Err(err) => { + reset_file_for_retry(&state, file.id, "对象存储暂时不可用").await?; + return Err(err); + } + }; + if ctx.is_anonymous && charge_units { let Some(session_id) = ctx.session_id.as_deref() else { + let _ = storage::delete_object(&state, &stored_locator(&stored)).await; mark_file_failed(&state, task_id, file.id, "匿名任务缺少 session_id").await?; let _ = tokio::fs::remove_file(&input_path).await; return Ok(()); }; let Some(ip) = ctx.anon_ip else { + let _ = storage::delete_object(&state, &stored_locator(&stored)).await; mark_file_failed(&state, task_id, file.id, "匿名任务缺少 client_ip").await?; let _ = tokio::fs::remove_file(&input_path).await; return Ok(()); }; if let Err(err) = quota::consume_anonymous_units(&state, session_id, ip, 1).await { + let _ = storage::delete_object(&state, &stored_locator(&stored)).await; mark_file_failed(&state, task_id, file.id, &err.message).await?; let _ = tokio::fs::remove_file(&input_path).await; return Ok(()); } } - let output_path = format!( - "{}/{}.{}", - state.config.storage_path, - file.id, - format_out.extension() - ); - if let Err(err) = tokio::fs::write(&output_path, &compressed).await { - mark_file_failed(&state, task_id, file.id, "写入压缩文件失败").await?; - let _ = tokio::fs::remove_file(&input_path).await; - return Err( - AppError::new(ErrorCode::StorageUnavailable, "写入压缩文件失败").with_source(err), - ); - } - if is_task_cancelled(&state, task_id).await? { - let _ = tokio::fs::remove_file(&output_path).await; + let _ = storage::delete_object(&state, &stored_locator(&stored)).await; mark_file_failed(&state, task_id, file.id, "已取消").await?; let _ = tokio::fs::remove_file(&input_path).await; return Ok(()); @@ -508,7 +538,7 @@ async fn process_task_file( &ctx.source, task_id, file.id, - &output_path, + &stored, original_size as i64, compressed_size as i64, saved_percent, @@ -518,12 +548,8 @@ async fn process_task_file( ) .await { - if err.code == ErrorCode::QuotaExceeded { - let _ = tokio::fs::remove_file(&output_path).await; - mark_file_failed(&state, task_id, file.id, &err.message).await?; - } else { - mark_file_failed(&state, task_id, file.id, &err.message).await?; - } + let _ = storage::delete_object(&state, &stored_locator(&stored)).await; + mark_file_failed(&state, task_id, file.id, &err.message).await?; let _ = tokio::fs::remove_file(&input_path).await; return Ok(()); } @@ -540,7 +566,7 @@ async fn finalize_file( source: &str, task_id: Uuid, task_file_id: Uuid, - output_path: &str, + stored: &storage::StoredObject, bytes_in: i64, bytes_out: i64, saved_percent: f64, @@ -577,15 +603,28 @@ async fn finalize_file( r#" UPDATE task_files SET storage_path = $2, - compressed_size = $3, - saved_percent = $4, + storage_backend = $3, + storage_endpoint_id = $4, + storage_key = $5, + storage_etag = $6, + input_path = NULL, + compressed_size = $7, + saved_percent = $8, status = 'completed', completed_at = NOW() WHERE id = $1 "#, ) .bind(task_file_id) - .bind(output_path) + .bind(if stored.backend == "local" { + Some(stored.key.as_str()) + } else { + None + }) + .bind(&stored.backend) + .bind(stored.endpoint_id) + .bind(&stored.key) + .bind(&stored.etag) .bind(bytes_out) .bind(saved_percent) .execute(&mut *tx) @@ -631,6 +670,7 @@ async fn mark_file_failed( SET status = 'failed', error_message = $2, storage_path = NULL, + input_path = NULL, completed_at = NOW() WHERE id = $1 AND status NOT IN ('completed', 'failed') @@ -663,6 +703,22 @@ async fn mark_file_failed( Ok(()) } +async fn reset_file_for_retry( + state: &AppState, + task_file_id: Uuid, + message: &str, +) -> Result<(), AppError> { + sqlx::query( + "UPDATE task_files SET status = 'pending', error_message = $2 WHERE id = $1 AND status = 'processing'", + ) + .bind(task_file_id) + .bind(message) + .execute(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "恢复待重试文件失败").with_source(err))?; + Ok(()) +} + async fn finalize_task_status(state: &AppState, task_id: Uuid) -> Result<(), AppError> { let row: Option<(i32, i32, i32, String)> = sqlx::query_as( "SELECT total_files, completed_files, failed_files, status::text AS status FROM tasks WHERE id = $1", @@ -677,7 +733,7 @@ async fn finalize_task_status(state: &AppState, task_id: Uuid) -> Result<(), App }; if status == "cancelled" { let paths: Vec> = sqlx::query_scalar( - "SELECT storage_path FROM task_files WHERE task_id = $1 AND status IN ('pending','processing')", + "SELECT input_path FROM task_files WHERE task_id = $1 AND status IN ('pending','processing')", ) .bind(task_id) .fetch_all(&state.db) @@ -688,7 +744,7 @@ async fn finalize_task_status(state: &AppState, task_id: Uuid) -> Result<(), App } let _ = sqlx::query( - "UPDATE task_files SET status = 'failed', error_message = '已取消', storage_path = NULL, completed_at = NOW() WHERE task_id = $1 AND status IN ('pending','processing')", + "UPDATE task_files SET status = 'failed', error_message = '已取消', storage_path = NULL, input_path = NULL, completed_at = NOW() WHERE task_id = $1 AND status IN ('pending','processing')", ) .bind(task_id) .execute(&state.db) @@ -810,10 +866,47 @@ async fn charge_one_unit( async fn maintenance(state: &AppState) -> Result<(), AppError> { cleanup_expired_tasks(state).await?; + cleanup_stale_zip_temp(state).await?; cleanup_expired_records(state).await?; Ok(()) } +async fn cleanup_stale_zip_temp(state: &AppState) -> Result<(), AppError> { + let root = std::path::Path::new(&state.config.storage_path).join("tmp/zips"); + let mut entries = match tokio::fs::read_dir(&root).await { + Ok(entries) => entries, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(err) => { + return Err( + AppError::new(ErrorCode::StorageUnavailable, "读取 ZIP 临时目录失败") + .with_source(err), + ) + } + }; + let cutoff = std::time::SystemTime::now() + .checked_sub(std::time::Duration::from_secs(6 * 60 * 60)) + .unwrap_or(std::time::UNIX_EPOCH); + + while let Some(entry) = entries.next_entry().await.map_err(|err| { + AppError::new(ErrorCode::StorageUnavailable, "遍历 ZIP 临时目录失败").with_source(err) + })? { + let metadata = match entry.metadata().await { + Ok(metadata) => metadata, + Err(_) => continue, + }; + if !metadata.is_dir() + || metadata + .modified() + .map(|time| time >= cutoff) + .unwrap_or(true) + { + continue; + } + let _ = tokio::fs::remove_dir_all(entry.path()).await; + } + Ok(()) +} + async fn cleanup_expired_records(state: &AppState) -> Result<(), AppError> { let _ = sqlx::query("DELETE FROM idempotency_keys WHERE expires_at < NOW()") .execute(&state.db) @@ -834,6 +927,17 @@ async fn cleanup_expired_records(state: &AppState) -> Result<(), AppError> { .execute(&state.db) .await; + let _ = sqlx::query( + r#" + DELETE FROM storage_endpoints e + WHERE e.deleted_at < NOW() - INTERVAL '30 days' + AND NOT EXISTS (SELECT 1 FROM task_files f WHERE f.storage_endpoint_id = e.id) + AND NOT EXISTS (SELECT 1 FROM tasks t WHERE t.zip_storage_endpoint_id = e.id) + "#, + ) + .execute(&state.db) + .await; + Ok(()) } @@ -848,33 +952,87 @@ async fn cleanup_expired_tasks(state: &AppState) -> Result<(), AppError> { return Ok(()); } - if state.config.storage_type.eq_ignore_ascii_case("local") { - for task_id in &task_ids { - let paths: Vec> = - sqlx::query_scalar("SELECT storage_path FROM task_files WHERE task_id = $1") - .bind(task_id) - .fetch_all(&state.db) - .await - .unwrap_or_default(); - - for p in paths.into_iter().flatten() { - let _ = tokio::fs::remove_file(p).await; - } - - let zip_path = format!("{}/zips/{task_id}.zip", state.config.storage_path); - let _ = tokio::fs::remove_file(zip_path).await; - - let orig_dir = format!("{}/orig/{task_id}", state.config.storage_path); - let _ = tokio::fs::remove_dir_all(orig_dir).await; - } - } - for task_id in task_ids { - let _ = sqlx::query("DELETE FROM tasks WHERE id = $1") - .bind(task_id) - .execute(&state.db) - .await; + if let Err(err) = cleanup_expired_task(state, task_id).await { + tracing::warn!(task_id = %task_id, error = %err, "expired task cleanup deferred"); + } } Ok(()) } + +async fn cleanup_expired_task(state: &AppState, task_id: Uuid) -> Result<(), AppError> { + let files: Vec = sqlx::query_as( + r#" + SELECT storage_backend, storage_endpoint_id, + COALESCE(storage_key, storage_path) AS storage_key, + input_path + FROM task_files + WHERE task_id = $1 + "#, + ) + .bind(task_id) + .fetch_all(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询过期任务文件失败").with_source(err))?; + + for file in files { + if let Some(key) = file.storage_key { + storage::delete_object( + state, + &storage::ObjectLocator { + backend: file.storage_backend, + endpoint_id: file.storage_endpoint_id, + key, + }, + ) + .await?; + } + if let Some(input_path) = file.input_path { + let _ = tokio::fs::remove_file(input_path).await; + } + } + + let zip: Option = sqlx::query_as( + "SELECT zip_storage_backend, zip_storage_endpoint_id, zip_storage_key FROM tasks WHERE id = $1", + ) + .bind(task_id) + .fetch_optional(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询过期 ZIP 失败").with_source(err))?; + if let Some(zip) = zip { + if let (Some(backend), Some(key)) = (zip.zip_storage_backend, zip.zip_storage_key) { + storage::delete_object( + state, + &storage::ObjectLocator { + backend, + endpoint_id: zip.zip_storage_endpoint_id, + key, + }, + ) + .await?; + } + } + + let legacy_zip_path = format!("{}/zips/{task_id}.zip", state.config.storage_path); + let _ = tokio::fs::remove_file(legacy_zip_path).await; + let orig_dir = format!("{}/orig/{task_id}", state.config.storage_path); + let _ = tokio::fs::remove_dir_all(orig_dir).await; + + sqlx::query("DELETE FROM tasks WHERE id = $1 AND expires_at < NOW()") + .bind(task_id) + .execute(&state.db) + .await + .map_err(|err| { + AppError::new(ErrorCode::Internal, "删除过期任务记录失败").with_source(err) + })?; + Ok(()) +} + +fn stored_locator(stored: &storage::StoredObject) -> storage::ObjectLocator { + storage::ObjectLocator { + backend: stored.backend.clone(), + endpoint_id: stored.endpoint_id, + key: stored.key.clone(), + } +}