feat(activity): activity_events table + consumer - #5495
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe pull request adds a new 🚥 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 |
5592c93 to
83bd0db
Compare
One append-only table of activities: a principal did something to an entity at a time. Activity ids are uuidv5(source event id, ordinal) so at-least-once redelivery is absorbed by ON CONFLICT (id) DO NOTHING; one broker event may yield several activities (participant adds, call started). crates/activity (hexagonal, per soup_realtime) consumes macro.documents, .channels, .chats, .projects, .email, .properties, and .calls under group activity-materializer, mapping attributed mutations to activities: created / edited / deleted / messaged / sent / property_changed / participant_added / participant_removed / call_started. Unattributable mutations (no actor) are dropped; provider-initiated email changes are dropped; assistant chat messages are dropped (only the user's own prompts are their activity). Hard deletes purge the entity's activities, including project purge cascades. Delegation resolves at ingestion: subject = on_behalf_of ?? actor (channel messages carry triggered_by for agent-sent messages). The consumer runs in shadow inside document_storage_service — activities accumulate, nothing reads them yet.
83bd0db to
e1b8e61
Compare
PropertyChange / ParticipantChange / CallStart are defined once and shared: the write codec serializes them and the future read codec deserializes the same definitions, instead of hand-mirroring field shapes in json! blocks. CommonAction::PropertyChanged reuses the same payload struct.
The tag strings were hand-typed duplicates of the variant names. strum's IntoStaticStr (snake_case) derives them, matching the model-entity convention. Renaming a variant now silently renames the stored tag — the pinned full-vocabulary codec test is what makes that a loud failure instead of a quiet storage migration.
![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
🧹 Nitpick comments (3)
crates/projects/src/domain/activity/test.rs (1)
52-74: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover the chat branch of the purge cascade.
The test sets
purged_chat_ids: vec![]. The mapper mapspurged_chat_idstoEntityType::Chatentries atcrates/projects/src/domain/activity.rslines 87-91. That branch stays unexercised, so the test name "the whole cascade" overstates the coverage. A regression that drops the chat entries would still pass.♻️ Proposed fixture and assertion update
purged_project_ids: vec!["proj-2".to_string()], purged_document_ids: vec!["doc-1".to_string()], - purged_chat_ids: vec![], + purged_chat_ids: vec!["chat-1".to_string()], }, )); assert_eq!( event.event.ingest(event.event_id), Ingest::Purge(vec![ (EntityType::Project, "proj-1".to_string()), (EntityType::Project, "proj-2".to_string()), (EntityType::Document, "doc-1".to_string()), + (EntityType::Chat, "chat-1".to_string()), ]) );🤖 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/projects/src/domain/activity/test.rs` around lines 52 - 74, Update permanent_delete_purges_the_whole_cascade to include at least one chat ID in purged_chat_ids and add the corresponding (EntityType::Chat, chat ID) entry to the expected Ingest::Purge list, covering the chat branch alongside the existing project and document entries.crates/projects/src/domain/activity.rs (1)
63-71: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueConsider a distinct action for
Restored.The
Restoredarm maps toCommonAction::Edited. A restore and an edit then become indistinguishable in the durable table. The activity table is append-only, so a later split of the two actions requires a backfill of existing rows. If the shared vocabulary should stay small for now, record the decision in the comment above this arm.🤖 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/projects/src/domain/activity.rs` around lines 63 - 71, Update the ProjectTopicEvent::Restored arm to use a distinct CommonAction variant for restore events instead of CommonAction::Edited, ensuring restored and edited activities remain distinguishable in the append-only activity table; if no distinct variant is currently appropriate, document that decision in a comment directly above the arm.crates/chat/src/domain/activity/test.rs (1)
40-53: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAssert the activity count before indexing.
Lines 43 and 53 index
activities[0]directly. Thelet ... elseonly proves theInsertvariant, not a non-empty vector. If the mapper returns an empty vector, the test fails with an index panic instead of a clear assertion message.crates/projects/src/domain/activity/test.rsline 33 already assertsactivities.len().♻️ Proposed test tightening
let Ingest::Insert(activities) = created.event.ingest(created.event_id) else { panic!("expected activities"); }; + assert_eq!(activities.len(), 1); assert_eq!(activities[0].action, Action::Created);🤖 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/chat/src/domain/activity/test.rs` around lines 40 - 53, Update the activity assertions in the test around the created and sent events to assert activities.len() before indexing activities[0], matching the established pattern in the projects activity test. Preserve the existing action and entity_type assertions after confirming the expected activity count.
🤖 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/call/src/domain/activity.rs`:
- Around line 53-83: Update the CallTopicEvent::RecordDeleted handling to purge
the related channel activity using the deletion key for the call, in addition to
purging the Call entity activity. Limit the channel purge to activities
associated with that specific call and do not purge the entire channel; reuse
the existing deletion-key mechanism and Ingest::Purge flow.
In `@crates/email/src/domain/activity.rs`:
- Around line 78-91: Update the EmailTopicEvent::MessageSent arm to emit the
activity only when both m.actor is present and m.origin is
EmailEventOrigin::UserAction; return Ingest::Ignore for provider-synchronized
sends even when an actor exists. Add coverage for a ProviderSync MessageSent
carrying an actor.
---
Nitpick comments:
In `@crates/chat/src/domain/activity/test.rs`:
- Around line 40-53: Update the activity assertions in the test around the
created and sent events to assert activities.len() before indexing
activities[0], matching the established pattern in the projects activity test.
Preserve the existing action and entity_type assertions after confirming the
expected activity count.
In `@crates/projects/src/domain/activity.rs`:
- Around line 63-71: Update the ProjectTopicEvent::Restored arm to use a
distinct CommonAction variant for restore events instead of
CommonAction::Edited, ensuring restored and edited activities remain
distinguishable in the append-only activity table; if no distinct variant is
currently appropriate, document that decision in a comment directly above the
arm.
In `@crates/projects/src/domain/activity/test.rs`:
- Around line 52-74: Update permanent_delete_purges_the_whole_cascade to include
at least one chat ID in purged_chat_ids and add the corresponding
(EntityType::Chat, chat ID) entry to the expected Ingest::Purge list, covering
the chat branch alongside the existing project and document entries.
🪄 Autofix
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: 227db723-286b-410e-81b4-6b484b815699
⛔ Files ignored due to path filters (3)
.sqlx/query-29b8ca0e672d412aeeaf19c3f633fb10fb5b41ac6a0ff074eefcc71123353872.jsonis excluded by!**/.sqlx/**.sqlx/query-6d61d631a10b803cb269fa6992ef84adbf8b0ad0940ab61b8a649f74584d8b0e.jsonis excluded by!**/.sqlx/**Cargo.lockis excluded by!**/*.lock,!**/Cargo.lock
📒 Files selected for processing (46)
.github/workspace-dep-closures.jsonCargo.tomlcrates/activity/Cargo.tomlcrates/activity/src/domain/mod.rscrates/activity/src/domain/models.rscrates/activity/src/domain/models/test.rscrates/activity/src/domain/ports.rscrates/activity/src/inbound/kafka_consumer.rscrates/activity/src/inbound/mod.rscrates/activity/src/lib.rscrates/activity/src/outbound/mod.rscrates/activity/src/outbound/pg_activity_repo.rscrates/activity/src/outbound/pg_activity_repo/test.rscrates/call/Cargo.tomlcrates/call/src/domain/activity.rscrates/call/src/domain/activity/test.rscrates/call/src/domain/mod.rscrates/channels/Cargo.tomlcrates/channels/src/domain/activity.rscrates/channels/src/domain/activity/test.rscrates/channels/src/domain/mod.rscrates/chat/Cargo.tomlcrates/chat/src/domain.rscrates/chat/src/domain/activity.rscrates/chat/src/domain/activity/test.rscrates/documents/Cargo.tomlcrates/documents/src/domain.rscrates/documents/src/domain/activity.rscrates/documents/src/domain/activity/test.rscrates/email/Cargo.tomlcrates/email/src/domain.rscrates/email/src/domain/activity.rscrates/email/src/domain/activity/test.rscrates/macro_db_client/migrations/20260805180315_create_activity_events.sqlcrates/projects/Cargo.tomlcrates/projects/src/domain.rscrates/projects/src/domain/activity.rscrates/projects/src/domain/activity/test.rscrates/properties/Cargo.tomlcrates/properties/src/domain.rscrates/properties/src/domain/activity.rscrates/properties/src/domain/activity/test.rsservices/document_storage_service/Cargo.tomlservices/document_storage_service/src/main.rsservices/document_storage_service/src/service/activity.rsservices/document_storage_service/src/service/mod.rs
| // A storage failure must abort the run: continuing and | ||
| // committing a later record on this partition would | ||
| // cumulatively commit past the failed one, losing it | ||
| // forever. Returning Err restarts the consumer from the |
There was a problem hiding this comment.
Instead of doing this we could have a activity-dead-letter-queue topic that we publish messages to that failed to run. This is a better pattern than having the worker kill itself if it fails as it will cause a massive backup of messages if there is a temporary outage or 1-off issue
| -- uuidv5(source event id, ordinal): one broker event may yield several | ||
| -- facts; replays re-derive the same ids, making inserts idempotent. |
| entity_type TEXT NOT NULL, | ||
| entity_id TEXT NOT NULL, |
There was a problem hiding this comment.
Should we have triggers on the db to automatically delete these rows if the underlying entity is deleted?
I could see us using activity_events as an audit trail table though so we probably wouldn't want deletion of these ever
| payloads.as_slice() as &[Option<serde_json::Value>], | ||
| &entity_types, | ||
| &entity_ids, | ||
| &occurred_ats, |
There was a problem hiding this comment.
we may potentially want to batch and parallelize this call if [Activity] is large enough
There was a problem hiding this comment.
I think we can leave this for the future
| -- One append-only table of activity facts: a principal did something to an | ||
| -- entity at a time. Every activity surface is a query over this table; no | ||
| -- derived tables. See the activity crate for the fact vocabulary. | ||
| CREATE TABLE activity_events ( |
There was a problem hiding this comment.
You need an index on entity_id + entity_type since you have a delete call that uses that index. This will currently result in a full table scan
There was a problem hiding this comment.
It's there — idx_activity_events_entity ON (entity_type, entity_id, occurred_at DESC, id DESC) at the bottom of this migration. The purge's WHERE (entity_type, entity_id) IN (...) uses its leading two columns, so no full scan. (The trailing occurred_at, id columns are there so the same index serves the future entity-timeline keyset reads.)
…l serde_json - email: MessageSent now requires origin = user_action as well as an actor (CodeRabbit) — a send synced from another client no longer records Sent; test added for the provider_sync-with-actor case. - properties: serde_json becomes an unconditional dep — the ungated domain::activity module uses it, so minimal-feature consumers broke. - projects test: purge-cascade test now covers the chat branch (CodeRabbit nitpick). - call: document the deliberate keep of the channel call_started activity on RecordDeleted — channel history, dangling payload ref tolerated by readers.
2 participants
Uh oh!
There was an error while loading. Please reload this page.