Incremental search indexing (#6453)

* wip: initial kafka support

* wip: kafka search queue

* wip: kafka consumer

* wip: create/delete docs

* reindexing works

* reindex in more places

* also incrementally update server pings

* cleanup

* tombi

* fix ci

* update tombi

* prepare

* fix ci

* fix

* fix PLEASE

* 🙏

* add route to force reindex project

.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

.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:

.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
+          packages: cmake libcurl4-openssl-dev libwebkit2gtk-4.1-dev libayatana-appindicator3-dev librsvg2-dev
-          version: v1 # cache key
+          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

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"

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 }

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
 

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

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"
+}

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"
+}

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"
+}

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"
+}

apps/labrinth/.sqlx/query-66868d0b35229abffa3f668eaef19dc0cdef502935dd5939a2f3ab89b0766ad2.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "CREATE SCHEMA public;",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "66868d0b35229abffa3f668eaef19dc0cdef502935dd5939a2f3ab89b0766ad2"
+}

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<exp::ProjectSerial>\"\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<exp::ProjectSerial>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "TextArray",
+        "Int8Array"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      true,
+      false,
+      true,
+      false,
+      false,
+      true,
+      true,
+      false
+    ]
+  },
+  "hash": "881de045a10c1debe041ec558fd92fdfbf8b8ffa1b21d3d4a271ab6f9c8adb78"
+}

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"
+}

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"
+}

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"
+}

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"
+}

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"
+}

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"
+}

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 }

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<dyn SearchBackend>,
+        kafka_client: web::Data<crate::util::kafka::KafkaClientState>,
         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<crate::util::kafka::KafkaClientState>,
 ) -> 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");

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?;
 

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 = "";
 

apps/labrinth/src/lib.rs

@@ -61,7 +61,7 @@ pub struct LabrinthConfig {
     pub file_host: Arc<dyn file_hosting::FileHost + Send + Sync>,
     pub scheduler: Arc<scheduler::Scheduler>,
     pub ip_salt: Pepper,
-    pub search_backend: web::Data<dyn search::SearchBackend>,
+    pub search_state: web::Data<search::SearchState>,
     pub session_queue: web::Data<AuthQueue>,
     pub payouts_queue: web::Data<PayoutsQueue>,
     pub analytics_queue: Arc<AnalyticsQueue>,
@@ -75,6 +75,7 @@ pub struct LabrinthConfig {
     pub gotenberg_client: GotenbergClient,
     pub http_client: web::Data<HttpClient>,
     pub tiltify_client: web::Data<TiltifyClient>,
+    pub kafka_client: web::Data<util::kafka::KafkaClientState>,
 }
 
 #[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<util::kafka::KafkaClientState>,
     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)

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,
     );
 

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::<Vec<_>>(),
+                &version_downloads.values().copied().collect::<Vec<_>>(),
             )
-            .bind(version_downloads.keys().copied().collect::<Vec<_>>())
-            .bind(version_downloads.values().copied().collect::<Vec<_>>())
             .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::<Vec<_>>(),
+                &project_downloads.values().copied().collect::<Vec<_>>(),
             )
-            .bind(project_downloads.keys().copied().collect::<Vec<_>>())
-            .bind(project_downloads.values().copied().collect::<Vec<_>>())
             .execute(&mut transaction)
             .await?;
 

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
+                    let queue_search =
-                    .inspect_err(|err| {
+                        self.incremental_search_queue.push(*project_id);
-                        warn!("failed to clear project cache: {err:#}")
+
-                    })
+                    let (clear_cache_result, _) =
-                    .ok();
+                        join(clear_cache, queue_search).await;
+                    clear_cache_result
+                        .inspect_err(|err| {
+                            warn!("failed to clear project cache: {err:#}")
+                        })
+                        .ok();
                 }
             }
 

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<PgPool>,
+    redis: web::Data<RedisPool>,
+    search_backend: web::Data<dyn SearchBackend>,
+) -> Result<HttpResponse, ApiError> {
+    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())
+}

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")

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<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
     web::Json(submit_report): web::Json<SubmitReport>,
     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(())
 }
 

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<Arc<dyn FileHost + Send + Sync>>,
     session_queue: Data<AuthQueue>,
     http: Data<HttpClient>,
