feat(search_processing_service): chats kafka - #5304
Conversation
📝 WalkthroughSummary by CodeRabbit
WalkthroughChat message metadata now includes owner and timestamp fields, and search queue chat payloads use a required message identifier with user and timestamp data. Direct chat search indexing and removal calls were removed from message storage, project deletion, and deleted-item polling paths. Chat purge events are published through the macro event broker. The search processing service now consumes chat lifecycle events, maps them to OpenSearch operations with retries, and removes purged chat messages during project reconciliation. Tests cover metadata, serialization, event mapping, malformed records, and publication failures. 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/macro_db_client/src/chat/get.rs (1)
279-296: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winRegenerate the SQLx cache for this changed query.
crates/macro_db_client/src/chat/get.rschanged thesqlx::query!SELECT list to includeowner_user_id,created_at, andupdated_at, but the.sqlxfiles are unchanged. Runnix develop --command just prepare_dbfrom the repository root; do not edit.sqlx/query-*.jsonmanually.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/macro_db_client/src/chat/get.rs` around lines 279 - 296, Regenerate the SQLx cache for the changed query in the get.rs query! invocation, including owner_user_id, created_at, and updated_at, by running nix develop --command just prepare_db from the repository root. Do not edit any .sqlx/query-*.json files manually.Sources: Coding guidelines, Path instructions
🧹 Nitpick comments (3)
services/search_processing_service/src/inbound/kafka_consumer/chat.rs (1)
139-141: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate retry-delay computation across chat.rs and project.rs. Both Kafka consumer handlers independently compute the same exponential backoff (
PROCESSING_RETRY_BASE_DELAY * 2u32.pow(attempt.saturating_sub(1))), risking drift if the backoff policy changes.
services/search_processing_service/src/inbound/kafka_consumer/chat.rs#L139-L141: extract this into a shared helper (e.g.retry_delay(attempt) -> Duration) in the parentkafka_consumermodule alongsideretry_processing/MAX_PROCESSING_ATTEMPTS.services/search_processing_service/src/inbound/kafka_consumer/project.rs#L181-L183: call the same shared helper instead of duplicating the formula.♻️ Example shared helper
// in kafka_consumer/mod.rs pub(super) fn retry_delay(attempt: u32) -> std::time::Duration { PROCESSING_RETRY_BASE_DELAY * 2u32.pow(attempt.saturating_sub(1)) }- let retry_delay = - PROCESSING_RETRY_BASE_DELAY * 2u32.pow(attempt.saturating_sub(1)); + let retry_delay = super::retry_delay(attempt);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/search_processing_service/src/inbound/kafka_consumer/chat.rs` around lines 139 - 141, The retry-delay formula is duplicated between the Kafka consumer handlers and should be centralized. In services/search_processing_service/src/inbound/kafka_consumer/chat.rs:139-141, add a shared retry_delay(attempt) helper in the parent kafka_consumer module near retry_processing and MAX_PROCESSING_ATTEMPTS, then use it from chat.rs. In services/search_processing_service/src/inbound/kafka_consumer/project.rs:181-183, replace the local exponential-backoff calculation with the same helper, preserving the existing Duration behavior.services/deleted_item_poller/src/handler.rs (1)
96-114: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated publish-and-await block; consider a shared helper.
Lines 96-111 are identical to
publish_project_purge_events(Lines 60-75) apart from the context strings. A small generic helper (publish_all(event_broker, &events, "chat purge")) would keep the retry/await semantics in one place.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/deleted_item_poller/src/handler.rs` around lines 96 - 114, Extract the duplicated concurrent publish-and-await logic from the current handler and publish_project_purge_events into a shared generic helper, such as publish_all, accepting the event broker, event collection, and purge context. Preserve the existing join_all behavior, error propagation, and retry/await semantics while allowing each caller to supply its context-specific messages.services/search_processing_service/src/inbound/kafka_consumer.rs (1)
61-61: 🩺 Stability & Availability | 🔵 TrivialNote the offset-reset behavior when adding
macro.chatsto an existing consumer group.
search-processing-servicealready has committed offsets for the other topics; the newly subscribed chat topic starts whereverauto.offset.resetpoints. If it'searliest, the first deploy replays the full chat topic retention (upserts are idempotent, so this is safe but potentially a large burst); iflatest, chat events published between the producer rollout and this deploy are lost. Worth confirming which is configured and, if needed, pre-seeding offsets.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/search_processing_service/src/inbound/kafka_consumer.rs` at line 61, Review the consumer configuration associated with the ChatMacroEvent subscription and confirm the configured auto.offset.reset behavior for the newly added macro.chats topic. Ensure deployment preserves the intended event coverage by pre-seeding the topic’s consumer-group offsets when latest would skip events, or explicitly accept and plan for the earliest replay burst.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/sqs_client/src/search/mod.rs`:
- Around line 78-79: The chat search-queue schema needs one-release
compatibility for already-enqueued payloads. In
crates/sqs_client/src/search/mod.rs lines 78-79, retain a deprecated
RemoveChatMessage enum arm and map it to a no-op or equivalent removal; in
crates/sqs_client/src/search/chat.rs lines 12-24, add serde defaults to the
newly required message_id, user_id, created_at, and updated_at fields so
pre-migration ChatMessage payloads continue deserializing.
In `@services/document_cognition_service/src/api/stream/util/chat_message/mod.rs`:
- Line 16: Update the tracing instrumentation on the surrounding chat-message
handler to prevent recording user-identifying data: exclude user_id from span
fields or replace it with a hashed/opaque identifier, and ensure the
MacroUserIdStr value containing the email is never emitted. Preserve recording
non-sensitive fields such as model where appropriate.
---
Outside diff comments:
In `@crates/macro_db_client/src/chat/get.rs`:
- Around line 279-296: Regenerate the SQLx cache for the changed query in the
get.rs query! invocation, including owner_user_id, created_at, and updated_at,
by running nix develop --command just prepare_db from the repository root. Do
not edit any .sqlx/query-*.json files manually.
---
Nitpick comments:
In `@services/deleted_item_poller/src/handler.rs`:
- Around line 96-114: Extract the duplicated concurrent publish-and-await logic
from the current handler and publish_project_purge_events into a shared generic
helper, such as publish_all, accepting the event broker, event collection, and
purge context. Preserve the existing join_all behavior, error propagation, and
retry/await semantics while allowing each caller to supply its context-specific
messages.
In `@services/search_processing_service/src/inbound/kafka_consumer.rs`:
- Line 61: Review the consumer configuration associated with the ChatMacroEvent
subscription and confirm the configured auto.offset.reset behavior for the newly
added macro.chats topic. Ensure deployment preserves the intended event coverage
by pre-seeding the topic’s consumer-group offsets when latest would skip events,
or explicitly accept and plan for the earliest replay burst.
In `@services/search_processing_service/src/inbound/kafka_consumer/chat.rs`:
- Around line 139-141: The retry-delay formula is duplicated between the Kafka
consumer handlers and should be centralized. In
services/search_processing_service/src/inbound/kafka_consumer/chat.rs:139-141,
add a shared retry_delay(attempt) helper in the parent kafka_consumer module
near retry_processing and MAX_PROCESSING_ATTEMPTS, then use it from chat.rs. In
services/search_processing_service/src/inbound/kafka_consumer/project.rs:181-183,
replace the local exponential-backoff calculation with the same helper,
preserving the existing Duration behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 63f3dcee-9710-4953-b841-edcf71a7910c
⛔ Files ignored due to path filters (2)
.sqlx/query-eb360ba6aad69fe2c501baa8b61b4243051bd3bde6e517014cd5768334fd7757.jsonis excluded by!**/.sqlx/**Cargo.lockis excluded by!**/*.lock,!**/Cargo.lock
📒 Files selected for processing (28)
.github/workspace-dep-closures.jsoncrates/macro_db_client/fixtures/chat_message_info.sqlcrates/macro_db_client/src/chat/get.rscrates/macro_db_client/src/chat/get/test.rscrates/projects/src/domain/ports.rscrates/projects/src/domain/service.rscrates/projects/src/domain/service/tests.rscrates/projects/src/outbound/sqs_search_indexer.rscrates/sqs_client/src/search/chat.rscrates/sqs_client/src/search/chat/test.rscrates/sqs_client/src/search/mod.rsservices/deleted_item_poller/Cargo.tomlservices/deleted_item_poller/src/config.rsservices/deleted_item_poller/src/handler.rsservices/deleted_item_poller/src/handler/test.rsservices/document_cognition_service/Cargo.tomlservices/document_cognition_service/src/api/stream/chat_message/mod.rsservices/document_cognition_service/src/api/stream/util/chat_message/mod.rsservices/document_cognition_service/src/api/utils/mod.rsservices/document_cognition_service/src/api/utils/search.rsservices/document_cognition_service/src/main.rsservices/search_processing_service/Cargo.tomlservices/search_processing_service/src/inbound/kafka_consumer.rsservices/search_processing_service/src/inbound/kafka_consumer/chat.rsservices/search_processing_service/src/inbound/kafka_consumer/project.rsservices/search_processing_service/src/inbound/kafka_consumer/test.rsservices/search_processing_service/src/process/chat.rsservices/search_processing_service/src/process/mod.rs
💤 Files with no reviewable changes (4)
- services/document_cognition_service/Cargo.toml
- services/document_cognition_service/src/api/utils/search.rs
- services/document_cognition_service/src/api/utils/mod.rs
- services/document_cognition_service/src/main.rs
1 participant
Wires up SPS to use chat kafka topic for chats instead of sqs