[PB-6473]: feat/mail blob pointer table#74
Open
jzunigax2 wants to merge 7 commits into
Open
Conversation
jzunigax2
force-pushed
the
feat/stalwart-events-webhook
branch
from
96cf7aa to
2a9c787
Compare
- Added `findBucketContextByProviderInternalId` method to `AddressRepository` for resolving user and bucket context. - Updated `AccountService` to delete provider links during account provisioning failure. - Implemented `createBucketEntry` method in `BridgeClient` for creating bucket entries in response to Stalwart events. - Enhanced `StalwartEventsService` to handle batch events and create bucket entries based on resolved account context. - Added unit tests for the new functionality in `StalwartEventsService` and `AddressRepository` to ensure proper behavior.
- Updated `deleteEmail` method in `EmailService` to handle quota entry release upon email deletion. - Introduced `releaseQuotaEntry` method to manage quota entries based on email deletion results. - Modified `deleteEmail` in `MailProvider` to return a result object containing the deleted entry key. - Enhanced unit tests for `EmailService` and `JmapMailProvider` to cover new deletion behavior and quota management scenarios. - Integrated `BridgeClient` for handling quota entry deletions in the bridge service.
- Introduced `decodeStalwartIdBig` function to handle Stalwart IDs exceeding JavaScript's safe integer range, allowing for proper decoding of email IDs. - Updated `decodeStalwartId` to utilize the new decoding function and throw an error for IDs exceeding safe integer limits. - Enhanced `StalwartEventsAuthGuard` to throw an `UnauthorizedException` for malformed authorization headers lacking a colon separator. - Added unit tests for the new decoding function and authentication guard to ensure robust error handling and functionality.
- Added migration for creating the `mail_bucket_entries` table to store email usage data. - Introduced `MailUsageModule` and `MailUsageService` for managing email bucket entries and tracking usage. - Updated `AddressRepository`, `EmailService`, and `StalwartEventsService` to integrate with the new mail usage functionality. - Refactored `BridgeClient` methods to accommodate changes in bucket entry management. - Enhanced models and repositories to support the new mail bucket entry structure.
- Modified `AddressRepository` to include `mailAccountId` in the returned results and updated account structure. - Refactored `EmailService` to replace `BridgeClient` with `MailUsageService` for managing email quota entries. - Enhanced unit tests across `AddressRepository`, `EmailService`, and `StalwartEventsService` to reflect changes in account ID handling and usage tracking. - Introduced new tests for `MailUsageService` and `MailBucketEntryRepository` to ensure proper functionality and error handling.
…nd repositories - Updated migration to change `mail_account_id` to `mail_address_id` in the `mail_bucket_entries` table. - Refactored `AddressRepository`, `EmailService`, and `StalwartEventsService` to use `mailAddressId` instead of `mailAccountId`. - Adjusted related models, repositories, and unit tests to reflect the new naming convention for better clarity and consistency. - Enhanced tests in `MailUsageService` and `MailBucketEntryRepository` to ensure proper functionality with the updated identifiers.
jzunigax2
force-pushed
the
feat/mail-blob-pointer-table
branch
from
efddac8 to
6692bb4
Compare
📝 WalkthroughWalkthroughAdds persisted mail bucket entries and a ChangesMail usage tracking
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant StalwartEventsService
participant MailUsageService
participant BridgeClient
participant MailBucketEntryRepository
StalwartEventsService->>MailUsageService: trackStoredMessage(params)
MailUsageService->>MailBucketEntryRepository: findByEntryKey(entryKey)
MailUsageService->>BridgeClient: createBucketEntry(userUuid, bucketId, size)
MailUsageService->>MailBucketEntryRepository: create(mailAddressId, entryKey, bridgeEntryId, size)
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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: 3
🧹 Nitpick comments (1)
src/modules/email/email.service.ts (1)
373-377: 🗄️ Data Integrity & Integration | 🔵 TrivialProvide a durable retry for failed quota release.
releaseStoredMessageruns after permanent deletion, while the surrounding catch suppresses failures. If usage release fails, the message is gone and no later deletion retries the quota update. Confirm that an outbox, reconciliation job, or equivalent retry path exists; otherwise quota can remain overstated.🤖 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 `@src/modules/email/email.service.ts` around lines 373 - 377, Ensure the permanent-deletion flow around releaseStoredMessage has a durable retry mechanism for failed quota release, such as an existing outbox or reconciliation job. If no equivalent path exists, persist the release operation for later retry instead of allowing the surrounding suppressed failure to lose the quota update, while preserving the current deletion behavior.
🤖 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 `@src/modules/infrastructure/jmap/jmap-mail.provider.ts`:
- Around line 845-847: The entry-key construction must preserve complete
Stalwart IDs instead of truncating them with the 32-bit masks in the visible
account/email ID conversion logic. Update the identifiers used by the returned
key to retain full decoded values (or the original ID strings), ensuring IDs
differing by 2^32 produce distinct keys, and add a regression test covering that
case.
In `@src/modules/usage/mail-usage.service.ts`:
- Around line 43-60: Update the persistence error handling around entries.create
in the mail usage flow to roll back the minted bridge bucket entry for every
failure, while preserving the existing duplicate-entry debug behavior. Guard
bridge.deleteBucketEntry so rollback failures are caught and logged with enough
context for reconciliation, then rethrow the original persistence error. Update
mail-usage.service.spec.ts to expect rollback for generic persistence failures
and verify rollback errors are handled.
- Around line 68-93: The release flow in releaseStoredMessage must be idempotent
when BridgeClient.deleteBucketEntry reports a missing bridge entry. Handle the
bridge client's 404/not-found error as successful and continue deleting the
local entry, while still propagating other bridge deletion failures; preserve
the existing missing-local-entry early return and success logging.
---
Nitpick comments:
In `@src/modules/email/email.service.ts`:
- Around line 373-377: Ensure the permanent-deletion flow around
releaseStoredMessage has a durable retry mechanism for failed quota release,
such as an existing outbox or reconciliation job. If no equivalent path exists,
persist the release operation for later retry instead of allowing the
surrounding suppressed failure to lose the quota update, while preserving the
current deletion 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 54d91ed6-bec9-4e5b-bed7-3966d50ae1c2
📒 Files selected for processing (19)
migrations/20260623120000-create-mail-bucket-entries.jssrc/modules/account/repositories/address.repository.spec.tssrc/modules/account/repositories/address.repository.tssrc/modules/email/email.module.tssrc/modules/email/email.service.spec.tssrc/modules/email/email.service.tssrc/modules/infrastructure/bridge/bridge.service.spec.tssrc/modules/infrastructure/bridge/bridge.service.tssrc/modules/infrastructure/jmap/jmap-mail.provider.tssrc/modules/stalwart-events/stalwart-events.module.tssrc/modules/stalwart-events/stalwart-events.service.spec.tssrc/modules/stalwart-events/stalwart-events.service.tssrc/modules/usage/domain/mail-bucket-entry.domain.tssrc/modules/usage/mail-usage.module.tssrc/modules/usage/mail-usage.service.spec.tssrc/modules/usage/mail-usage.service.tssrc/modules/usage/models/mail-bucket-entry.model.tssrc/modules/usage/repositories/mail-bucket-entry.repository.spec.tssrc/modules/usage/repositories/mail-bucket-entry.repository.ts
Comment on lines
+845
to
847
| const numericAccountId = decodeStalwartIdBig(accountId) & 0xffffffffn; | ||
| const documentId = decodeStalwartIdBig(emailId) & 0xffffffffn; | ||
| return `${numericAccountId}:${documentId}`; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
node <<'NODE'
const mask = (value) => value & 0xffffffffn;
const a = 1n;
const b = (1n << 32n) + 1n;
if (mask(a) !== mask(b)) process.exit(1);
console.log(`${mask(a)} === ${mask(b)}`);
NODERepository: internxt/mail-server
Length of output: 166
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '\n## provider outline\n'
ast-grep outline src/modules/infrastructure/jmap/jmap-mail.provider.ts --view expanded | sed -n '1,220p'
printf '\n## usage service outline\n'
ast-grep outline src/modules/usage/mail-usage.service.ts --view expanded | sed -n '1,240p'
printf '\n## entryKey references\n'
rg -n "entryKey|decodeStalwartIdBig|0xffffffffn|release lookup|duplicate" src/modules -SRepository: internxt/mail-server
Length of output: 10366
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '\n## provider slice around target lines\n'
sed -n '820,860p' src/modules/infrastructure/jmap/jmap-mail.provider.ts | cat -n
printf '\n## usage service slice around entryKey usage\n'
sed -n '1,260p' src/modules/usage/mail-usage.service.ts | cat -nRepository: internxt/mail-server
Length of output: 4734
Do not truncate Stalwart IDs to 32 bits. & 0xffffffffn makes distinct large IDs collide in entryKey, which this path uses for duplicate detection and release. That can skip tracking or release the wrong stored message. Keep the full decoded IDs, or use the original ID strings, and add a regression test for values that differ by 2^32.
🤖 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 `@src/modules/infrastructure/jmap/jmap-mail.provider.ts` around lines 845 -
847, The entry-key construction must preserve complete Stalwart IDs instead of
truncating them with the 32-bit masks in the visible account/email ID conversion
logic. Update the identifiers used by the returned key to retain full decoded
values (or the original ID strings), ensuring IDs differing by 2^32 produce
distinct keys, and add a regression test covering that case.
Comment on lines
+43
to
+60
| try { | ||
| await this.entries.create({ | ||
| mailAddressId, | ||
| entryKey, | ||
| bridgeEntryId, | ||
| size, | ||
| }); | ||
| } catch (error) { | ||
| if (error instanceof DuplicateEntryKeyError) { | ||
| this.logger.debug( | ||
| { entryKey, bridgeEntryId }, | ||
| 'Concurrent tracking detected; rolling back minted bucket entry', | ||
| ); | ||
| await this.bridge.deleteBucketEntry(userUuid, bucketId, bridgeEntryId); | ||
| return; | ||
| } | ||
| throw error; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
Missing rollback on non-duplicate persistence failures leaks minted bridge quota.
Only DuplicateEntryKeyError triggers a rollback of the just-minted bridge bucket entry (line 56). Any other failure from entries.create (DB timeout, connection drop, etc.) is rethrown at line 59 with the bridge entry left dangling — and since no row was persisted, a retried ingest call will mint a second bridge entry via findByEntryKey returning null again, compounding the leak on every retry. Additionally, the rollback call itself (line 56) is unguarded: if bridge.deleteBucketEntry throws there, that failure propagates unhandled and the orphaned entry is never logged for manual reconciliation.
mail-usage.service.spec.ts (Lines 101-114) currently asserts this "no rollback on generic errors" behavior, so that test will need updating alongside this fix.
🛡️ Proposed fix: roll back on any persistence failure and guard the rollback itself
try {
await this.entries.create({
mailAddressId,
entryKey,
bridgeEntryId,
size,
});
} catch (error) {
if (error instanceof DuplicateEntryKeyError) {
this.logger.debug(
{ entryKey, bridgeEntryId },
'Concurrent tracking detected; rolling back minted bucket entry',
);
- await this.bridge.deleteBucketEntry(userUuid, bucketId, bridgeEntryId);
- return;
+ await this.bridge
+ .deleteBucketEntry(userUuid, bucketId, bridgeEntryId)
+ .catch((rollbackError) =>
+ this.logger.error(
+ { entryKey, bridgeEntryId, rollbackError },
+ 'Failed to roll back minted bucket entry after duplicate-key race',
+ ),
+ );
+ return;
}
- throw error;
+ this.logger.error(
+ { entryKey, bridgeEntryId, error },
+ 'Failed to persist tracked message; rolling back minted bucket entry',
+ );
+ await this.bridge
+ .deleteBucketEntry(userUuid, bucketId, bridgeEntryId)
+ .catch((rollbackError) =>
+ this.logger.error(
+ { entryKey, bridgeEntryId, rollbackError },
+ 'Failed to roll back minted bucket entry after persistence failure',
+ ),
+ );
+ throw error;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| try { | |
| await this.entries.create({ | |
| mailAddressId, | |
| entryKey, | |
| bridgeEntryId, | |
| size, | |
| }); | |
| } catch (error) { | |
| if (error instanceof DuplicateEntryKeyError) { | |
| this.logger.debug( | |
| { entryKey, bridgeEntryId }, | |
| 'Concurrent tracking detected; rolling back minted bucket entry', | |
| ); | |
| await this.bridge.deleteBucketEntry(userUuid, bucketId, bridgeEntryId); | |
| return; | |
| } | |
| throw error; | |
| } | |
| try { | |
| await this.entries.create({ | |
| mailAddressId, | |
| entryKey, | |
| bridgeEntryId, | |
| size, | |
| }); | |
| } catch (error) { | |
| if (error instanceof DuplicateEntryKeyError) { | |
| this.logger.debug( | |
| { entryKey, bridgeEntryId }, | |
| 'Concurrent tracking detected; rolling back minted bucket entry', | |
| ); | |
| await this.bridge | |
| .deleteBucketEntry(userUuid, bucketId, bridgeEntryId) | |
| .catch((rollbackError) => | |
| this.logger.error( | |
| { entryKey, bridgeEntryId, rollbackError }, | |
| 'Failed to roll back minted bucket entry after duplicate-key race', | |
| ), | |
| ); | |
| return; | |
| } | |
| this.logger.error( | |
| { entryKey, bridgeEntryId, error }, | |
| 'Failed to persist tracked message; rolling back minted bucket entry', | |
| ); | |
| await this.bridge | |
| .deleteBucketEntry(userUuid, bucketId, bridgeEntryId) | |
| .catch((rollbackError) => | |
| this.logger.error( | |
| { entryKey, bridgeEntryId, rollbackError }, | |
| 'Failed to roll back minted bucket entry after persistence failure', | |
| ), | |
| ); | |
| throw error; | |
| } |
🤖 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 `@src/modules/usage/mail-usage.service.ts` around lines 43 - 60, Update the
persistence error handling around entries.create in the mail usage flow to roll
back the minted bridge bucket entry for every failure, while preserving the
existing duplicate-entry debug behavior. Guard bridge.deleteBucketEntry so
rollback failures are caught and logged with enough context for reconciliation,
then rethrow the original persistence error. Update mail-usage.service.spec.ts
to expect rollback for generic persistence failures and verify rollback errors
are handled.
There was a problem hiding this comment.
@xabg2 Sounds good, I'll leave this open for @jzunigax2's input. For a quick recap of the concern: currently only DuplicateEntryKeyError triggers rollback of the minted bridge bucket entry — any other error from entries.create (DB timeout, connection drop, etc.) leaves the bridge entry orphaned, and retries will mint additional entries since findByEntryKey still returns null. Happy to help implement the guarded rollback (or a narrower version of it, e.g. just logging orphaned entries for manual reconciliation without rethrowing changes) once you've settled on the desired behavior.
Comment on lines
+68
to
+93
| async releaseStoredMessage( | ||
| params: ReleaseStoredMessageParams, | ||
| ): Promise<void> { | ||
| const { userUuid, bucketId, entryKey } = params; | ||
|
|
||
| const existing = await this.entries.findByEntryKey(entryKey); | ||
| if (!existing) { | ||
| this.logger.debug( | ||
| { entryKey }, | ||
| 'No tracked entry for message; skipping release', | ||
| ); | ||
| return; | ||
| } | ||
|
|
||
| await this.bridge.deleteBucketEntry( | ||
| userUuid, | ||
| bucketId, | ||
| existing.bridgeEntryId, | ||
| ); | ||
| await this.entries.deleteByEntryKey(entryKey); | ||
|
|
||
| this.logger.log( | ||
| { entryKey, bridgeEntryId: existing.bridgeEntryId }, | ||
| 'Released stored message', | ||
| ); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
ast-grep run --pattern 'async deleteBucketEntry($$$) { $$$ }' --lang typescript src/modules/infrastructure/bridge/bridge.service.tsRepository: internxt/mail-server
Length of output: 486
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== mail-usage.service.ts =="
cat -n src/modules/usage/mail-usage.service.ts | sed -n '1,170p'
echo
echo "== bridge.service.ts outline =="
ast-grep outline src/modules/infrastructure/bridge/bridge.service.ts --view expanded || true
echo
echo "== deleteBucketEntry references =="
rg -n "deleteBucketEntry|findByEntryKey|deleteByEntryKey" src/modules -S
echo
echo "== bridge.service.ts relevant slice =="
cat -n src/modules/infrastructure/bridge/bridge.service.ts | sed -n '1,260p'Repository: internxt/mail-server
Length of output: 14358
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== bridge.service.spec.ts slice =="
cat -n src/modules/infrastructure/bridge/bridge.service.spec.ts | sed -n '180,260p'
echo
echo "== mail-usage.service.spec.ts slice =="
cat -n src/modules/usage/mail-usage.service.spec.ts | sed -n '1,240p'Repository: internxt/mail-server
Length of output: 8914
Treat missing bridge entries as a successful release.
BridgeClient.deleteBucketEntry() throws on 404/non-200, so if the bridge delete succeeds but entries.deleteByEntryKey() fails, a retry will hit the same already-removed bridgeEntryId and fail again. Make the release path idempotent by ignoring “not found” from the bridge delete or by deleting the local row first.
🤖 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 `@src/modules/usage/mail-usage.service.ts` around lines 68 - 93, The release
flow in releaseStoredMessage must be idempotent when
BridgeClient.deleteBucketEntry reports a missing bridge entry. Handle the bridge
client's 404/not-found error as successful and continue deleting the local
entry, while still propagating other bridge deletion failures; preserve the
existing missing-local-entry early return and success logging.
xabg2
reviewed
Jul 14, 2026
| const { statusCode, body } = await this.httpClient.request({ | ||
| method: 'DELETE', | ||
| path: `${this.basePath}/v2/gateway/users/${encodeURIComponent(userUuid)}/buckets/${encodeURIComponent(bucketId)}/entries/${encodeURIComponent(key)}`, | ||
| path: `${this.basePath}/v2/gateway/users/${encodeURIComponent(userUuid)}/buckets/${encodeURIComponent(bucketId)}/entries/${encodeURIComponent(entryId)}`, |
Contributor
There was a problem hiding this comment.
What about extracting the encodeURIComponent(userUuid) / encodeURIComponent(bucketId) / encodeURIComponent(entryId) to a constant?
Something like encodedUserUuid = ... and so on.
|
|
||
| describe('handleBatch', () => { | ||
| it('when an ingest event resolves to a bucket, then creates a bucket entry keyed by accountId:documentId', async () => { | ||
| it('when an ingest event resolves to a bucket, then tracks the stored message keyed by accountId:documentId', async () => { |
| }); | ||
|
|
||
| describe('trackStoredMessage', () => { | ||
| it('when the entry is new, then mints a bridge entry and persists the pointer', async () => { |
Comment on lines
+43
to
+60
| try { | ||
| await this.entries.create({ | ||
| mailAddressId, | ||
| entryKey, | ||
| bridgeEntryId, | ||
| size, | ||
| }); | ||
| } catch (error) { | ||
| if (error instanceof DuplicateEntryKeyError) { | ||
| this.logger.debug( | ||
| { entryKey, bridgeEntryId }, | ||
| 'Concurrent tracking detected; rolling back minted bucket entry', | ||
| ); | ||
| await this.bridge.deleteBucketEntry(userUuid, bucketId, bridgeEntryId); | ||
| return; | ||
| } | ||
| throw error; | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Keep track of mail blobs using the new bucket entries table, on ingest event firing we store a pointer to the related network bucket entry so we can later on a delete event decrement size
Summary by CodeRabbit
New Features
Bug Fixes