Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
57 commits
Select commit Hold shift + click to select a range
28f29b7
feat(cdp): deliver sandbox email to members' stored addresses
Silthus Nov 4, 2026
288a9e0
Merge branch 'workflows-sandbox-split/member-cache' into workflows-sa…
Silthus Nov 4, 2026
21e142c
Merge branch 'workflows-sandbox-split/member-cache' into workflows-sa…
Silthus Nov 4, 2026
bbe93cc
Merge branch 'workflows-sandbox-split/member-cache' into workflows-sa…
Silthus Nov 4, 2026
3de8380
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
583c9a0
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
b2e033d
fix(cdp): map each sandbox recipient list to stored addresses by address
Silthus Nov 4, 2026
fecd931
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
03ff7a6
fix(cdp): block a sandbox recipient that matches no stored case exactly
Silthus Nov 4, 2026
47e543d
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
38df23e
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
f87c82d
chore(cdp): let names carry the stored-address rules instead of comments
Silthus Nov 4, 2026
7889056
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
e9df423
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
c2f9eeb
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
1cf76e7
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
c23bdea
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
122e919
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
908602f
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
c026479
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
2aaf2eb
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
0acee96
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
78464d0
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
9cd7c10
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
e7ce517
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
2284461
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
60984d6
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
23b63c8
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
e8ace11
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 4, 2026
0377249
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
47f7910
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
1967ef9
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
6ca50ee
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
0ee3154
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
47fab12
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
a9ca33e
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
47e749e
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 5, 2026
91d9a03
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
ffd2177
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
2bf004d
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
c8cab96
chore(cdp): apply prettier formatting
Silthus Nov 6, 2026
c3a7a79
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
4e0011a
fix(cdp): deliver to a member address stored twice after trimming
Silthus Nov 6, 2026
4009d6e
chore(cdp): return the member check result directly
Silthus Nov 6, 2026
86bed64
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
55b357b
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
7d9d0d4
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
1783da7
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
6cc3c56
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
088fe47
Merge branch 'fr/member-cache' into fr/stored-member-address
Silthus Nov 6, 2026
6c013e5
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
5fe4fef
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
7c2d6b8
chore(cdp): check the shared member reload through checkRecipients
Silthus Nov 6, 2026
32dd6ba
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
8e6c8ad
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 6, 2026
d6014a2
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 9, 2026
862aed8
Merge workflows-sandbox-split/member-cache into workflows-sandbox-spl…
Silthus Nov 9, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/internal/workflows-email-sending-tiers.md
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,8 @@ The worker sends through a `sandbox` email integration only when `WORKFLOWS_SAND
Otherwise every send through it skips with a reason in the run log and a `workflows sandbox email blocked` event with reason `switch_off`.
It sends workflow email steps and test sends only: a broadcast, a transactional API email, or a destination send through it skips with a reason in the run log and a blocked event with reason `not_workflow_step`.
Every To, CC, and BCC address must belong to an active member of the team's active organization with a verified email address.
The check ignores case, and the worker delivers to the address as the member stored it, not as the step wrote it.
When two members stored the address in different cases, the step must write one of them exactly, or the send skips.
The worker checks recipients against a membership snapshot at most one minute old when it prepares the send, reloading a snapshot older than 5 seconds before it blocks on it, and reads membership fresh right before delivery.
A membership check that fails or takes longer than 3 seconds skips the send and lists the blocked addresses in the run log.
The blocked event fires at most once an hour per team, reason, and send kind (workflow or test); `cdp_sandbox_email_blocked_total` counts every blocked send by reason.
Expand Down
108 changes: 68 additions & 40 deletions nodejs/src/cdp/services/managers/organization-members.service.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -102,10 +102,12 @@ describe('OrganizationMembersService', () => {
await change(userId)
const address = ` ${memberEmail.toUpperCase()} `

const blocked = await service.getBlockedRecipients(team.id, [address], { fresh: false })
const check = await service.checkRecipients(team.id, [address], { fresh: false })

expect(blocked).toEqual(
expected === 'allowed' ? [] : [{ address, unverifiedMember: expected === 'unverified' }]
expect(check).toEqual(
expected === 'allowed'
? { blocked: [], deliverTo: new Map([[address, memberEmail]]) }
: { blocked: [{ address, unverifiedMember: expected === 'unverified' }], deliverTo: new Map() }
)
})

Expand All @@ -115,9 +117,32 @@ describe('OrganizationMembersService', () => {
await createMember(outsider, otherOrganizationId, true)
await createMember(memberEmail, organizationId, true)

expect(await service.getBlockedRecipients(team.id, [memberEmail, outsider], { fresh: false })).toEqual([
{ address: outsider, unverifiedMember: false },
])
expect(await service.checkRecipients(team.id, [memberEmail, outsider], { fresh: false })).toEqual({
blocked: [{ address: outsider, unverifiedMember: false }],
deliverTo: new Map([[memberEmail, memberEmail]]),
})
})

it.each([
['delivers to the address that matches one exactly', ['Member', 'member'], 'Member', 'Member'],
['blocks an address that matches neither', ['Member', 'member'], 'MEMBER', undefined],
[
'delivers to the one stored address when padding is all they differ by',
['member', ' member'],
'MEMBER',
'member',
],
])('%s of two members whose stored addresses differ', async (_case, stored, written, delivered) => {
for (const name of stored) {
await createMember(`${name}-${team.id}@example.com`, organizationId, true)
}
const address = `${written}-${team.id}@example.com`
const deliveredAddress = delivered && `${delivered}-${team.id}@example.com`

const check = await service.checkRecipients(team.id, [address], { fresh: false })

expect(check.deliverTo.get(address)).toBe(deliveredAddress)
expect(check.blocked).toEqual(deliveredAddress ? [] : [{ address, unverifiedMember: false }])
})

type FakeClient = EventEmitter & { query: jest.Mock; release: jest.Mock }
Expand Down Expand Up @@ -153,7 +178,7 @@ describe('OrganizationMembersService', () => {

const checks = Promise.allSettled(
Array.from({ length: 5 }, () =>
stalledService.getBlockedRecipients(team.id, [memberEmail], { fresh: true })
stalledService.checkRecipients(team.id, [memberEmail], { fresh: true })
)
)
await jest.advanceTimersByTimeAsync(3_000)
Expand All @@ -169,9 +194,12 @@ describe('OrganizationMembersService', () => {
).toEqual(Array(5).fill([!connectsLate]))

healthy = true
const next = stalledService.getBlockedRecipients(team.id, [memberEmail], { fresh: true })
const next = stalledService.checkRecipients(team.id, [memberEmail], { fresh: true })
await jest.advanceTimersByTimeAsync(0)
await expect(next).resolves.toEqual([{ address: memberEmail, unverifiedMember: false }])
await expect(next).resolves.toEqual({
blocked: [{ address: memberEmail, unverifiedMember: false }],
deliverTo: new Map(),
})
} finally {
jest.useRealTimers()
}
Expand Down Expand Up @@ -211,7 +239,7 @@ describe('OrganizationMembersService', () => {
const service = new OrganizationMembersService(pool)

const checks = Array.from({ length: 20 }, () =>
service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })
service.checkRecipients(1, ['member@example.com'], { fresh: true })
)
const outcomes = Promise.allSettled(checks)
await jest.advanceTimersByTimeAsync(3_000)
Expand All @@ -228,9 +256,9 @@ describe('OrganizationMembersService', () => {
lateClients.map((client) => [client.query.mock.calls.length, client.release.mock.calls.length])
).toEqual(Array(5).fill([0, 1]))

await expect(service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).resolves.toEqual(
[]
)
await expect(
service.checkRecipients(1, ['member@example.com'], { fresh: true })
).resolves.toMatchObject({ blocked: [] })
expect(memberSelects()).toBe(1)
} finally {
jest.useRealTimers()
Expand Down Expand Up @@ -260,17 +288,17 @@ describe('OrganizationMembersService', () => {
})
const service = new OrganizationMembersService(routerFor(query))
try {
const first = service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })
const first = service.checkRecipients(1, ['member@example.com'], { fresh: false })
await started
const joining = service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })
const joining = service.checkRecipients(1, ['member@example.com'], { fresh: false })
releaseLookup()
expect(await Promise.all([first, joining])).toEqual([[], []])
expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })).toEqual([])
expect((await Promise.all([first, joining])).map((check) => check.blocked)).toEqual([[], []])
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: false })).blocked).toEqual([])
expect(query).toHaveBeenCalledTimes(1)

clock.mockReturnValue(start + 57_001)
member = 'new-member@example.com'
expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })).toEqual([
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: false })).blocked).toEqual([
{ address: 'member@example.com', unverifiedMember: false },
])
} finally {
Expand All @@ -284,11 +312,13 @@ describe('OrganizationMembersService', () => {
const service = new OrganizationMembersService(routerFor(query))
const removed = [{ address: 'member@example.com', unverifiedMember: false }]

expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })).toEqual([])
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: false })).blocked).toEqual([])
member = 'new-member@example.com'

expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).toEqual(removed)
expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })).toEqual(removed)
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: true })).blocked).toEqual(removed)
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: false })).blocked).toEqual(
removed
)
})

it('reloads a cached block older than five seconds before reporting it, and reuses the reload for later blocks', async () => {
Expand All @@ -299,15 +329,15 @@ describe('OrganizationMembersService', () => {
const service = new OrganizationMembersService(routerFor(query))
const outsider = [{ address: 'outside@example.com', unverifiedMember: false }]
try {
expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })).toEqual([])
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: false })).blocked).toEqual([])
members = ['member@example.com', 'joined@example.com']
clock.mockReturnValue(start + 5_001)

expect(await service.getBlockedRecipients(1, ['joined@example.com'], { fresh: false })).toEqual([])
expect(await service.getBlockedRecipients(1, ['outside@example.com'], { fresh: false })).toEqual(
expect((await service.checkRecipients(1, ['joined@example.com'], { fresh: false })).blocked).toEqual([])
expect((await service.checkRecipients(1, ['outside@example.com'], { fresh: false })).blocked).toEqual(
outsider
)
expect(await service.getBlockedRecipients(1, ['outside@example.com'], { fresh: false })).toEqual(
expect((await service.checkRecipients(1, ['outside@example.com'], { fresh: false })).blocked).toEqual(
outsider
)
expect(query).toHaveBeenCalledTimes(2)
Expand All @@ -334,24 +364,24 @@ describe('OrganizationMembersService', () => {
const service = new OrganizationMembersService(routerFor(query))
const outsider = [{ address: 'outside@example.com', unverifiedMember: false }]
try {
const firstLoad = service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })
const firstLoad = service.checkRecipients(1, ['member@example.com'], { fresh: false })
await jest.advanceTimersByTimeAsync(0)
expect(await firstLoad).toEqual([])
expect((await firstLoad).blocked).toEqual([])
jest.setSystemTime(start + 5_001)

const first = service.getBlockedRecipients(1, ['outside@example.com'], { fresh: false })
const first = service.checkRecipients(1, ['outside@example.com'], { fresh: false })
await jest.advanceTimersByTimeAsync(0)
await started
const later = Array.from({ length: 5 }, () =>
service.getBlockedRecipients(1, ['outside@example.com'], { fresh: false })
service.checkRecipients(1, ['outside@example.com'], { fresh: false })
)
await jest.advanceTimersByTimeAsync(0)
expect(query).toHaveBeenCalledTimes(2)

releaseReload()
const results = Promise.all([first, ...later])
const checks = Promise.all([first, ...later])
await jest.advanceTimersByTimeAsync(100)
expect(await results).toEqual(Array(6).fill(outsider))
expect((await checks).map((check) => check.blocked)).toEqual(Array(6).fill(outsider))
expect(query).toHaveBeenCalledTimes(3)
} finally {
jest.useRealTimers()
Expand All @@ -367,13 +397,11 @@ describe('OrganizationMembersService', () => {
.mockRejectedValue(new Error('primary unavailable'))
const service = new OrganizationMembersService(routerFor(query))
try {
expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })).toEqual([])
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: false })).blocked).toEqual([])
clock.mockReturnValue(start + 5_001)

await expect(
service.getBlockedRecipients(1, ['outside@example.com'], { fresh: false })
).rejects.toThrow()
expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })).toEqual([])
await expect(service.checkRecipients(1, ['outside@example.com'], { fresh: false })).rejects.toThrow()
expect((await service.checkRecipients(1, ['member@example.com'], { fresh: false })).blocked).toEqual([])
expect(query).toHaveBeenCalledTimes(2)
} finally {
clock.mockRestore()
Expand All @@ -388,15 +416,15 @@ describe('OrganizationMembersService', () => {
.mockResolvedValue({ rows: [memberRow('new-member@example.com')] })
const service = new OrganizationMembersService(routerFor(query))

const earlierCheck = service.getBlockedRecipients(1, ['member@example.com'], { fresh: false })
const earlierCheck = service.checkRecipients(1, ['member@example.com'], { fresh: false })
while (query.mock.calls.length === 0) {
await new Promise((resolve) => setImmediate(resolve))
}
const freshCheck = service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })
const freshCheck = service.checkRecipients(1, ['member@example.com'], { fresh: true })
finishEarlierLookup({ rows: [memberRow('member@example.com')] })

expect(await earlierCheck).toEqual([])
expect(await freshCheck).toEqual([{ address: 'member@example.com', unverifiedMember: false }])
expect((await earlierCheck).blocked).toEqual([])
expect((await freshCheck).blocked).toEqual([{ address: 'member@example.com', unverifiedMember: false }])
expect(query).toHaveBeenCalledTimes(2)
})
})
Expand Down
62 changes: 48 additions & 14 deletions nodejs/src/cdp/services/managers/organization-members.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,20 @@ import { LazyLoader } from '~/common/utils/lazy-loader'

import { BoundedPrimaryLookup } from '../../utils/bounded-primary-lookup'

type StoredCasingsByNormalizedEmail = ReadonlyMap<string, ReadonlySet<string>>

type OrganizationMemberSnapshot = {
verifiedEmails: ReadonlySet<string>
verifiedEmails: StoredCasingsByNormalizedEmail
unverifiedEmails: ReadonlySet<string>
loadedAt: number
}

export type BlockedRecipient = { readonly address: string; readonly unverifiedMember: boolean }

export type StoredAddressByRecipient = ReadonlyMap<string, string>

export type RecipientCheck = { readonly blocked: BlockedRecipient[]; readonly deliverTo: StoredAddressByRecipient }

const MAX_SNAPSHOT_AGE_MS = 60_000
const LOOKUP_TIMEOUT_MS = 3_000
const REFRESH_AGE_MS = MAX_SNAPSHOT_AGE_MS - LOOKUP_TIMEOUT_MS
Expand All @@ -22,10 +28,26 @@ type MemberRow = { team_id: number; email: string; is_email_verified: boolean |

const normalizeEmail = (email: string): string => email.trim().toLowerCase()

const blockedBy = (snapshot: OrganizationMemberSnapshot, recipients: string[]): BlockedRecipient[] =>
recipients
.filter((email) => !snapshot.verifiedEmails.has(normalizeEmail(email)))
.map((address) => ({ address, unverifiedMember: snapshot.unverifiedEmails.has(normalizeEmail(address)) }))
const pickStoredMailbox = (storedCasings: ReadonlySet<string> | undefined, address: string): string | undefined => {
if (storedCasings?.size === 1) {
return [...storedCasings][0]
}
return storedCasings?.has(address.trim()) ? address.trim() : undefined
}

const checkAgainst = (snapshot: OrganizationMemberSnapshot, recipients: string[]): RecipientCheck => {
const blocked: BlockedRecipient[] = []
const deliverTo = new Map<string, string>()
for (const address of recipients) {
const match = pickStoredMailbox(snapshot.verifiedEmails.get(normalizeEmail(address)), address)
if (match === undefined) {
blocked.push({ address, unverifiedMember: snapshot.unverifiedEmails.has(normalizeEmail(address)) })
} else {
deliverTo.set(address, match)
}
}
return { blocked, deliverTo }
}

export class OrganizationMembersService {
private snapshots: LazyLoader<OrganizationMemberSnapshot>
Expand All @@ -47,25 +69,28 @@ export class OrganizationMembersService {
})
}

public async getBlockedRecipients(
public async checkRecipients(
teamId: number,
recipients: string[],
{ fresh }: { fresh: boolean }
): Promise<BlockedRecipient[]> {
): Promise<RecipientCheck> {
const id = String(teamId)
if (fresh) {
const blocked = blockedBy((await this.loadSnapshots([id]))[id], recipients)
if (blocked.length) {
const check = checkAgainst((await this.loadSnapshots([id]))[id], recipients)
if (check.blocked.length) {
this.snapshots.markForRefresh(id)
}
return blocked
return check
}
let snapshot = await this.cachedSnapshot(id)
if (blockedBy(snapshot, recipients).length && Date.now() - snapshot.loadedAt > BLOCK_RECHECK_AGE_MS) {
if (
checkAgainst(snapshot, recipients).blocked.length &&
Date.now() - snapshot.loadedAt > BLOCK_RECHECK_AGE_MS
) {
await this.snapshots.reload(id)
snapshot = await this.cachedSnapshot(id)
}
return blockedBy(snapshot, recipients)
return checkAgainst(snapshot, recipients)
}

private async cachedSnapshot(id: string): Promise<OrganizationMemberSnapshot> {
Expand All @@ -81,12 +106,21 @@ export class OrganizationMembersService {
const members = new Map(
teamIds.map((id) => [
id,
{ verifiedEmails: new Set<string>(), unverifiedEmails: new Set<string>(), loadedAt: Date.now() },
{
verifiedEmails: new Map<string, Set<string>>(),
unverifiedEmails: new Set<string>(),
loadedAt: Date.now(),
},
])
)
for (const row of rows) {
const { verifiedEmails, unverifiedEmails } = members.get(String(row.team_id))!
;(row.is_email_verified === true ? verifiedEmails : unverifiedEmails).add(normalizeEmail(row.email))
const key = normalizeEmail(row.email)
if (row.is_email_verified === true) {
verifiedEmails.set(key, (verifiedEmails.get(key) ?? new Set()).add(row.email.trim()))
} else {
unverifiedEmails.add(key)
}
}
return Object.fromEntries(members)
}
Expand Down
Loading
Loading