+    search_state: Data<SearchState>,
 ) -> Result<HttpResponse, CreateError> {
     // 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?;
 

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<PgPool>,
-    search_backend: web::Data<dyn SearchBackend>,
     new_project: web::Json<EditProject>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
     moderation_queue: web::Data<AutomatedModerationQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<BulkEditProject>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<Arc<dyn FileHost + Send + Sync>>,
     payload: web::Payload,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // 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<RedisPool>,
     file_host: web::Data<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // 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<Arc<dyn FileHost + Send + Sync>>,
     payload: web::Payload,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // 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<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // 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<RedisPool>,
     file_host: web::Data<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // 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<PgPool>,
     redis: web::Data<RedisPool>,
-    search_backend: web::Data<dyn SearchBackend>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // 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(""))

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<AuthQueue>,
     moderation_queue: Data<AutomatedModerationQueue>,
     http: Data<HttpClient>,
+    search_state: Data<SearchState>,
 ) -> Result<HttpResponse, CreateError> {
     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<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
     http: web::Data<HttpClient>,
+    search_state: Data<SearchState>,
 ) -> Result<HttpResponse, CreateError> {
     // 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)

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<RedisPool>,
     new_version: web::Json<EditVersion>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<RedisPool>,
     session_queue: web::Data<AuthQueue>,
     search_backend: web::Data<dyn SearchBackend>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // 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)

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<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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(
+        super::projects::clear_project_cache_and_queue_search(
-            project_id, None, None, &redis,
+            &redis,
+            &search_state,
+            project_id,
+            None,
+            None,
         )
         .await?;
     }
@@ -829,6 +835,7 @@ pub async fn organization_projects_add(
     pool: web::Data<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<OrganizationProjectRemoval>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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 {

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<Arc<dyn FileHost + Send + Sync>>,
     session_queue: Data<AuthQueue>,
     http: Data<HttpClient>,
+    search_state: Data<SearchState>,
 ) -> Result<HttpResponse, CreateError> {
     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<Arc<dyn FileHost + Send + Sync>>,
     session_queue: Data<AuthQueue>,
     http: Data<HttpClient>,
+    search_state: Data<SearchState>,
 ) -> Result<HttpResponse, CreateError> {
     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<Arc<dyn FileHost + Send + Sync>>,
     session_queue: Data<AuthQueue>,
     http: Data<HttpClient>,
+    search_state: Data<SearchState>,
     path: web::Path<(ProjectId,)>,
 ) -> Result<HttpResponse, CreateError> {
     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

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<String>,
+    clear_dependencies: Option<bool>,
+) -> 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<PgPool>,
-    search_backend: web::Data<dyn SearchBackend>,
     web::Json(new_project): web::Json<EditProject>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
     moderation_queue: web::Data<AutomatedModerationQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<PgPool>,
-    search_backend: web::Data<dyn SearchBackend>,
     web::Json(new_project): web::Json<EditProject>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
     moderation_queue: web::Data<AutomatedModerationQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<BulkEditProject>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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(
+        changed_projects.push((project.inner.id, project.inner.slug));
-            project.inner.id,
-            project.inner.slug,
-            None,
-            &redis,
-        )
-        .await?;
     }
 
     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<Arc<dyn FileHost + Send + Sync>>,
     payload: web::Payload,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<Arc<dyn FileHost + Send + Sync>>,
     mut payload: web::Payload,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<RedisPool>,
     file_host: web::Data<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<RedisPool>,
     file_host: web::Data<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<Arc<dyn FileHost + Send + Sync>>,
     payload: web::Payload,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<Arc<dyn FileHost + Send + Sync>>,
     mut payload: web::Payload,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<RedisPool>,
     file_host: web::Data<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<RedisPool>,
     file_host: web::Data<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<PgPool>,
     redis: web::Data<RedisPool>,
-    search_backend: web::Data<dyn SearchBackend>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<(), ApiError> {
-    project_delete_internal(
+    project_delete_internal(req, info, pool, redis, session_queue, search_state)
-        req,
+        .await
-        info,
-        pool,
-        redis,
-        search_backend,
-        session_queue,
-    )
-    .await
 }
 
 pub async fn project_delete_internal(
@@ -2484,8 +2528,8 @@ pub async fn project_delete_internal(
     info: web::Path<(String,)>,
     pool: web::Data<PgPool>,
     redis: web::Data<RedisPool>,
-    search_backend: web::Data<dyn SearchBackend>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> 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)

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<AuthQueue>,
     moderation_queue: web::Data<AutomatedModerationQueue>,
     http: web::Data<HttpClient>,
+    search_state: Data<SearchState>,
 ) -> Result<HttpResponse, CreateError> {
     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<HttpResponse, CreateError> {
+) -> 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<Arc<dyn FileHost + Send + Sync>>,
     session_queue: web::Data<AuthQueue>,
     http: web::Data<HttpClient>,
+    search_state: Data<SearchState>,
 ) -> Result<HttpResponse, CreateError> {
     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<HttpResponse, CreateError> {
+) -> Result<(HttpResponse, models::DBProjectId), CreateError> {
     let mut initial_file_data: Option<InitialFileData> = None;
     let mut file_builders: Vec<VersionFileBuilder> = 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

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<RedisPool>,
     new_version: web::Json<serde_json::Value>,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<RedisPool>,
     new_version: EditVersion,
     session_queue: web::Data<AuthQueue>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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<RedisPool>,
     session_queue: web::Data<AuthQueue>,
     search_backend: web::Data<dyn SearchBackend>,
+    search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     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

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::<Vec<_>>()
+            .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(());

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<Mutex<Vec<SearchIndexOperation>>>,
+    kafka_client: actix_web::web::Data<KafkaClientState>,
+}
+
+impl IncrementalSearchQueue {
+    pub fn new(kafka_client: actix_web::web::Data<KafkaClientState>) -> 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,
+}

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<dyn SearchBackend>,
+) -> 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::<SearchProjectIndexQueueEvent>(
+            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,
+}

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<String>,
+    updated: DateTime<Utc>,
+    approved: DateTime<Utc>,
+    slug: Option<String>,
+    color: Option<i32>,
+    license: String,
+    components: exp::ProjectSerial,
+}
+
 fn normalize_for_search(s: &str) -> String {
     static SPECIAL_CHARS_RE: LazyLock<Regex> =
         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<String> {
+    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<UploadSearchProject>, i64)> {
     info!("Indexing local projects!");
 
-    // todo: loaders, project type, game versions
+    let searchable_statuses = searchable_statuses();
-    struct PartialProject {
-        id: DBProjectId,
-        name: String,
-        summary: String,
-        downloads: i32,
-        follows: i32,
-        icon_url: Option<String>,
-        updated: DateTime<Utc>,
-        approved: DateTime<Utc>,
-        slug: Option<String>,
-        color: Option<i32>,
-        license: String,
-        components: exp::ProjectSerial,
-    }
-
-    let searchable_statuses =
-        crate::models::projects::ProjectStatus::iterator()
-            .filter(|x| x.is_searchable())
-            .map(|x| x.to_string())
-            .collect::<Vec<String>>();
 
     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<exp::ProjectSerial>"
@@ -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::<Vec<i64>>();
+    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<Vec<UploadSearchProject>> {
+    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<exp::ProjectSerial>"
+		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::<Vec<PartialProject>>()
+    .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<PartialProject>,
+) -> eyre::Result<Vec<UploadSearchProject>> {
+    let searchable_statuses = searchable_statuses();
     let project_ids = db_projects.iter().map(|x| x.id.0).collect::<Vec<i64>>();
     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<DBProjectId, Vec<SearchProjectDependency>> =
@@ -573,7 +629,7 @@ pub async fn index_local(
         }
     }
 
-    Ok((uploads, *largest))
+    Ok(uploads)
 }
 
 struct PartialVersion {

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<dyn SearchBackend>,
+    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<Value>;

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)")
+                sqlx::query_scalar!("SELECT pg_advisory_unlock(1)")
-                    .execute(&main_pool)
+                    .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");

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)")
+    sqlx::raw_sql(&format!(
-        .bind(DUMMY_DATA_UPDATE)
+        "INSERT INTO dummy_data (update_id) VALUES ({DUMMY_DATA_UPDATE})"
-        .execute(pool)
+    ))
-        .await
+    .execute(pool)
-        .unwrap();
+    .await
+    .unwrap();
 
     DummyData::new(
         alpha_project,

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,
     )
 }

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<Self> {
+        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<T> {
+    pub event_type: &'static str,
+    pub event_metadata: EventMetadata,
+    #[serde(flatten)]
+    pub data: T,
+}
+
+impl<T> KafkaEvent<T> {
+    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<Utc>,
+    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"),
+};

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;

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")
+    sqlx::query!(
-        .bind(project_id)
+        "UPDATE mods SET status = 'processing' WHERE id = $1",
-        .execute(pool)
+        project_id
-        .await
+    )
-        .expect("failed to set project to processing status");
+    .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");

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<ApiV3>| 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<ApiV3>| 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')

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:

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"]