diff --git a/.github/workflows/check-generic.yml b/.github/workflows/check-generic.yml index a22c8c44a..5b614cf67 100644 --- a/.github/workflows/check-generic.yml +++ b/.github/workflows/check-generic.yml @@ -20,7 +20,7 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 - - uses: taiki-e/install-action@b5fddbb5361bce8a06fb168c9d403a6cc552b084 # v2.75.29 + - uses: taiki-e/install-action@fd2f5e3d644b484055ebf4268f474c565f148f25 # v2.81.9 with: tool: tombi - run: tombi lint diff --git a/.github/workflows/labrinth-docker.yml b/.github/workflows/labrinth-docker.yml index 3ecc90fea..ca249994b 100644 --- a/.github/workflows/labrinth-docker.yml +++ b/.github/workflows/labrinth-docker.yml @@ -99,6 +99,12 @@ jobs: - name: Setup mold uses: rui314/setup-mold@9c9c13bf4c3f1adef0cc596abc155580bcb04444 # v1 / Mold 2.41.0 + - name: Install build dependencies + uses: awalsh128/cache-apt-pkgs-action@acb598e5ddbc6f68a970c5da0688d2f3a9f04d05 # v1.6.0 + with: + packages: cmake libcurl4-openssl-dev + version: v2 # cache key + - name: Cache Cargo registry and index uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5 with: diff --git a/.github/workflows/turbo-ci.yml b/.github/workflows/turbo-ci.yml index a31fcb8bc..db0340dae 100644 --- a/.github/workflows/turbo-ci.yml +++ b/.github/workflows/turbo-ci.yml @@ -99,8 +99,8 @@ jobs: - name: Install build dependencies uses: awalsh128/cache-apt-pkgs-action@acb598e5ddbc6f68a970c5da0688d2f3a9f04d05 # v1.6.0 with: - packages: libwebkit2gtk-4.1-dev libayatana-appindicator3-dev librsvg2-dev - version: v1 # cache key + packages: cmake libcurl4-openssl-dev libwebkit2gtk-4.1-dev libayatana-appindicator3-dev librsvg2-dev + version: v2 # cache key - name: Setup Node uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 @@ -153,7 +153,7 @@ jobs: uses: cargo-bins/cargo-binstall@dc19f1e48450eefe5a29b8da6c6b00a87d730b37 # v1.18.1 - name: Setup nextest - run: cargo binstall --no-confirm --secure cargo-nextest@0.9.133 + run: cargo binstall --no-confirm --secure --locked cargo-nextest@0.9.133 # cargo-binstall does not have pre-built binaries for sqlx-cli, so we fall # back to a cached cargo install diff --git a/Cargo.lock b/Cargo.lock index 907395468..56812adc3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4817,7 +4817,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b0f83760fb341a774ed326568e19f5a863af4a952def8c39f9ab92fd95b88e5" dependencies = [ "equivalent", - "hashbrown 0.15.5", + "hashbrown 0.16.0", "serde", "serde_core", ] @@ -5277,6 +5277,7 @@ dependencies = [ "quick-xml 0.38.3", "rand 0.8.5", "rand_chacha 0.3.1", + "rdkafka", "redis", "regex", "reqwest 0.12.24", @@ -5489,6 +5490,18 @@ dependencies = [ "zlib-rs", ] +[[package]] +name = "libz-sys" +version = "1.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85bc9657773828b90eeb625adff10eeac83cc21bbfd8e23a03eaa8a33c9e28d9" +dependencies = [ + "cc", + "libc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linked-hash-map" version = "0.5.6" @@ -7967,6 +7980,37 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rdkafka" +version = "0.36.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1beea247b9a7600a81d4cc33f659ce1a77e1988323d7d2809c7ed1c21f4c316d" +dependencies = [ + "futures-channel", + "futures-util", + "libc", + "log", + "rdkafka-sys", + "serde", + "serde_derive", + "serde_json", + "slab", + "tokio", +] + +[[package]] +name = "rdkafka-sys" +version = "4.10.0+2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e234cf318915c1059d4921ef7f75616b5219b10b46e9f3a511a15eb4b56a3f77" +dependencies = [ + "cmake", + "libc", + "libz-sys", + "num_enum", + "pkg-config", +] + [[package]] name = "redis" version = "0.32.7" diff --git a/Cargo.toml b/Cargo.toml index 7ce30a47f..29e6d78cc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -140,6 +140,7 @@ quick-xml = "0.38.3" quote = { version = "1.0" } rand = "=0.8.5" # Locked on 0.8 until argon2 and p256 update to 0.9 rand_chacha = "=0.3.1" # Locked on 0.3 until we can update rand to 0.9 +rdkafka = { version = "0.36.2", features = ["cmake-build"] } redis = "0.32.7" regex = "1.12.2" reqwest = { version = "0.12.24", default-features = false } diff --git a/apps/labrinth/.env.docker-compose b/apps/labrinth/.env.docker-compose index 045cfd800..097a47c64 100644 --- a/apps/labrinth/.env.docker-compose +++ b/apps/labrinth/.env.docker-compose @@ -33,6 +33,9 @@ REDIS_URL=redis://labrinth-redis REDIS_MIN_CONNECTIONS=0 REDIS_MAX_CONNECTIONS=10000 +KAFKA_BOOTSTRAP_SERVERS=redpanda:9092 +KAFKA_CLIENT_ID=labrinth + BIND_ADDR=0.0.0.0:8000 SELF_ADDR=http://labrinth:8000 diff --git a/apps/labrinth/.env.local b/apps/labrinth/.env.local index 19a7d9528..7925e52a1 100644 --- a/apps/labrinth/.env.local +++ b/apps/labrinth/.env.local @@ -51,6 +51,9 @@ REDIS_URL=redis://localhost REDIS_MIN_CONNECTIONS=0 REDIS_MAX_CONNECTIONS=10000 +KAFKA_BOOTSTRAP_SERVERS=localhost:19092 +KAFKA_CLIENT_ID=labrinth + # Must bind to broadcast, not localhost, because some # Docker services (Gotenberg) must be able to reach the backend # from a different network interface diff --git a/apps/labrinth/.sqlx/query-0ffebf149d50149d0c128478e6267a2296efc16763cdbd5d8c295a1a7ca08021.json b/apps/labrinth/.sqlx/query-0ffebf149d50149d0c128478e6267a2296efc16763cdbd5d8c295a1a7ca08021.json new file mode 100644 index 000000000..c8a82d60e --- /dev/null +++ b/apps/labrinth/.sqlx/query-0ffebf149d50149d0c128478e6267a2296efc16763cdbd5d8c295a1a7ca08021.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO users (id, username, email, role)\n VALUES\n (1000, 'userland', 'userland@modrinth.com', 'developer'),\n (1001, 'useful', 'useful@modrinth.com', 'developer'),\n (1002, 'Useless', 'useless@modrinth.com', 'developer')\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "0ffebf149d50149d0c128478e6267a2296efc16763cdbd5d8c295a1a7ca08021" +} diff --git a/apps/labrinth/.sqlx/query-329c009635ad2a4c214da2f4651b8c14fe329deac983bc2f43ba74e066f9cf8d.json b/apps/labrinth/.sqlx/query-329c009635ad2a4c214da2f4651b8c14fe329deac983bc2f43ba74e066f9cf8d.json new file mode 100644 index 000000000..2731d395f --- /dev/null +++ b/apps/labrinth/.sqlx/query-329c009635ad2a4c214da2f4651b8c14fe329deac983bc2f43ba74e066f9cf8d.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE mods SET status = 'processing' WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [] + }, + "hash": "329c009635ad2a4c214da2f4651b8c14fe329deac983bc2f43ba74e066f9cf8d" +} diff --git a/apps/labrinth/.sqlx/query-38ace0abad724d1111b79a72c7c8518f1272690ce7766dbdc41ec1e98f7f1fec.json b/apps/labrinth/.sqlx/query-38ace0abad724d1111b79a72c7c8518f1272690ce7766dbdc41ec1e98f7f1fec.json new file mode 100644 index 000000000..815e7795e --- /dev/null +++ b/apps/labrinth/.sqlx/query-38ace0abad724d1111b79a72c7c8518f1272690ce7766dbdc41ec1e98f7f1fec.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO users (id, username, email, role)\n VALUES (2100, 'prefix_under_score', 'prefix_under_score@modrinth.com', 'developer')\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "38ace0abad724d1111b79a72c7c8518f1272690ce7766dbdc41ec1e98f7f1fec" +} diff --git a/apps/labrinth/.sqlx/query-5cb65218a2a4a7343130d838f8a78df08d25844f1019d539712b0cb686e4ae81.json b/apps/labrinth/.sqlx/query-5cb65218a2a4a7343130d838f8a78df08d25844f1019d539712b0cb686e4ae81.json new file mode 100644 index 000000000..55ba90262 --- /dev/null +++ b/apps/labrinth/.sqlx/query-5cb65218a2a4a7343130d838f8a78df08d25844f1019d539712b0cb686e4ae81.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO users (id, username, email, role)\n VALUES ($1, $2, $3, 'developer')\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "5cb65218a2a4a7343130d838f8a78df08d25844f1019d539712b0cb686e4ae81" +} diff --git a/apps/labrinth/.sqlx/query-66868d0b35229abffa3f668eaef19dc0cdef502935dd5939a2f3ab89b0766ad2.json b/apps/labrinth/.sqlx/query-66868d0b35229abffa3f668eaef19dc0cdef502935dd5939a2f3ab89b0766ad2.json new file mode 100644 index 000000000..5d509fdf0 --- /dev/null +++ b/apps/labrinth/.sqlx/query-66868d0b35229abffa3f668eaef19dc0cdef502935dd5939a2f3ab89b0766ad2.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "CREATE SCHEMA public;", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "66868d0b35229abffa3f668eaef19dc0cdef502935dd5939a2f3ab89b0766ad2" +} diff --git a/apps/labrinth/.sqlx/query-881de045a10c1debe041ec558fd92fdfbf8b8ffa1b21d3d4a271ab6f9c8adb78.json b/apps/labrinth/.sqlx/query-881de045a10c1debe041ec558fd92fdfbf8b8ffa1b21d3d4a271ab6f9c8adb78.json new file mode 100644 index 000000000..ee7ce35fb --- /dev/null +++ b/apps/labrinth/.sqlx/query-881de045a10c1debe041ec558fd92fdfbf8b8ffa1b21d3d4a271ab6f9c8adb78.json @@ -0,0 +1,95 @@ +{ + "db_name": "PostgreSQL", + "query": "\n\t\tSELECT m.id id, m.name name, m.summary summary, m.downloads downloads, m.follows follows,\n\t\tm.icon_url icon_url, m.updated updated, m.approved approved, m.published, m.license license, m.slug slug, m.color,\n\t\tm.components AS \"components: sqlx::types::Json\"\n\t\tFROM mods m\n\t\tWHERE m.status = ANY($1) AND m.id = ANY($2)\n\t\tGROUP BY m.id\n\t\tORDER BY m.id ASC;\n\t\t", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "name", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "summary", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "downloads", + "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "follows", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "icon_url", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "updated", + "type_info": "Timestamptz" + }, + { + "ordinal": 7, + "name": "approved", + "type_info": "Timestamptz" + }, + { + "ordinal": 8, + "name": "published", + "type_info": "Timestamptz" + }, + { + "ordinal": 9, + "name": "license", + "type_info": "Varchar" + }, + { + "ordinal": 10, + "name": "slug", + "type_info": "Varchar" + }, + { + "ordinal": 11, + "name": "color", + "type_info": "Int4" + }, + { + "ordinal": 12, + "name": "components: sqlx::types::Json", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Int8Array" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + true, + false, + true, + false, + false, + true, + true, + false + ] + }, + "hash": "881de045a10c1debe041ec558fd92fdfbf8b8ffa1b21d3d4a271ab6f9c8adb78" +} diff --git a/apps/labrinth/.sqlx/query-8ca7f46618bd72cfd5fe1f246d41a31da1885cc22bb61be6853950ba52be9aa7.json b/apps/labrinth/.sqlx/query-8ca7f46618bd72cfd5fe1f246d41a31da1885cc22bb61be6853950ba52be9aa7.json new file mode 100644 index 000000000..444f57bf3 --- /dev/null +++ b/apps/labrinth/.sqlx/query-8ca7f46618bd72cfd5fe1f246d41a31da1885cc22bb61be6853950ba52be9aa7.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE moderation_locks SET locked_at = NOW() - ($1::bigint * INTERVAL '1 minute') - INTERVAL '1 second' WHERE project_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "8ca7f46618bd72cfd5fe1f246d41a31da1885cc22bb61be6853950ba52be9aa7" +} diff --git a/apps/labrinth/.sqlx/query-a34294b38639cb96d0c2490773de6c5e7c4bdcbeb85c84d0b23550ddb83c431f.json b/apps/labrinth/.sqlx/query-a34294b38639cb96d0c2490773de6c5e7c4bdcbeb85c84d0b23550ddb83c431f.json new file mode 100644 index 000000000..a1028223d --- /dev/null +++ b/apps/labrinth/.sqlx/query-a34294b38639cb96d0c2490773de6c5e7c4bdcbeb85c84d0b23550ddb83c431f.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO moderation_locks (project_id, moderator_id, locked_at)\n VALUES ($1, $2, NOW())\n ON CONFLICT (project_id) DO UPDATE SET\n moderator_id = EXCLUDED.moderator_id,\n locked_at = EXCLUDED.locked_at\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "a34294b38639cb96d0c2490773de6c5e7c4bdcbeb85c84d0b23550ddb83c431f" +} diff --git a/apps/labrinth/.sqlx/query-b7a3ebd72c8685c590298c154fb3da625f34589540efbb08201c0921f96d510b.json b/apps/labrinth/.sqlx/query-b7a3ebd72c8685c590298c154fb3da625f34589540efbb08201c0921f96d510b.json new file mode 100644 index 000000000..2769fe4fe --- /dev/null +++ b/apps/labrinth/.sqlx/query-b7a3ebd72c8685c590298c154fb3da625f34589540efbb08201c0921f96d510b.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE versions v\n SET downloads = v.downloads + x.amount\n FROM unnest($1::BIGINT[], $2::int[]) AS x(id, amount)\n WHERE v.id = x.id\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8Array", + "Int4Array" + ] + }, + "nullable": [] + }, + "hash": "b7a3ebd72c8685c590298c154fb3da625f34589540efbb08201c0921f96d510b" +} diff --git a/apps/labrinth/.sqlx/query-c65295f61b18f7be4fdf30b3e2c2c36b9db3770508be4acbb6b441ed8006d9c7.json b/apps/labrinth/.sqlx/query-c65295f61b18f7be4fdf30b3e2c2c36b9db3770508be4acbb6b441ed8006d9c7.json new file mode 100644 index 000000000..f245ea1c8 --- /dev/null +++ b/apps/labrinth/.sqlx/query-c65295f61b18f7be4fdf30b3e2c2c36b9db3770508be4acbb6b441ed8006d9c7.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "DROP SCHEMA public CASCADE;", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "c65295f61b18f7be4fdf30b3e2c2c36b9db3770508be4acbb6b441ed8006d9c7" +} diff --git a/apps/labrinth/.sqlx/query-e6287177bd5725f587327a9bd5bcce60358fe9401e0477c39c8a910412c3865c.json b/apps/labrinth/.sqlx/query-e6287177bd5725f587327a9bd5bcce60358fe9401e0477c39c8a910412c3865c.json new file mode 100644 index 000000000..aa5014a01 --- /dev/null +++ b/apps/labrinth/.sqlx/query-e6287177bd5725f587327a9bd5bcce60358fe9401e0477c39c8a910412c3865c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE mods m\n SET downloads = m.downloads + x.amount\n FROM unnest($1::BIGINT[], $2::int[]) AS x(id, amount)\n WHERE m.id = x.id\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8Array", + "Int4Array" + ] + }, + "nullable": [] + }, + "hash": "e6287177bd5725f587327a9bd5bcce60358fe9401e0477c39c8a910412c3865c" +} diff --git a/apps/labrinth/.sqlx/query-fabba52b34e14196b6305671c060d934b9afc6dccc5fc77454fbe45365bdf429.json b/apps/labrinth/.sqlx/query-fabba52b34e14196b6305671c060d934b9afc6dccc5fc77454fbe45365bdf429.json new file mode 100644 index 000000000..006563f45 --- /dev/null +++ b/apps/labrinth/.sqlx/query-fabba52b34e14196b6305671c060d934b9afc6dccc5fc77454fbe45365bdf429.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT pg_advisory_unlock(1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "pg_advisory_unlock", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "fabba52b34e14196b6305671c060d934b9afc6dccc5fc77454fbe45365bdf429" +} diff --git a/apps/labrinth/Cargo.toml b/apps/labrinth/Cargo.toml index 9d99dc500..f8cf06386 100644 --- a/apps/labrinth/Cargo.toml +++ b/apps/labrinth/Cargo.toml @@ -83,6 +83,7 @@ prometheus = { workspace = true } quick-xml = { workspace = true } rand = { workspace = true } rand_chacha = { workspace = true } +rdkafka = { workspace = true } redis = { workspace = true, features = ["ahash", "r2d2", "tokio-comp"] } regex = { workspace = true } reqwest = { workspace = true, features = [ @@ -128,7 +129,7 @@ urlencoding = { workspace = true } utoipa = { workspace = true, features = ["url"] } utoipa-actix-web = { workspace = true } utoipa-scalar = { workspace = true, features = ["actix-web"] } -uuid = { workspace = true, features = ["fast-rng", "serde", "v4"] } +uuid = { workspace = true, features = ["fast-rng", "serde", "v4", "v7"] } validator = { workspace = true, features = ["derive"] } webp = { workspace = true } woothee = { workspace = true } diff --git a/apps/labrinth/src/background_task.rs b/apps/labrinth/src/background_task.rs index 6cbc76f8e..319eb721d 100644 --- a/apps/labrinth/src/background_task.rs +++ b/apps/labrinth/src/background_task.rs @@ -29,6 +29,7 @@ pub enum BackgroundTask { SyncPayoutStatuses, IndexBilling, IndexSubscriptions, + IncrementalIndexSearch, Migrations, Mail, /// Queries server project analytics (e.g. number of verified plays in last @@ -49,6 +50,7 @@ impl BackgroundTask { ro_pool: PgPool, redis_pool: RedisPool, search_backend: web::Data, + kafka_client: web::Data, clickhouse: clickhouse::Client, stripe_client: stripe::Client, anrok_client: anrok::Client, @@ -88,12 +90,26 @@ impl BackgroundTask { .await; Ok(()) } + IncrementalIndexSearch => { + crate::search::incremental::consume::run( + ro_pool, + redis_pool, + search_backend, + ) + .await + } Mail => run_email(email_queue).await, CacheAnalytics => { cache_analytics(&pool, &redis_pool, &clickhouse).await } PingMinecraftJavaServers => { - ping_minecraft_java_servers(pool, redis_pool, clickhouse).await + ping_minecraft_java_servers( + pool, + redis_pool, + clickhouse, + kafka_client, + ) + .await } DiscordRoleEmailCampaign => { discord_role_email_campaign(pool, redis_pool).await @@ -322,17 +338,27 @@ pub async fn ping_minecraft_java_servers( pool: PgPool, redis_pool: RedisPool, clickhouse: clickhouse::Client, + kafka_client: web::Data, ) -> eyre::Result<()> { info!("Started pinging Minecraft Java servers"); + let incremental_search_queue = + crate::search::incremental::IncrementalSearchQueue::new(kafka_client); let server_ping_queue = crate::queue::server_ping::ServerPingQueue::new( - pool, redis_pool, clickhouse, + pool, + redis_pool, + clickhouse, + incremental_search_queue.clone(), ); server_ping_queue .ping_minecraft_java_servers() .await .wrap_err("failed to ping Minecraft Java servers")?; + incremental_search_queue + .drain() + .await + .wrap_err("failed to drain incremental search queue")?; info!("Successfully pinged Minecraft Java servers"); info!("Done pinging Minecraft Java servers"); diff --git a/apps/labrinth/src/database/models/moderation_lock_item.rs b/apps/labrinth/src/database/models/moderation_lock_item.rs index 46413698a..6a458f03d 100644 --- a/apps/labrinth/src/database/models/moderation_lock_item.rs +++ b/apps/labrinth/src/database/models/moderation_lock_item.rs @@ -101,7 +101,7 @@ impl DBModerationLock { moderator_id: DBUserId, pool: &PgPool, ) -> Result<(), sqlx::Error> { - sqlx::query( + sqlx::query!( r#" INSERT INTO moderation_locks (project_id, moderator_id, locked_at) VALUES ($1, $2, NOW()) @@ -109,9 +109,9 @@ impl DBModerationLock { moderator_id = EXCLUDED.moderator_id, locked_at = EXCLUDED.locked_at "#, + project_id as DBProjectId, + moderator_id as DBUserId, ) - .bind(project_id) - .bind(moderator_id) .execute(pool) .await?; diff --git a/apps/labrinth/src/env.rs b/apps/labrinth/src/env.rs index de5549764..0b88cdb49 100644 --- a/apps/labrinth/src/env.rs +++ b/apps/labrinth/src/env.rs @@ -130,6 +130,8 @@ vars! { RATE_LIMIT_IGNORE_KEY: String = ""; DATABASE_URL: String = "postgresql://labrinth:labrinth@localhost/labrinth"; REDIS_URL: String = "redis://localhost"; + KAFKA_BOOTSTRAP_SERVERS: StringCsv = StringCsv(vec!["localhost:19092".into()]); + KAFKA_CLIENT_ID: String = "labrinth"; BIND_ADDR: String = ""; SELF_ADDR: String = ""; diff --git a/apps/labrinth/src/lib.rs b/apps/labrinth/src/lib.rs index a97ec86bd..d6da0785b 100644 --- a/apps/labrinth/src/lib.rs +++ b/apps/labrinth/src/lib.rs @@ -61,7 +61,7 @@ pub struct LabrinthConfig { pub file_host: Arc, pub scheduler: Arc, pub ip_salt: Pepper, - pub search_backend: web::Data, + pub search_state: web::Data, pub session_queue: web::Data, pub payouts_queue: web::Data, pub analytics_queue: Arc, @@ -75,6 +75,7 @@ pub struct LabrinthConfig { pub gotenberg_client: GotenbergClient, pub http_client: web::Data, pub tiltify_client: web::Data, + pub kafka_client: web::Data, } #[allow(clippy::too_many_arguments)] @@ -89,6 +90,7 @@ pub fn app_setup( anrok_client: anrok::Client, email_queue: EmailQueue, gotenberg_client: GotenbergClient, + kafka_client: web::Data, enable_background_tasks: bool, ) -> LabrinthConfig { info!("Starting labrinth on {}", &ENV.BIND_ADDR); @@ -112,6 +114,18 @@ pub fn app_setup( let http_client = web::Data::new(HttpClient::new()); let tiltify_client = web::Data::new(TiltifyClient::new(http_client.get_ref().clone())); + let search_state = web::Data::new(search::SearchState { + backend: search_backend.clone().into_inner(), + queue: search::incremental::IncrementalSearchQueue::new( + kafka_client.clone(), + ), + }); + { + let incremental_search_queue = search_state.queue.clone(); + actix_rt::spawn(async move { + incremental_search_queue.run().await; + }); + } { let pool_ref = pool.clone(); let http_ref = http_client.clone(); @@ -305,7 +319,7 @@ pub fn app_setup( file_host, scheduler: Arc::new(scheduler), ip_salt, - search_backend, + search_state, session_queue, payouts_queue: web::Data::new(PayoutsQueue::new()), analytics_queue, @@ -317,6 +331,7 @@ pub fn app_setup( gotenberg_client, http_client, tiltify_client, + kafka_client, archon_client: web::Data::new( ArchonClient::from_env() .expect("ARCHON_URL and PYRO_API_KEY must be set"), @@ -345,7 +360,9 @@ pub fn app_config( .app_data(web::Data::new(labrinth_config.pool.clone())) .app_data(web::Data::new(labrinth_config.ro_pool.clone())) .app_data(web::Data::new(labrinth_config.file_host.clone())) - .app_data(labrinth_config.search_backend.clone()) + .app_data(web::Data::from( + labrinth_config.search_state.backend.clone(), + )) .app_data(web::Data::new(labrinth_config.gotenberg_client.clone())) .app_data(labrinth_config.http_client.clone()) .app_data(labrinth_config.tiltify_client.clone()) @@ -361,6 +378,8 @@ pub fn app_config( .app_data(web::Data::new(labrinth_config.stripe_client.clone())) .app_data(web::Data::new(labrinth_config.anrok_client.clone())) .app_data(labrinth_config.rate_limiter.clone()) + .app_data(labrinth_config.kafka_client.clone()) + .app_data(labrinth_config.search_state.clone()) .configure(routes::v3::config) .configure(routes::internal::config) .configure(routes::root_config) diff --git a/apps/labrinth/src/main.rs b/apps/labrinth/src/main.rs index f24a2fb79..adda6b7fa 100644 --- a/apps/labrinth/src/main.rs +++ b/apps/labrinth/src/main.rs @@ -166,6 +166,10 @@ async fn app() -> std::io::Result<()> { .expect("Failed to create Gotenberg client"); let muralpay = labrinth::queue::payouts::create_muralpay_client() .expect("Failed to create MuralPay client"); + let kafka_client = actix_web::web::Data::new( + labrinth::util::kafka::KafkaClientState::new() + .expect("Kafka connection failed"), + ); if let Some(task) = args.run_background_task { info!("Running task {task:?} and exiting"); @@ -174,6 +178,7 @@ async fn app() -> std::io::Result<()> { ro_pool.into_inner(), redis_pool, search_backend, + kafka_client, clickhouse, stripe_client, anrok_client.clone(), @@ -216,6 +221,7 @@ async fn app() -> std::io::Result<()> { anrok_client.clone(), email_queue, gotenberg_client, + kafka_client, !args.no_background_tasks, ); diff --git a/apps/labrinth/src/queue/analytics/mod.rs b/apps/labrinth/src/queue/analytics/mod.rs index fc271fcf4..6d4716c69 100644 --- a/apps/labrinth/src/queue/analytics/mod.rs +++ b/apps/labrinth/src/queue/analytics/mod.rs @@ -339,29 +339,29 @@ impl AnalyticsQueue { downloads.write(&download).await?; } - sqlx::query( + sqlx::query!( " UPDATE versions v SET downloads = v.downloads + x.amount FROM unnest($1::BIGINT[], $2::int[]) AS x(id, amount) WHERE v.id = x.id ", + &version_downloads.keys().copied().collect::>(), + &version_downloads.values().copied().collect::>(), ) - .bind(version_downloads.keys().copied().collect::>()) - .bind(version_downloads.values().copied().collect::>()) .execute(&mut transaction) .await?; - sqlx::query( + sqlx::query!( " UPDATE mods m SET downloads = m.downloads + x.amount FROM unnest($1::BIGINT[], $2::int[]) AS x(id, amount) WHERE m.id = x.id ", + &project_downloads.keys().copied().collect::>(), + &project_downloads.values().copied().collect::>(), ) - .bind(project_downloads.keys().copied().collect::>()) - .bind(project_downloads.values().copied().collect::>()) .execute(&mut transaction) .await?; diff --git a/apps/labrinth/src/queue/server_ping.rs b/apps/labrinth/src/queue/server_ping.rs index 2698aad5e..5b6305961 100644 --- a/apps/labrinth/src/queue/server_ping.rs +++ b/apps/labrinth/src/queue/server_ping.rs @@ -5,10 +5,12 @@ use crate::env::ENV; use crate::models::exp; use crate::models::ids::ProjectId; use crate::models::projects::ProjectStatus; +use crate::search::incremental::IncrementalSearchQueue; use crate::{database::PgPool, util::error::Context}; use async_minecraft_ping::ServerDescription; use chrono::{TimeDelta, Utc}; use clickhouse::{Client, Row}; +use futures::future::join; use serde::Serialize; use sqlx::types::Json; use std::sync::Arc; @@ -21,6 +23,7 @@ pub struct ServerPingQueue { pub db: PgPool, pub redis: RedisPool, pub clickhouse: Client, + pub incremental_search_queue: IncrementalSearchQueue, } pub const REDIS_NAMESPACE: &str = "minecraft_java_server_ping"; @@ -28,11 +31,17 @@ pub const REDIS_FAILURE_NAMESPACE: &str = "minecraft_java_server_ping_failures"; pub const CLICKHOUSE_TABLE: &str = "minecraft_java_server_pings"; impl ServerPingQueue { - pub fn new(db: PgPool, redis: RedisPool, clickhouse: Client) -> Self { + pub fn new( + db: PgPool, + redis: RedisPool, + clickhouse: Client, + incremental_search_queue: IncrementalSearchQueue, + ) -> Self { Self { db, redis, clickhouse, + incremental_search_queue, } } @@ -166,17 +175,22 @@ impl ServerPingQueue { } if updated_project { - DBProject::clear_cache( + let clear_cache = DBProject::clear_cache( (*project_id).into(), None, None, &self.redis, - ) - .await - .inspect_err(|err| { - warn!("failed to clear project cache: {err:#}") - }) - .ok(); + ); + let queue_search = + self.incremental_search_queue.push(*project_id); + + let (clear_cache_result, _) = + join(clear_cache, queue_search).await; + clear_cache_result + .inspect_err(|err| { + warn!("failed to clear project cache: {err:#}") + }) + .ok(); } } diff --git a/apps/labrinth/src/routes/internal/admin.rs b/apps/labrinth/src/routes/internal/admin.rs index f0d5c2617..00eb79356 100644 --- a/apps/labrinth/src/routes/internal/admin.rs +++ b/apps/labrinth/src/routes/internal/admin.rs @@ -8,6 +8,7 @@ use crate::queue::analytics::AnalyticsQueue; use crate::queue::session::AuthQueue; use crate::routes::ApiError; use crate::search::SearchBackend; +use crate::search::incremental::consume::reindex_project; use crate::util::date::get_current_tenths_of_ms; use crate::util::error::Context; use crate::util::guards::admin_key_guard; @@ -26,7 +27,8 @@ pub fn config(cfg: &mut utoipa_actix_web::service_config::ServiceConfig) { cfg.service( utoipa_actix_web::scope("/admin") .service(count_download) - .service(force_reindex), + .service(force_reindex) + .service(force_reindex_project), ); } @@ -327,3 +329,33 @@ pub async fn force_reindex( .wrap_internal_err("failed to index projects")?; Ok(HttpResponse::NoContent().finish()) } + +#[utoipa::path( + post, + operation_id = "forceReindexProject", + responses( + (status = 204, description = "Project search documents rebuilt successfully"), + (status = 401, description = "Unauthorized") + ) +)] +#[post("/_force_reindex/{project_id}", guard = "admin_key_guard")] +pub async fn force_reindex_project( + path: web::Path<(ProjectId,)>, + pool: web::Data, + redis: web::Data, + search_backend: web::Data, +) -> Result { + let (project_id,) = path.into_inner(); + reindex_project( + pool.as_ref(), + redis.as_ref(), + search_backend.as_ref(), + project_id, + ) + .await + .wrap_internal_err_with(|| { + eyre!("failed to reindex project `{project_id}`") + })?; + + Ok(HttpResponse::NoContent().finish()) +} diff --git a/apps/labrinth/src/routes/internal/mod.rs b/apps/labrinth/src/routes/internal/mod.rs index 5106bc2d0..afc19bfc8 100644 --- a/apps/labrinth/src/routes/internal/mod.rs +++ b/apps/labrinth/src/routes/internal/mod.rs @@ -29,7 +29,8 @@ pub fn config(cfg: &mut actix_web::web::ServiceConfig) { cfg.service( actix_web::web::scope("/admin") .service(admin::count_download) - .service(admin::force_reindex), + .service(admin::force_reindex) + .service(admin::force_reindex_project), ); cfg.service( actix_web::web::scope("/session") diff --git a/apps/labrinth/src/routes/internal/moderation/tech_review.rs b/apps/labrinth/src/routes/internal/moderation/tech_review.rs index d201f081e..05fae0e88 100644 --- a/apps/labrinth/src/routes/internal/moderation/tech_review.rs +++ b/apps/labrinth/src/routes/internal/moderation/tech_review.rs @@ -32,6 +32,7 @@ use crate::{ }, queue::session::AuthQueue, routes::{ApiError, internal::moderation::Ownership}, + search::SearchState, util::error::Context, }; use eyre::eyre; @@ -978,6 +979,7 @@ async fn submit_report( pool: web::Data, redis: web::Data, session_queue: web::Data, + search_state: web::Data, web::Json(submit_report): web::Json, path: web::Path<(ProjectId,)>, ) -> Result<(), ApiError> { @@ -1132,16 +1134,23 @@ async fn submit_report( .insert(&mut txn) .await .wrap_internal_err("failed to add tech review message")?; - - DBProject::clear_cache(project_id, None, None, &redis) - .await - .wrap_internal_err("failed to clear project cache")?; } txn.commit() .await .wrap_internal_err("failed to commit transaction")?; + if verdict == DelphiVerdict::Unsafe { + crate::routes::v3::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, + project_id, + None, + None, + ) + .await?; + } + Ok(()) } diff --git a/apps/labrinth/src/routes/v2/project_creation.rs b/apps/labrinth/src/routes/v2/project_creation.rs index a6b7a6292..ded3fd4f0 100644 --- a/apps/labrinth/src/routes/v2/project_creation.rs +++ b/apps/labrinth/src/routes/v2/project_creation.rs @@ -12,6 +12,7 @@ use crate::queue::session::AuthQueue; use crate::routes::v3::project_creation::default_project_type; use crate::routes::v3::project_creation::{CreateError, NewGalleryItem}; use crate::routes::{v2_reroute, v3}; +use crate::search::SearchState; use crate::util::http::HttpClient; use actix_multipart::Multipart; use actix_web::web::Data; @@ -161,6 +162,7 @@ pub async fn project_create( file_host: Data>, session_queue: Data, http: Data, + search_state: Data, ) -> Result { // Convert V2 multipart payload to V3 multipart payload let payload = v2_reroute::alter_actix_multipart( @@ -279,6 +281,7 @@ pub async fn project_create( file_host, session_queue, http, + search_state, ) .await?; diff --git a/apps/labrinth/src/routes/v2/projects.rs b/apps/labrinth/src/routes/v2/projects.rs index ab24386b8..77deda18d 100644 --- a/apps/labrinth/src/routes/v2/projects.rs +++ b/apps/labrinth/src/routes/v2/projects.rs @@ -14,7 +14,7 @@ use crate::queue::moderation::AutomatedModerationQueue; use crate::queue::session::AuthQueue; use crate::routes::v3::projects::ProjectIds; use crate::routes::{ApiError, v2_reroute, v3}; -use crate::search::{SearchBackend, SearchRequest}; +use crate::search::{SearchBackend, SearchRequest, SearchState}; use actix_web::{HttpRequest, HttpResponse, delete, get, patch, post, web}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -532,11 +532,11 @@ pub async fn project_edit( req: HttpRequest, info: web::Path<(String,)>, pool: web::Data, - search_backend: web::Data, new_project: web::Json, redis: web::Data, session_queue: web::Data, moderation_queue: web::Data, + search_state: web::Data, ) -> Result { let v2_new_project = new_project.into_inner(); let client_side = v2_new_project.client_side; @@ -646,11 +646,11 @@ pub async fn project_edit( req.clone(), info, pool.clone(), - search_backend, web::Json(new_project), redis.clone(), session_queue.clone(), moderation_queue, + search_state.clone(), ) .await .or_else(v2_reroute::flatten_404_error)?; @@ -694,6 +694,7 @@ pub async fn project_edit( ..Default::default() }, session_queue.clone(), + search_state.clone(), ) .await?; } @@ -794,6 +795,7 @@ pub async fn projects_edit( bulk_edit_project: web::Json, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { let bulk_edit_project = bulk_edit_project.into_inner(); @@ -878,6 +880,7 @@ pub async fn projects_edit( }), redis, session_queue, + search_state, ) .await .or_else(v2_reroute::flatten_404_error) @@ -927,6 +930,7 @@ pub async fn project_icon_edit( file_host: web::Data>, payload: web::Payload, session_queue: web::Data, + search_state: web::Data, ) -> Result { // Returns NoContent, so no need to convert v3::projects::project_icon_edit_internal( @@ -938,6 +942,7 @@ pub async fn project_icon_edit( file_host, payload, session_queue, + search_state, ) .await .or_else(v2_reroute::flatten_404_error) @@ -966,6 +971,7 @@ pub async fn delete_project_icon( redis: web::Data, file_host: web::Data>, session_queue: web::Data, + search_state: web::Data, ) -> Result { // Returns NoContent, so no need to convert v3::projects::delete_project_icon_internal( @@ -975,6 +981,7 @@ pub async fn delete_project_icon( redis, file_host, session_queue, + search_state, ) .await .or_else(v2_reroute::flatten_404_error) @@ -1058,6 +1065,7 @@ pub async fn add_gallery_item( file_host: web::Data>, payload: web::Payload, session_queue: web::Data, + search_state: web::Data, ) -> Result { // Returns NoContent, so no need to convert v3::projects::add_gallery_item_internal( @@ -1075,6 +1083,7 @@ pub async fn add_gallery_item( file_host, payload, session_queue, + search_state, ) .await .or_else(v2_reroute::flatten_404_error) @@ -1150,6 +1159,7 @@ pub async fn edit_gallery_item( pool: web::Data, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { // Returns NoContent, so no need to convert v3::projects::edit_gallery_item_internal( @@ -1164,6 +1174,7 @@ pub async fn edit_gallery_item( pool, redis, session_queue, + search_state, ) .await .or_else(v2_reroute::flatten_404_error) @@ -1200,6 +1211,7 @@ pub async fn delete_gallery_item( redis: web::Data, file_host: web::Data>, session_queue: web::Data, + search_state: web::Data, ) -> Result { // Returns NoContent, so no need to convert v3::projects::delete_gallery_item_internal( @@ -1209,6 +1221,7 @@ pub async fn delete_gallery_item( redis, file_host, session_queue, + search_state, ) .await .or_else(v2_reroute::flatten_404_error) @@ -1235,8 +1248,8 @@ pub async fn project_delete( info: web::Path<(String,)>, pool: web::Data, redis: web::Data, - search_backend: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { // Returns NoContent, so no need to convert v3::projects::project_delete_internal( @@ -1244,8 +1257,8 @@ pub async fn project_delete( info, pool, redis, - search_backend, session_queue, + search_state, ) .await .map(|()| HttpResponse::NoContent().body("")) diff --git a/apps/labrinth/src/routes/v2/version_creation.rs b/apps/labrinth/src/routes/v2/version_creation.rs index c60247e10..1b8be2733 100644 --- a/apps/labrinth/src/routes/v2/version_creation.rs +++ b/apps/labrinth/src/routes/v2/version_creation.rs @@ -13,6 +13,7 @@ use crate::queue::session::AuthQueue; use crate::routes::v3::project_creation::CreateError; use crate::routes::v3::version_creation; use crate::routes::{v2_reroute, v3}; +use crate::search::SearchState; use crate::util::http::HttpClient; use actix_multipart::Multipart; use actix_web::http::header::ContentDisposition; @@ -104,6 +105,7 @@ pub async fn version_create( session_queue: Data, moderation_queue: Data, http: Data, + search_state: Data, ) -> Result { let payload = v2_reroute::alter_actix_multipart( payload, @@ -263,6 +265,7 @@ pub async fn version_create( session_queue, moderation_queue, http, + search_state, ) .await?; @@ -335,6 +338,7 @@ pub async fn upload_file_to_version( file_host: Data>, session_queue: web::Data, http: web::Data, + search_state: Data, ) -> Result { // Returns NoContent, so no need to convert to V2 let response = v3::version_creation::upload_file_to_version( @@ -346,6 +350,7 @@ pub async fn upload_file_to_version( file_host, session_queue, http, + search_state, ) .await?; Ok(response) diff --git a/apps/labrinth/src/routes/v2/versions.rs b/apps/labrinth/src/routes/v2/versions.rs index 438c33175..6184ea213 100644 --- a/apps/labrinth/src/routes/v2/versions.rs +++ b/apps/labrinth/src/routes/v2/versions.rs @@ -11,7 +11,7 @@ use crate::models::projects::{ use crate::models::v2::projects::LegacyVersion; use crate::queue::session::AuthQueue; use crate::routes::{v2_reroute, v3}; -use crate::search::SearchBackend; +use crate::search::{SearchBackend, SearchState}; use actix_web::{HttpRequest, HttpResponse, delete, get, patch, web}; use serde::{Deserialize, Serialize}; use validator::Validate; @@ -367,6 +367,7 @@ pub async fn version_edit( redis: web::Data, new_version: web::Json, session_queue: web::Data, + search_state: web::Data, ) -> Result { let new_version = new_version.into_inner(); @@ -445,6 +446,7 @@ pub async fn version_edit( redis, web::Json(serde_json::to_value(new_version)?), session_queue, + search_state, ) .await .or_else(v2_reroute::flatten_404_error)?; @@ -477,6 +479,7 @@ pub async fn version_delete( redis: web::Data, session_queue: web::Data, search_backend: web::Data, + search_state: web::Data, ) -> Result { // Returns NoContent, so we don't need to convert the response v3::versions::version_delete( @@ -486,6 +489,7 @@ pub async fn version_delete( redis, session_queue, search_backend, + search_state, ) .await .or_else(v2_reroute::flatten_404_error) diff --git a/apps/labrinth/src/routes/v3/organizations.rs b/apps/labrinth/src/routes/v3/organizations.rs index aba34ca27..cd6a119da 100644 --- a/apps/labrinth/src/routes/v3/organizations.rs +++ b/apps/labrinth/src/routes/v3/organizations.rs @@ -17,6 +17,7 @@ use crate::models::teams::{OrganizationPermissions, ProjectPermissions}; use crate::models::v3::user_limits::UserLimits; use crate::queue::session::AuthQueue; use crate::routes::v3::project_creation::CreateError; +use crate::search::SearchState; use crate::util::img::delete_old_images; use crate::util::routes::read_limited_from_payload; use crate::util::validate::validation_errors_to_string; @@ -664,6 +665,7 @@ pub async fn organization_delete( pool: web::Data, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -800,8 +802,12 @@ pub async fn organization_delete( } for project_id in organization_project_ids { - database::models::DBProject::clear_cache( - project_id, None, None, &redis, + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, + project_id, + None, + None, ) .await?; } @@ -829,6 +835,7 @@ pub async fn organization_projects_add( pool: web::Data, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { let info = info.into_inner().0; let current_user = get_user_from_headers( @@ -962,11 +969,12 @@ pub async fn organization_projects_add( &redis, ) .await?; - database::models::DBProject::clear_cache( + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; } else { @@ -992,6 +1000,7 @@ pub async fn organization_projects_remove( data: web::Json, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { let (organization_id, project_id) = info.into_inner(); let current_user = get_user_from_headers( @@ -1150,11 +1159,12 @@ pub async fn organization_projects_remove( &redis, ) .await?; - database::models::DBProject::clear_cache( + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; } else { diff --git a/apps/labrinth/src/routes/v3/project_creation.rs b/apps/labrinth/src/routes/v3/project_creation.rs index 42f1e4624..b216d9e22 100644 --- a/apps/labrinth/src/routes/v3/project_creation.rs +++ b/apps/labrinth/src/routes/v3/project_creation.rs @@ -22,6 +22,7 @@ use crate::models::teams::{OrganizationPermissions, ProjectPermissions}; use crate::models::threads::ThreadType; use crate::models::v3::user_limits::UserLimits; use crate::queue::session::AuthQueue; +use crate::search::SearchState; use crate::util::guards::admin_key_guard; use crate::util::http::HttpClient; use crate::util::img::upload_image_optimized; @@ -293,6 +294,7 @@ pub async fn project_create( file_host: Data>, session_queue: Data, http: Data, + search_state: Data, ) -> Result { project_create_internal( req, @@ -302,6 +304,7 @@ pub async fn project_create( file_host, session_queue, http, + search_state, ) .await } @@ -314,6 +317,7 @@ pub async fn project_create_internal( file_host: Data>, session_queue: Data, http: Data, + search_state: Data, ) -> Result { let mut transaction = client.begin().await?; let mut uploaded_files = Vec::new(); @@ -345,6 +349,14 @@ pub async fn project_create_internal( } } else { transaction.commit().await?; + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, + project_id.into(), + None, + None, + ) + .await?; } result @@ -363,6 +375,7 @@ pub async fn project_create_with_id( file_host: Data>, session_queue: Data, http: Data, + search_state: Data, path: web::Path<(ProjectId,)>, ) -> Result { let mut transaction = client.begin().await?; @@ -394,6 +407,14 @@ pub async fn project_create_with_id( } } else { transaction.commit().await?; + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, + project_id.into(), + None, + None, + ) + .await?; } result diff --git a/apps/labrinth/src/routes/v3/projects.rs b/apps/labrinth/src/routes/v3/projects.rs index a40f93305..88e695a4b 100644 --- a/apps/labrinth/src/routes/v3/projects.rs +++ b/apps/labrinth/src/routes/v3/projects.rs @@ -29,7 +29,9 @@ use crate::queue::moderation::AutomatedModerationQueue; use crate::queue::session::AuthQueue; use crate::routes::ApiError; use crate::routes::internal::delphi; -use crate::search::{SearchBackend, SearchQuery, SearchRequest, SearchResults}; +use crate::search::{ + SearchBackend, SearchQuery, SearchRequest, SearchResults, SearchState, +}; use crate::util::error::Context; use crate::util::img; use crate::util::img::{delete_old_images, upload_image_optimized}; @@ -73,6 +75,25 @@ pub fn utoipa_config( .service(dependency_list); } +pub async fn clear_project_cache_and_queue_search( + redis: &RedisPool, + search_state: &SearchState, + project_id: db_ids::DBProjectId, + slug: Option, + clear_dependencies: Option, +) -> Result<(), ApiError> { + db_models::DBProject::clear_cache( + project_id, + slug, + clear_dependencies, + redis, + ) + .await?; + search_state.queue.push(project_id.into()).await; + + Ok(()) +} + #[derive(Deserialize, Validate)] pub struct RandomProjects { #[validate(range(min = 1, max = 100))] @@ -291,21 +312,21 @@ async fn project_edit( req: HttpRequest, info: web::Path<(String,)>, pool: web::Data, - search_backend: web::Data, web::Json(new_project): web::Json, redis: web::Data, session_queue: web::Data, moderation_queue: web::Data, + search_state: web::Data, ) -> Result { project_edit_internal( req, info, pool, - search_backend, web::Json(new_project), redis, session_queue, moderation_queue, + search_state, ) .await } @@ -314,11 +335,11 @@ pub async fn project_edit_internal( req: HttpRequest, info: web::Path<(String,)>, pool: web::Data, - search_backend: web::Data, web::Json(new_project): web::Json, redis: web::Data, session_queue: web::Data, moderation_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -977,6 +998,7 @@ pub async fn project_edit_internal( ..Default::default() }, session_queue.clone(), + search_state.clone(), ) .await { @@ -1113,11 +1135,12 @@ pub async fn project_edit_internal( transaction.commit().await?; - db_models::DBProject::clear_cache( + clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; @@ -1126,7 +1149,8 @@ pub async fn project_edit_internal( project_item.inner.status.is_searchable(), new_project.status.map(|status| status.is_searchable()), ) { - search_backend + search_state + .backend .remove_documents( &project_item .versions @@ -1398,6 +1422,7 @@ pub async fn projects_edit( bulk_edit_project: web::Json, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -1471,6 +1496,7 @@ pub async fn projects_edit( db_models::categories::LinkPlatform::list(&**pool, &redis).await?; let mut transaction = pool.begin().await?; + let mut changed_projects = Vec::new(); for project in projects_data { if !user.role.is_mod() { @@ -1595,17 +1621,22 @@ pub async fn projects_edit( } } - db_models::DBProject::clear_cache( - project.inner.id, - project.inner.slug, - None, - &redis, - ) - .await?; + changed_projects.push((project.inner.id, project.inner.slug)); } transaction.commit().await?; + for (project_id, slug) in changed_projects { + clear_project_cache_and_queue_search( + &redis, + &search_state, + project_id, + slug, + None, + ) + .await?; + } + Ok(HttpResponse::NoContent().body("")) } @@ -1697,6 +1728,7 @@ async fn project_icon_edit( file_host: web::Data>, payload: web::Payload, session_queue: web::Data, + search_state: web::Data, ) -> Result { project_icon_edit_internal( web::Query(ext), @@ -1707,6 +1739,7 @@ async fn project_icon_edit( file_host, payload, session_queue, + search_state, ) .await } @@ -1720,6 +1753,7 @@ pub async fn project_icon_edit_internal( file_host: web::Data>, mut payload: web::Payload, session_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -1815,11 +1849,12 @@ pub async fn project_icon_edit_internal( .await?; transaction.commit().await?; - db_models::DBProject::clear_cache( + clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; @@ -1835,6 +1870,7 @@ async fn delete_project_icon( redis: web::Data, file_host: web::Data>, session_queue: web::Data, + search_state: web::Data, ) -> Result { delete_project_icon_internal( req, @@ -1843,6 +1879,7 @@ async fn delete_project_icon( redis, file_host, session_queue, + search_state, ) .await } @@ -1854,6 +1891,7 @@ pub async fn delete_project_icon_internal( redis: web::Data, file_host: web::Data>, session_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -1926,11 +1964,12 @@ pub async fn delete_project_icon_internal( .await?; transaction.commit().await?; - db_models::DBProject::clear_cache( + clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; @@ -1960,6 +1999,7 @@ pub async fn add_gallery_item( file_host: web::Data>, payload: web::Payload, session_queue: web::Data, + search_state: web::Data, ) -> Result { add_gallery_item_internal( web::Query(ext), @@ -1971,6 +2011,7 @@ pub async fn add_gallery_item( file_host, payload, session_queue, + search_state, ) .await } @@ -1985,6 +2026,7 @@ pub async fn add_gallery_item_internal( file_host: web::Data>, mut payload: web::Payload, session_queue: web::Data, + search_state: web::Data, ) -> Result { item.validate().map_err(|err| { ApiError::Validation(validation_errors_to_string(err, None)) @@ -2109,11 +2151,12 @@ pub async fn add_gallery_item_internal( .await?; transaction.commit().await?; - db_models::DBProject::clear_cache( + clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; @@ -2150,6 +2193,7 @@ async fn edit_gallery_item( pool: web::Data, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { edit_gallery_item_internal( req, @@ -2157,6 +2201,7 @@ async fn edit_gallery_item( pool, redis, session_queue, + search_state, ) .await } @@ -2167,6 +2212,7 @@ pub async fn edit_gallery_item_internal( pool: web::Data, redis: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -2311,11 +2357,12 @@ pub async fn edit_gallery_item_internal( transaction.commit().await?; - db_models::DBProject::clear_cache( + clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; @@ -2336,6 +2383,7 @@ async fn delete_gallery_item( redis: web::Data, file_host: web::Data>, session_queue: web::Data, + search_state: web::Data, ) -> Result { delete_gallery_item_internal( req, @@ -2344,6 +2392,7 @@ async fn delete_gallery_item( redis, file_host, session_queue, + search_state, ) .await } @@ -2355,6 +2404,7 @@ pub async fn delete_gallery_item_internal( redis: web::Data, file_host: web::Data>, session_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -2447,11 +2497,12 @@ pub async fn delete_gallery_item_internal( transaction.commit().await?; - db_models::DBProject::clear_cache( + clear_project_cache_and_queue_search( + &redis, + &search_state, project_item.inner.id, project_item.inner.slug, None, - &redis, ) .await?; @@ -2465,18 +2516,11 @@ async fn project_delete( info: web::Path<(String,)>, pool: web::Data, redis: web::Data, - search_backend: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result<(), ApiError> { - project_delete_internal( - req, - info, - pool, - redis, - search_backend, - session_queue, - ) - .await + project_delete_internal(req, info, pool, redis, session_queue, search_state) + .await } pub async fn project_delete_internal( @@ -2484,8 +2528,8 @@ pub async fn project_delete_internal( info: web::Path<(String,)>, pool: web::Data, redis: web::Data, - search_backend: web::Data, session_queue: web::Data, + search_state: web::Data, ) -> Result<(), ApiError> { let (_, user) = get_user_from_headers( &req, @@ -2596,7 +2640,8 @@ pub async fn project_delete_internal( .await .wrap_internal_err("failed to commit transaction")?; - search_backend + search_state + .backend .remove_documents( &project .versions @@ -2608,6 +2653,14 @@ pub async fn project_delete_internal( .wrap_internal_err("failed to remove project version documents")?; if result.is_some() { + clear_project_cache_and_queue_search( + &redis, + &search_state, + project.inner.id, + project.inner.slug, + None, + ) + .await?; Ok(()) } else { Err(ApiError::NotFound) diff --git a/apps/labrinth/src/routes/v3/version_creation.rs b/apps/labrinth/src/routes/v3/version_creation.rs index 26e335639..650f153b9 100644 --- a/apps/labrinth/src/routes/v3/version_creation.rs +++ b/apps/labrinth/src/routes/v3/version_creation.rs @@ -27,6 +27,7 @@ use crate::models::projects::{DependencyType, ProjectStatus, skip_nulls}; use crate::models::teams::ProjectPermissions; use crate::queue::moderation::AutomatedModerationQueue; use crate::queue::session::AuthQueue; +use crate::search::SearchState; use crate::util::http::HttpClient; use crate::util::routes::read_from_field; use crate::util::validate::validation_errors_to_string; @@ -114,6 +115,7 @@ pub async fn version_create( session_queue: Data, moderation_queue: web::Data, http: web::Data, + search_state: Data, ) -> Result { let mut transaction = client.begin().await?; let mut uploaded_files = Vec::new(); @@ -144,11 +146,19 @@ pub async fn version_create( if let Err(e) = rollback_result { return Err(e.into()); } - } else { + } else if let Ok((_, project_id)) = &result { transaction.commit().await?; + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, + *project_id, + None, + Some(true), + ) + .await?; } - result + result.map(|(response, _)| response) } #[allow(clippy::too_many_arguments)] @@ -163,7 +173,7 @@ async fn version_create_inner( session_queue: &AuthQueue, moderation_queue: &AutomatedModerationQueue, http: &reqwest::Client, -) -> Result { +) -> Result<(HttpResponse, models::DBProjectId), CreateError> { let mut initial_version_data = None; let mut version_builder = None; let mut selected_loaders = None; @@ -520,8 +530,6 @@ async fn version_create_inner( } } - models::DBProject::clear_cache(project_id, None, Some(true), redis).await?; - let project_status = sqlx::query!( "SELECT status FROM mods WHERE id = $1", project_id as models::DBProjectId, @@ -535,7 +543,7 @@ async fn version_create_inner( moderation_queue.projects.insert(project_id.into()); } - Ok(HttpResponse::Ok().json(response)) + Ok((HttpResponse::Ok().json(response), project_id)) } pub async fn upload_file_to_version( @@ -547,6 +555,7 @@ pub async fn upload_file_to_version( file_host: Data>, session_queue: web::Data, http: web::Data, + search_state: Data, ) -> Result { let mut transaction = client.begin().await?; let mut uploaded_files = Vec::new(); @@ -558,7 +567,7 @@ pub async fn upload_file_to_version( &mut payload, client, &mut transaction, - redis, + redis.clone(), &***file_host, &mut uploaded_files, version_id, @@ -579,11 +588,19 @@ pub async fn upload_file_to_version( if let Err(e) = rollback_result { return Err(e.into()); } - } else { + } else if let Ok((_, project_id)) = &result { transaction.commit().await?; + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, + *project_id, + None, + Some(true), + ) + .await?; } - result + result.map(|(response, _)| response) } #[allow(clippy::too_many_arguments)] @@ -598,7 +615,7 @@ async fn upload_file_to_version_inner( version_id: models::DBVersionId, session_queue: &AuthQueue, http: &reqwest::Client, -) -> Result { +) -> Result<(HttpResponse, models::DBProjectId), CreateError> { let mut initial_file_data: Option = None; let mut file_builders: Vec = Vec::new(); @@ -691,6 +708,7 @@ async fn upload_file_to_version_inner( } let project_id = ProjectId(version.inner.project_id.0 as u64); + let db_project_id = version.inner.project_id; let mut error = None; while let Some(item) = payload.next().await { let mut field: Field = item?; @@ -788,7 +806,7 @@ async fn upload_file_to_version_inner( // Clear version cache models::DBVersion::clear_cache(&version, &redis).await?; - Ok(HttpResponse::NoContent().body("")) + Ok((HttpResponse::NoContent().body(""), db_project_id)) } // This function is used for adding a file to a version, uploading the initial diff --git a/apps/labrinth/src/routes/v3/versions.rs b/apps/labrinth/src/routes/v3/versions.rs index 5011d346a..8a1098732 100644 --- a/apps/labrinth/src/routes/v3/versions.rs +++ b/apps/labrinth/src/routes/v3/versions.rs @@ -26,7 +26,7 @@ use crate::models::projects::{Loader, skip_nulls}; use crate::models::teams::ProjectPermissions; use crate::queue::session::AuthQueue; use crate::routes::internal::delphi; -use crate::search::SearchBackend; +use crate::search::{SearchBackend, SearchState}; use crate::util::error::Context; use crate::util::img; use crate::util::validate::validation_errors_to_string; @@ -267,6 +267,7 @@ pub async fn version_edit( redis: web::Data, new_version: web::Json, session_queue: web::Data, + search_state: web::Data, ) -> Result { let new_version: EditVersion = serde_json::from_value(new_version.into_inner())?; @@ -277,6 +278,7 @@ pub async fn version_edit( redis, new_version, session_queue, + search_state, ) .await } @@ -287,6 +289,7 @@ pub async fn version_edit_helper( redis: web::Data, new_version: EditVersion, session_queue: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -535,14 +538,6 @@ pub async fn version_edit_helper( } DBLoaderVersion::insert_many(loader_versions, &mut transaction) .await?; - - crate::database::models::DBProject::clear_cache( - version_item.inner.project_id, - None, - None, - &redis, - ) - .await?; } if let Some(featured) = &new_version.featured { @@ -696,11 +691,12 @@ pub async fn version_edit_helper( transaction.commit().await?; database::models::DBVersion::clear_cache(&version_item, &redis) .await?; - database::models::DBProject::clear_cache( + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, version_item.inner.project_id, None, Some(true), - &redis, ) .await?; Ok(HttpResponse::NoContent().body("")) @@ -920,6 +916,7 @@ pub async fn version_delete( redis: web::Data, session_queue: web::Data, search_backend: web::Data, + search_state: web::Data, ) -> Result { let user = get_user_from_headers( &req, @@ -1019,11 +1016,12 @@ pub async fn version_delete( transaction.commit().await?; - database::models::DBProject::clear_cache( + super::projects::clear_project_cache_and_queue_search( + &redis, + &search_state, version.inner.project_id, None, Some(true), - &redis, ) .await?; search_backend diff --git a/apps/labrinth/src/search/backend/typesense/mod.rs b/apps/labrinth/src/search/backend/typesense/mod.rs index 20122c7f9..27fe25601 100644 --- a/apps/labrinth/src/search/backend/typesense/mod.rs +++ b/apps/labrinth/src/search/backend/typesense/mod.rs @@ -12,7 +12,7 @@ use tracing::{info, warn}; use crate::database::PgPool; use crate::database::redis::RedisPool; use crate::env::ENV; -use crate::models::ids::VersionId; +use crate::models::ids::{ProjectId, VersionId}; use crate::routes::ApiError; use crate::search::backend::{ SearchIndex, SearchIndexName, combined_search_filters, parse_search_index, @@ -989,6 +989,75 @@ impl SearchBackend for Typesense { Ok(()) } + async fn index_documents( + &self, + documents: &[UploadSearchProject], + ) -> eyre::Result<()> { + if documents.is_empty() { + return Ok(()); + } + + let jsonl = documents_to_jsonl(documents)?; + for alias in [ + self.config.get_alias_name("projects"), + self.config.get_alias_name("projects_filtered"), + ] { + let live = self.client.get_alias(&alias).await?; + let shadow_alt = self.config.get_next_collection_name(&alias, true); + let shadow_current = + self.config.get_next_collection_name(&alias, false); + + for collection in + live.into_iter().chain([shadow_alt, shadow_current]) + { + if self.client.collection_exists(&collection).await? { + self.client + .import_documents(&collection, jsonl.clone()) + .await?; + } + } + } + + Ok(()) + } + + async fn remove_project_documents( + &self, + ids: &[ProjectId], + ) -> eyre::Result<()> { + if ids.is_empty() { + return Ok(()); + } + + let id_list = ids + .iter() + .map(|id| to_base62(id.0)) + .collect::>() + .join(", "); + let filter = format!("project_id:[{id_list}]"); + + for alias in [ + self.config.get_alias_name("projects"), + self.config.get_alias_name("projects_filtered"), + ] { + let live = self.client.get_alias(&alias).await?; + let shadow_alt = self.config.get_next_collection_name(&alias, true); + let shadow_current = + self.config.get_next_collection_name(&alias, false); + + for collection in + live.into_iter().chain([shadow_alt, shadow_current]) + { + if self.client.collection_exists(&collection).await? { + self.client + .delete_documents_by_filter(&collection, &filter) + .await?; + } + } + } + Ok(()) + } + async fn remove_documents(&self, ids: &[VersionId]) -> eyre::Result<()> { if ids.is_empty() { return Ok(()); diff --git a/apps/labrinth/src/search/incremental.rs b/apps/labrinth/src/search/incremental.rs new file mode 100644 index 000000000..ed2c6965b --- /dev/null +++ b/apps/labrinth/src/search/incremental.rs @@ -0,0 +1,101 @@ +pub mod consume; + +use std::{mem, sync::Arc}; + +use rdkafka::{producer::FutureRecord, util::Timeout}; +use serde::Serialize; +use tokio::sync::Mutex; + +use crate::{ + models::ids::ProjectId, + util::kafka::{KAFKA_OPERATION_INTERVAL, KafkaClientState, KafkaEvent}, +}; + +pub const SEARCH_PROJECT_INDEX_QUEUE_TOPIC: &str = + "public.labrinth.search_project_index_queue.v1"; + +#[derive(Clone)] +pub struct IncrementalSearchQueue { + operations: Arc>>, + kafka_client: actix_web::web::Data, +} + +impl IncrementalSearchQueue { + pub fn new(kafka_client: actix_web::web::Data) -> Self { + Self { + operations: Arc::new(Mutex::new(Vec::new())), + kafka_client, + } + } + + pub async fn push(&self, project_id: ProjectId) { + self.operations + .lock() + .await + .push(SearchIndexOperation { project_id }); + } + + pub async fn run(self) { + loop { + tokio::time::sleep(KAFKA_OPERATION_INTERVAL).await; + + if let Err(err) = self.drain().await { + tracing::error!( + "Failed to drain incremental search queue: {err:?}" + ); + } + } + } + + pub async fn drain(&self) -> eyre::Result<()> { + let operations = { + let mut operations = self.operations.lock().await; + mem::take(&mut *operations) + }; + + if operations.is_empty() { + return Ok(()); + } + + let mut operations = operations.into_iter(); + while let Some(operation) = operations.next() { + let event = KafkaEvent::new( + SEARCH_PROJECT_INDEX_QUEUE_TOPIC, + SearchProjectIndexQueueEventData { + project_id: operation.project_id, + }, + ); + let event_id = event.event_metadata.event_id; + let key = event_id.to_string(); + let payload = serde_json::to_vec(&event)?; + let record = FutureRecord::to(SEARCH_PROJECT_INDEX_QUEUE_TOPIC) + .key(&key) + .payload(&payload); + + if let Err((err, _)) = self + .kafka_client + .client + .send(record, Timeout::After(KAFKA_OPERATION_INTERVAL)) + .await + { + let mut queued_operations = self.operations.lock().await; + queued_operations.push(operation); + queued_operations.extend(operations); + + return Err(err.into()); + } + } + + Ok(()) + } +} + +#[derive(Debug, Clone)] +pub struct SearchIndexOperation { + pub project_id: ProjectId, +} + +#[derive(Debug, Serialize)] +pub struct SearchProjectIndexQueueEventData { + pub project_id: ProjectId, +} diff --git a/apps/labrinth/src/search/incremental/consume.rs b/apps/labrinth/src/search/incremental/consume.rs new file mode 100644 index 000000000..7c3ee44a2 --- /dev/null +++ b/apps/labrinth/src/search/incremental/consume.rs @@ -0,0 +1,132 @@ +use actix_web::web; +use eyre::WrapErr; +use rdkafka::{ + ClientConfig, Message, + consumer::{CommitMode, Consumer, StreamConsumer}, +}; +use serde::Deserialize; + +use crate::{ + database::{PgPool, redis::RedisPool}, + env::ENV, + models::ids::ProjectId, + search::{ + SearchBackend, incremental::SEARCH_PROJECT_INDEX_QUEUE_TOPIC, + indexing::index_project_documents, + }, + util::kafka::KAFKA_OPERATION_INTERVAL, +}; + +pub const INCREMENTAL_INDEX_SEARCH_TASK: &str = "incremental-index-search"; + +pub async fn run( + ro_pool: PgPool, + redis_pool: RedisPool, + search_backend: web::Data, +) -> eyre::Result<()> { + loop { + if let Err(err) = + consume(&ro_pool, &redis_pool, search_backend.as_ref()).await + { + tracing::error!( + "Background task {INCREMENTAL_INDEX_SEARCH_TASK} failed: {err:?}" + ); + tokio::time::sleep(KAFKA_OPERATION_INTERVAL).await; + } + } +} + +async fn consume( + ro_pool: &PgPool, + redis_pool: &RedisPool, + search_backend: &dyn SearchBackend, +) -> eyre::Result<()> { + let consumer: StreamConsumer = ClientConfig::new() + .set("bootstrap.servers", ENV.KAFKA_BOOTSTRAP_SERVERS.0.join(",")) + .set("client.id", &ENV.KAFKA_CLIENT_ID) + .set("group.id", INCREMENTAL_INDEX_SEARCH_TASK) + .set("enable.auto.commit", "false") + .set("auto.offset.reset", "earliest") + .set("broker.address.family", "v4") + .create() + .wrap_err("failed to create Kafka consumer")?; + consumer + .subscribe(&[SEARCH_PROJECT_INDEX_QUEUE_TOPIC]) + .wrap_err("failed to subscribe to Kafka topic")?; + + tracing::info!( + kafka.bootstrap_servers = ?ENV.KAFKA_BOOTSTRAP_SERVERS.0, + kafka.client_id = %ENV.KAFKA_CLIENT_ID, + kafka.topic = SEARCH_PROJECT_INDEX_QUEUE_TOPIC, + kafka.consumer_group = INCREMENTAL_INDEX_SEARCH_TASK, + "Started Kafka consumer" + ); + + loop { + let message = consumer + .recv() + .await + .wrap_err("failed to receive Kafka message")?; + let payload = message.payload().ok_or_else(|| { + eyre::eyre!("Kafka message did not have a payload") + })?; + let event = match serde_json::from_slice::( + payload, + ) { + Ok(event) => event, + Err(err) => { + tracing::error!( + kafka.topic = message.topic(), + kafka.partition = message.partition(), + kafka.offset = message.offset(), + "Skipping malformed incremental search index event: {err:?}" + ); + consumer + .commit_message(&message, CommitMode::Async) + .wrap_err("failed to commit malformed Kafka message")?; + continue; + } + }; + + tracing::info!( + kafka.topic = message.topic(), + kafka.partition = message.partition(), + kafka.offset = message.offset(), + project_id = %event.project_id, + "Consumed incremental search index event" + ); + + reindex_project(ro_pool, redis_pool, search_backend, event.project_id) + .await + .wrap_err_with(|| { + format!("failed to reindex project {}", event.project_id) + })?; + + consumer + .commit_message(&message, CommitMode::Async) + .wrap_err("failed to commit Kafka message")?; + } +} + +pub async fn reindex_project( + ro_pool: &PgPool, + redis_pool: &RedisPool, + search_backend: &dyn SearchBackend, + project_id: ProjectId, +) -> eyre::Result<()> { + search_backend + .remove_project_documents(&[project_id]) + .await?; + + let documents = index_project_documents(ro_pool, redis_pool, project_id) + .await + .wrap_err("failed to build project search documents")?; + search_backend.index_documents(&documents).await?; + + Ok(()) +} + +#[derive(Debug, Deserialize)] +struct SearchProjectIndexQueueEvent { + project_id: ProjectId, +} diff --git a/apps/labrinth/src/search/indexing.rs b/apps/labrinth/src/search/indexing.rs index 96793510f..3c4c8a762 100644 --- a/apps/labrinth/src/search/indexing.rs +++ b/apps/labrinth/src/search/indexing.rs @@ -27,6 +27,21 @@ use crate::routes::v2_reroute; use crate::search::{SearchProjectDependency, UploadSearchProject}; use crate::util::error::Context; +struct PartialProject { + id: DBProjectId, + name: String, + summary: String, + downloads: i32, + follows: i32, + icon_url: Option, + updated: DateTime, + approved: DateTime, + slug: Option, + color: Option, + license: String, + components: exp::ProjectSerial, +} + fn normalize_for_search(s: &str) -> String { static SPECIAL_CHARS_RE: LazyLock = LazyLock::new(|| Regex::new(r"[^a-zA-Z0-9-.\s]").expect("valid regex")); @@ -34,6 +49,13 @@ fn normalize_for_search(s: &str) -> String { SPECIAL_CHARS_RE.replace_all(s, "").to_kebab_case() } +fn searchable_statuses() -> Vec { + crate::models::projects::ProjectStatus::iterator() + .filter(|x| x.is_searchable()) + .map(|x| x.to_string()) + .collect() +} + struct ProjectOwner { username: String, user_id: DBUserId, @@ -49,30 +71,10 @@ pub async fn index_local( ) -> eyre::Result<(Vec, i64)> { info!("Indexing local projects!"); - // todo: loaders, project type, game versions - struct PartialProject { - id: DBProjectId, - name: String, - summary: String, - downloads: i32, - follows: i32, - icon_url: Option, - updated: DateTime, - approved: DateTime, - slug: Option, - color: Option, - license: String, - components: exp::ProjectSerial, - } - - let searchable_statuses = - crate::models::projects::ProjectStatus::iterator() - .filter(|x| x.is_searchable()) - .map(|x| x.to_string()) - .collect::>(); + let searchable_statuses = searchable_statuses(); let db_projects = sqlx::query!( - r#" + r#" SELECT m.id id, m.name name, m.summary summary, m.downloads downloads, m.follows follows, m.icon_url icon_url, m.updated updated, m.approved approved, m.published, m.license license, m.slug slug, m.color, m.components AS "components: sqlx::types::Json" @@ -107,6 +109,64 @@ pub async fn index_local( .await .wrap_err("failed to fetch projects")?; + let project_ids = db_projects.iter().map(|x| x.id.0).collect::>(); + let Some(largest) = project_ids.iter().max() else { + return Ok((vec![], i64::MAX)); + }; + + let uploads = build_search_documents(pool, redis, db_projects).await?; + Ok((uploads, *largest)) +} + +pub async fn index_project_documents( + pool: &PgPool, + redis: &RedisPool, + project_id: ProjectId, +) -> eyre::Result> { + let searchable_statuses = searchable_statuses(); + let project_ids = vec![DBProjectId::from(project_id).0]; + + let db_projects = sqlx::query!( + r#" + SELECT m.id id, m.name name, m.summary summary, m.downloads downloads, m.follows follows, + m.icon_url icon_url, m.updated updated, m.approved approved, m.published, m.license license, m.slug slug, m.color, + m.components AS "components: sqlx::types::Json" + FROM mods m + WHERE m.status = ANY($1) AND m.id = ANY($2) + GROUP BY m.id + ORDER BY m.id ASC; + "#, + &searchable_statuses, + &project_ids, + ) + .fetch(pool) + .map_ok(|m| PartialProject { + id: DBProjectId(m.id), + name: m.name, + summary: m.summary, + downloads: m.downloads, + follows: m.follows, + icon_url: m.icon_url, + updated: m.updated, + approved: m.approved.unwrap_or(m.published), + slug: m.slug, + color: m.color, + license: m.license, + components: m.components.0, + }) + .try_collect::>() + .await + .wrap_err("failed to fetch project")?; + + build_search_documents(pool, redis, db_projects).await +} + +async fn build_search_documents( + pool: &PgPool, + redis: &RedisPool, + db_projects: Vec, +) -> eyre::Result> { + let searchable_statuses = searchable_statuses(); let project_ids = db_projects.iter().map(|x| x.id.0).collect::>(); let project_components = db_projects .iter() @@ -117,10 +177,6 @@ pub async fn index_local( .await .wrap_err("failed to fetch query context")?; - let Some(largest) = project_ids.iter().max() else { - return Ok((vec![], i64::MAX)); - }; - info!("Indexing local dependencies!"); let dependencies: DashMap> = @@ -573,7 +629,7 @@ pub async fn index_local( } } - Ok((uploads, *largest)) + Ok(uploads) } struct PartialVersion { diff --git a/apps/labrinth/src/search/mod.rs b/apps/labrinth/src/search/mod.rs index 0f46a8111..9507aafc2 100644 --- a/apps/labrinth/src/search/mod.rs +++ b/apps/labrinth/src/search/mod.rs @@ -11,13 +11,20 @@ use async_trait::async_trait; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use serde_json::Value; -use std::{collections::HashMap, str::FromStr}; +use std::{collections::HashMap, str::FromStr, sync::Arc}; use thiserror::Error; use utoipa::ToSchema; pub mod backend; +pub mod incremental; pub mod indexing; +#[derive(Clone)] +pub struct SearchState { + pub backend: Arc, + pub queue: incremental::IncrementalSearchQueue, +} + /// Search parameters which can fit in a URL query string. /// /// Used with `GET /*/search` endpoints. @@ -104,6 +111,16 @@ pub trait SearchBackend: Send + Sync { redis: RedisPool, ) -> eyre::Result<()>; + async fn index_documents( + &self, + documents: &[UploadSearchProject], + ) -> eyre::Result<()>; + + async fn remove_project_documents( + &self, + ids: &[ProjectId], + ) -> eyre::Result<()>; + async fn remove_documents(&self, ids: &[VersionId]) -> eyre::Result<()>; async fn tasks(&self) -> eyre::Result; diff --git a/apps/labrinth/src/test/database.rs b/apps/labrinth/src/test/database.rs index cd6a5fe53..47d54dc19 100644 --- a/apps/labrinth/src/test/database.rs +++ b/apps/labrinth/src/test/database.rs @@ -170,11 +170,11 @@ impl TemporaryDatabase { "Dummy data updated, so template DB tables will be dropped and re-created" ); // Drop all tables in the database so they can be re-created and later filled with updated dummy data - sqlx::query("DROP SCHEMA public CASCADE;") + sqlx::query!("DROP SCHEMA public CASCADE;") .execute(&pool) .await .unwrap(); - sqlx::query("CREATE SCHEMA public;") + sqlx::query!("CREATE SCHEMA public;") .execute(&pool) .await .unwrap(); @@ -211,14 +211,14 @@ impl TemporaryDatabase { &temp_database_name, TEMPLATE_DATABASE_NAME ); - sqlx::query(&create_db_query) + sqlx::raw_sql(&create_db_query) .execute(&main_pool) .await .expect("Database creation failed"); // Release the advisory lock - sqlx::query("SELECT pg_advisory_unlock(1)") - .execute(&main_pool) + sqlx::query_scalar!("SELECT pg_advisory_unlock(1)") + .fetch_one(&main_pool) .await .unwrap(); @@ -248,7 +248,7 @@ impl TemporaryDatabase { "SELECT pg_terminate_backend(pg_stat_activity.pid) FROM pg_stat_activity WHERE datname = '{}' AND pid <> pg_backend_pid()", &self.database_name ); - sqlx::query(&terminate_query) + sqlx::raw_sql(&terminate_query) .execute(&self.pool) .await .unwrap(); @@ -256,7 +256,7 @@ impl TemporaryDatabase { // Execute the deletion query asynchronously let drop_db_query = format!("DROP DATABASE IF EXISTS {}", &self.database_name); - sqlx::query(&drop_db_query) + sqlx::raw_sql(&drop_db_query) .execute(&self.pool) .await .expect("Database deletion failed"); @@ -265,7 +265,7 @@ impl TemporaryDatabase { async fn create_template_database(pool: &PgPool) { let create_db_query = format!("CREATE DATABASE {TEMPLATE_DATABASE_NAME}"); - sqlx::query(&create_db_query) + sqlx::raw_sql(&create_db_query) .execute(pool) .await .expect("Database creation failed"); diff --git a/apps/labrinth/src/test/dummy_data.rs b/apps/labrinth/src/test/dummy_data.rs index 3211f32e2..547c58abe 100644 --- a/apps/labrinth/src/test/dummy_data.rs +++ b/apps/labrinth/src/test/dummy_data.rs @@ -296,11 +296,12 @@ pub async fn add_dummy_data(api: &ApiV3, db: TemporaryDatabase) -> DummyData { let oauth_client_alpha = get_oauth_client_alpha(api).await; - sqlx::query("INSERT INTO dummy_data (update_id) VALUES ($1)") - .bind(DUMMY_DATA_UPDATE) - .execute(pool) - .await - .unwrap(); + sqlx::raw_sql(&format!( + "INSERT INTO dummy_data (update_id) VALUES ({DUMMY_DATA_UPDATE})" + )) + .execute(pool) + .await + .unwrap(); DummyData::new( alpha_project, diff --git a/apps/labrinth/src/test/mod.rs b/apps/labrinth/src/test/mod.rs index 0e0852404..9a7520f28 100644 --- a/apps/labrinth/src/test/mod.rs +++ b/apps/labrinth/src/test/mod.rs @@ -42,6 +42,10 @@ pub async fn setup(db: &database::TemporaryDatabase) -> LabrinthConfig { EmailQueue::init(pool.clone(), redis_pool.clone()).unwrap(); let gotenberg_client = GotenbergClient::from_env(redis_pool.clone()) .expect("Failed to create Gotenberg client"); + let kafka_client = actix_web::web::Data::new( + crate::util::kafka::KafkaClientState::new() + .expect("Kafka connection failed"), + ); crate::app_setup( pool.clone(), @@ -54,6 +58,7 @@ pub async fn setup(db: &database::TemporaryDatabase) -> LabrinthConfig { anrok_client, email_queue, gotenberg_client, + kafka_client, false, ) } diff --git a/apps/labrinth/src/util/kafka.rs b/apps/labrinth/src/util/kafka.rs new file mode 100644 index 000000000..6f6d9a056 --- /dev/null +++ b/apps/labrinth/src/util/kafka.rs @@ -0,0 +1,83 @@ +use std::time::Duration; + +use crate::env::ENV; +use chrono::{DateTime, Utc}; +use rdkafka::{ClientConfig, producer::FutureProducer}; +use serde::Serialize; +use uuid::Uuid; + +pub const KAFKA_OPERATION_INTERVAL: Duration = Duration::from_secs(5); + +pub struct KafkaClientState { + pub client: FutureProducer, +} + +impl KafkaClientState { + pub fn new() -> eyre::Result { + let client = ClientConfig::new() + .set("bootstrap.servers", ENV.KAFKA_BOOTSTRAP_SERVERS.0.join(",")) + .set("client.id", &ENV.KAFKA_CLIENT_ID) + .set("broker.address.family", "v4") + .create()?; + + tracing::info!( + kafka.bootstrap_servers = ?ENV.KAFKA_BOOTSTRAP_SERVERS.0, + kafka.client_id = %ENV.KAFKA_CLIENT_ID, + "Connected to Kafka" + ); + + Ok(Self { client }) + } +} + +#[derive(Debug, Serialize)] +pub struct KafkaEvent { + pub event_type: &'static str, + pub event_metadata: EventMetadata, + #[serde(flatten)] + pub data: T, +} + +impl KafkaEvent { + pub fn new(event_type: &'static str, data: T) -> Self { + Self { + event_type, + event_metadata: EventMetadata::new(), + data, + } + } +} + +#[derive(Debug, Serialize)] +pub struct EventMetadata { + pub event_id: Uuid, + pub event_time: DateTime, + pub service: &'static ServiceMetadata, +} + +impl EventMetadata { + pub fn new() -> Self { + Self { + event_id: Uuid::now_v7(), + event_time: Utc::now(), + service: &SERVICE_METADATA, + } + } +} + +impl Default for EventMetadata { + fn default() -> Self { + Self::new() + } +} + +#[derive(Debug, Serialize)] +pub struct ServiceMetadata { + pub service_name: &'static str, + pub service_version: &'static str, +} + +pub static SERVICE_METADATA: ServiceMetadata = ServiceMetadata { + service_name: "labrinth", + service_version: env!("CARGO_PKG_VERSION"), +}; diff --git a/apps/labrinth/src/util/mod.rs b/apps/labrinth/src/util/mod.rs index 97fd0ea4a..87676d448 100644 --- a/apps/labrinth/src/util/mod.rs +++ b/apps/labrinth/src/util/mod.rs @@ -13,6 +13,7 @@ pub mod guards; pub mod http; pub mod img; pub mod ip; +pub mod kafka; pub mod ratelimit; pub mod redis; pub mod routes; diff --git a/apps/labrinth/tests/moderation_lock.rs b/apps/labrinth/tests/moderation_lock.rs index 04e3ee8ec..ef8cf0ea6 100644 --- a/apps/labrinth/tests/moderation_lock.rs +++ b/apps/labrinth/tests/moderation_lock.rs @@ -24,11 +24,13 @@ async fn set_project_processing( pool: &labrinth::database::PgPool, redis: &labrinth::database::redis::RedisPool, ) { - sqlx::query("UPDATE mods SET status = 'processing' WHERE id = $1") - .bind(project_id) - .execute(pool) - .await - .expect("failed to set project to processing status"); + sqlx::query!( + "UPDATE mods SET status = 'processing' WHERE id = $1", + project_id + ) + .execute(pool) + .await + .expect("failed to set project to processing status"); DBProject::clear_cache( DBProjectId(project_id), @@ -42,11 +44,11 @@ async fn set_project_processing( /// Back-date a lock's `locked_at` so it appears expired. async fn expire_lock(project_id: i64, pool: &labrinth::database::PgPool) { - sqlx::query( + sqlx::query!( "UPDATE moderation_locks SET locked_at = NOW() - ($1::bigint * INTERVAL '1 minute') - INTERVAL '1 second' WHERE project_id = $2", + LOCK_EXPIRY_MINUTES, + project_id, ) - .bind(LOCK_EXPIRY_MINUTES) - .bind(project_id) .execute(pool) .await .expect("failed to expire lock"); diff --git a/apps/labrinth/tests/user.rs b/apps/labrinth/tests/user.rs index 3bfd4ac78..ee2e84124 100644 --- a/apps/labrinth/tests/user.rs +++ b/apps/labrinth/tests/user.rs @@ -25,7 +25,7 @@ pub async fn search_users_returns_compact_prefix_matches_with_exact_first() { with_test_environment( None, |test_env: TestEnvironment| async move { - sqlx::query( + sqlx::query!( " INSERT INTO users (id, username, email, role) VALUES @@ -65,21 +65,21 @@ pub async fn search_users_escapes_wildcards_and_limits_results() { None, |test_env: TestEnvironment| async move { for i in 0..30 { - sqlx::query( + sqlx::query!( " INSERT INTO users (id, username, email, role) VALUES ($1, $2, $3, 'developer') ", + 2000 + i, + format!("prefix{i:02}"), + format!("prefix{i:02}@modrinth.com"), ) - .bind(2000 + i) - .bind(format!("prefix{i:02}")) - .bind(format!("prefix{i:02}@modrinth.com")) .execute(&*test_env.db.pool) .await .unwrap(); } - sqlx::query( + sqlx::query!( " INSERT INTO users (id, username, email, role) VALUES (2100, 'prefix_under_score', 'prefix_under_score@modrinth.com', 'developer') diff --git a/docker-compose.yml b/docker-compose.yml index f64fe828a..38227c433 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -106,6 +106,34 @@ services: # Gotenberg must send a message on a webhook to our backend, # so it must have access to our local network - 'host.docker.internal:host-gateway' + redpanda: + image: redpandadata/redpanda:v26.1.10 + container_name: labrinth-redpanda + command: + - redpanda + - start + - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092 + - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092 + - --pandaproxy-addr internal://0.0.0.0:8082,external://0.0.0.0:18082 + - --advertise-pandaproxy-addr internal://redpanda:8082,external://localhost:18082 + - --schema-registry-addr internal://0.0.0.0:8081,external://0.0.0.0:18081 + - --rpc-addr redpanda:33145 + - --advertise-rpc-addr redpanda:33145 + - --mode dev-container + - --smp 1 + - --default-log-level=info + ports: + - '127.0.0.1:18081:18081' + - '127.0.0.1:18082:18082' + - '127.0.0.1:19092:19092' + - '127.0.0.1:19644:9644' + volumes: + - redpanda-data:/var/lib/redpanda/data + healthcheck: + test: ['CMD', 'rpk', 'cluster', 'info', '-X', 'brokers=localhost:9092'] + interval: 10s + timeout: 15s + retries: 10 labrinth: profiles: - with-labrinth @@ -131,6 +159,8 @@ services: condition: service_healthy redis: condition: service_healthy + redpanda: + condition: service_healthy clickhouse: condition: service_healthy mail: @@ -204,8 +234,6 @@ services: networks: meilisearch-mesh: driver: bridge - elasticsearch-mesh: - driver: bridge volumes: typesense-data: meilisearch-data: @@ -216,4 +244,5 @@ volumes: elasticsearch-certs: db-data: redis-data: + redpanda-data: labrinth-cdn-data: diff --git a/tombi.toml b/tombi.toml index 849b86878..068909c9f 100644 --- a/tombi.toml +++ b/tombi.toml @@ -1,2 +1,9 @@ [files] exclude = ["target/", "**/node_modules/"] + +# required since Tauri doesn't support TOML v1.1.0 +# and breaks on inline tables with newlines +[[schemas]] +toml-version = "v1.0.0" +path = "tombi://www.schemastore.org/cargo.json" +include = ["Cargo.toml"]