diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index c1c81d1f8f33..420cfec8fdd0 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -52,6 +52,7 @@ None of them are wired there yet; the charts change is part of turning the rollo In order: staff suspension or an AWS-paused tenant drops to tier 0; a dirty 7-day window, a workflow auto-pause, or a HIGH tenant reputation impact demotes one tier (at most once per cooldown, anchored on the last rate demotion); a tier above 0 with zero sends for the decay period drops one step per period; promotion needs the dwell served, real use of the tier (at least the use ratio of the daily cap on the minimum number of separate days, counted only after the tier anchor), a clean 30-day window, and a clean tenant reputation. Rates only count on a meaningful denominator; below the complaint floor, the absolute complaint backstop still applies, including in windows with no sends. +Sandbox sender sends never count as sends: the sweep subtracts `email_sandbox_sent` from `email_sent` per day, floored at zero. Their bounces and complaints still count. Pinned teams never move automatically. While the tier is enforced, the sweep notifies teams it moved: a rate demotion sends an in-app notification plus an email to project admins, and an earned promotion sends an in-app notification only. @@ -81,6 +82,7 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent. A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Own-sender sends do not read this state. Test sends run inline in the CDP API, so the API reads the same `SES_RATE_LIMITER_VALKEY_*` settings as the email worker. Without them, every sandbox test send is skipped. CDP API opens the SES Valkey pool only when sandbox delivery is enabled; otherwise its MX validation uses the existing local cache and DNS. + The CDP API's SES webhook captures `workflows sandbox email delivered` for a delivery on `SES_SANDBOX_CONFIGURATION_SET`, test sends included. Without that setting on the CDP API, the event never fires. Deploy membership checks, daily caps, and the shared pause gate first. Configure the SES identity and Valkey pool, enable `WORKFLOWS_SANDBOX_SENDER_ENABLED`, then enable the UI flag. - The buckets are token buckets: a full idle bucket plus refill allows up to roughly twice the stated cap in the very first period. The bucket TTLs exceed the refill periods so this does not recur from idling. - A denied send parks until every short bucket has refilled enough to cover it, instead of retrying on a fixed few-minute cadence. diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts index abc282c22f5b..528a0212bc03 100644 --- a/nodejs/src/cdp/cdp-api.ts +++ b/nodejs/src/cdp/cdp-api.ts @@ -240,7 +240,8 @@ export class CdpApi { services.capturedEventsService, services.teamWorkflowsConfigService, new EmailTrackingCodeSigner(config.ENCRYPTION_SALT_KEYS, config.CDP_EMAIL_TRACKING_URL), - services.emailSuppressionService + services.emailSuppressionService, + services.sandboxEmailSender ) this.groupsManager = new GroupsManagerService(deps.teamManager, deps.groupRepository) this.batchExportHogFunctionService = new BatchExportHogFunctionService( diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts index 6770b009f0ba..5f867ebe435e 100644 --- a/nodejs/src/cdp/cdp-services.ts +++ b/nodejs/src/cdp/cdp-services.ts @@ -118,6 +118,7 @@ export interface CdpCoreServices { /** Resolved outputs shared across every CDP service/consumer. */ outputs: CdpOutputs emailService: EmailService + sandboxEmailSender: SandboxEmailSender /** Buffers rendered-email asset rows and bulk-flushes them at the batch boundary. */ messageAssetsService: MessageAssetsService } @@ -466,6 +467,20 @@ export function createCdpCoreServices( const sandboxEmailRateLimiter = deps.emailValidationValkey ? new RateLimiterService(deps.emailValidationValkey, { name: 'sandbox-email' }) : null + const sandboxEmailSender = new SandboxEmailSender( + { + enabled: config.WORKFLOWS_SANDBOX_SENDER_ENABLED, + tenantName: config.SES_SANDBOX_TENANT_NAME, + configurationSetName: config.SES_SANDBOX_CONFIGURATION_SET, + fromAddress: config.SES_SANDBOX_FROM_ADDRESS, + dailyTeamCap: config.WORKFLOWS_SANDBOX_DAILY_TEAM_CAP, + dailyRecipientCap: config.WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP, + }, + deps.teamManager, + sandboxEmailRateLimiter, + redis, + new SandboxSenderStateService(deps.postgres, deps.pubSub) + ) const emailService = new EmailService( { sesAccessKeyId: config.SES_ACCESS_KEY_ID, @@ -488,20 +503,7 @@ export function createCdpCoreServices( messageAssetsService, workflowEmailRateLimiter, teamEmailRateLimiter, - new SandboxEmailSender( - { - enabled: config.WORKFLOWS_SANDBOX_SENDER_ENABLED, - tenantName: config.SES_SANDBOX_TENANT_NAME, - configurationSetName: config.SES_SANDBOX_CONFIGURATION_SET, - fromAddress: config.SES_SANDBOX_FROM_ADDRESS, - dailyTeamCap: config.WORKFLOWS_SANDBOX_DAILY_TEAM_CAP, - dailyRecipientCap: config.WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP, - }, - deps.teamManager, - sandboxEmailRateLimiter, - redis, - new SandboxSenderStateService(deps.postgres, deps.pubSub) - ), + sandboxEmailSender, new OrganizationMembersService(deps.postgres) ) const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL) @@ -638,6 +640,7 @@ export function createCdpCoreServices( recipientTokensService, outputs, emailService, + sandboxEmailSender, messageAssetsService, } } diff --git a/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts b/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts index c76983c6f596..1b373db67e52 100644 --- a/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts @@ -17,6 +17,7 @@ import { KAFKA_APP_METRICS_2, KAFKA_LOG_ENTRIES } from '~/common/config/kafka-to import { closeHub, createHub } from '~/common/utils/db/hub' import { PostgresUse } from '~/common/utils/db/postgres' import * as envUtils from '~/common/utils/env-utils' +import * as posthog from '~/common/utils/posthog' import { createCdpConsumerDeps } from '~/tests/helpers/cdp' import { waitForExpect } from '~/tests/helpers/expectations' import { createTestTeamFixture } from '~/tests/helpers/sql' @@ -819,6 +820,119 @@ describe('EmailTrackingService', () => { }) }) + describe('SES webhook records sandbox deliveries', () => { + const sandboxConfigurationSet = 'posthog-messaging-sandbox' + let app: express.Application + let server: Server + let verifySignatureSpy: jest.SpyInstance + let capture: jest.SpyInstance + + beforeEach(() => { + hub.SES_SANDBOX_CONFIGURATION_SET = sandboxConfigurationSet + const api = new CdpApi(hub, createCdpConsumerDeps(hub), { + hogQueue: createMockJobQueue(), + hogflowQueue: createMockJobQueue(), + }) + app = setupExpressApp() + app.use('/', api.router()) + server = app.listen(0, () => {}) + + verifySignatureSpy = jest + .spyOn(SesWebhookHandler.prototype as any, 'verifySnsSignature') + .mockResolvedValue(true) + capture = jest.spyOn(posthog, 'captureTeamEvent').mockImplementation(() => {}) + }) + + afterEach(() => { + server.close() + verifySignatureSpy.mockRestore() + capture.mockRestore() + }) + + const postDelivery = async ( + configurationSet: string, + isTest: boolean, + signedCode = true, + recipients: string[] | null = ['member@example.com'] + ): Promise => { + const signedTrackingCode = signer.generate( + { functionId: 'function-id', id: 'invocation-id', teamId: team.id }, + isTest + ) + const [unsignedTrackingCode] = signedTrackingCode.split('.') + const sesRecord = { + eventType: 'Delivery', + mail: { + timestamp: '2024-01-01T00:00:00.000Z', + source: 'sandbox@example.com', + messageId: 'ses-message-id', + destination: ['member@example.com'], + headers: signedCode ? [{ name: TRACKING_CODE_HEADER_NAME, value: signedTrackingCode }] : [], + tags: { + 'ses:configuration-set': [configurationSet], + ...(signedCode ? {} : { ph_id: [unsignedTrackingCode] }), + }, + }, + delivery: { + timestamp: '2024-01-01T00:00:01.000Z', + ...(recipients ? { recipients } : {}), + }, + } + const envelope = { + Type: 'Notification', + MessageId: 'sns-message-id', + TopicArn: 'arn:aws:sns:us-east-1:123456789012:ses-events', + Message: JSON.stringify(sesRecord), + Timestamp: '2024-01-01T00:00:01.000Z', + SignatureVersion: '1', + Signature: 'stubbed', + SigningCertURL: 'https://sns.us-east-1.amazonaws.com/cert.pem', + } + return await supertest(app) + .post('/public/m/ses_webhook') + .set('Content-Type', 'text/plain') + .send(JSON.stringify(envelope)) + } + + it.each([false, true])( + 'captures one delivered event for a delivery on the sandbox set (test send: %s)', + async (isTest) => { + const res = await postDelivery(sandboxConfigurationSet, isTest) + + expect(res.status).toBe(200) + expect(capture).toHaveBeenCalledTimes(1) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id, organization_id: team.organization_id }), + 'workflows sandbox email delivered', + { is_test: isTest, recipient_count: 1 } + ) + } + ) + + it.each([ + ['a delivery on another configuration set', 'posthog-messaging-tracked', true], + ['a sandbox delivery whose tracking code is not signed', sandboxConfigurationSet, false], + ])('captures no delivered event for %s', async (_name, configurationSet, signedCode) => { + const res = await postDelivery(configurationSet, false, signedCode) + + expect(res.status).toBe(200) + expect(capture).not.toHaveBeenCalled() + }) + + it.each([ + { recipients: ['member@example.com', 'another@example.com'], count: 2 }, + { recipients: null, count: 0 }, + ])('counts recipient copies in a delivery notification ($count)', async ({ recipients, count }) => { + const res = await postDelivery(sandboxConfigurationSet, false, true, recipients) + expect(res.status).toBe(200) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email delivered', + { is_test: false, recipient_count: count } + ) + }) + }) + describe('resolveEmailEngagementDistinctId', () => { const buildInvocation = ( globals: Partial> diff --git a/nodejs/src/cdp/services/messaging/email-tracking.service.ts b/nodejs/src/cdp/services/messaging/email-tracking.service.ts index 645866c70885..6a8c60e881d6 100644 --- a/nodejs/src/cdp/services/messaging/email-tracking.service.ts +++ b/nodejs/src/cdp/services/messaging/email-tracking.service.ts @@ -24,6 +24,7 @@ import { EmailSuppressionService } from './email-suppression.service' import { escapeHtml } from './helpers/escape-html' import { SES_LINK_INDEX_TAG, SesWebhookHandler } from './helpers/ses' import { EmailTrackingCodeSigner, trackingCodeFormatCounter } from './helpers/tracking-code' +import { SandboxEmailSender } from './sandbox-email-sender' export const PIXEL_GIF = Buffer.from('R0lGODlhAQABAAAAACH5BAEKAAEALAAAAAABAAEAAAICTAEAOw==', 'base64') const LINK_REGEX = @@ -280,10 +281,15 @@ export class EmailTrackingService { private capturedEventsService: CapturedEventsService, private teamWorkflowsConfigService: TeamWorkflowsConfigService, private trackingCodeSigner: EmailTrackingCodeSigner, - private emailSuppressionService: EmailSuppressionService + private emailSuppressionService: EmailSuppressionService, + private sandboxEmailSender: SandboxEmailSender ) { const allowedTopicArns = (process.env.SES_ALLOWED_SNS_TOPIC_ARNS ?? '').split(',') - this.sesWebhookHandler = new SesWebhookHandler(this.trackingCodeSigner, allowedTopicArns) + this.sesWebhookHandler = new SesWebhookHandler( + this.trackingCodeSigner, + allowedTopicArns, + this.sandboxEmailSender.config.configurationSetName + ) } public async trackMetric({ @@ -493,6 +499,7 @@ export class EmailTrackingService { hardBounceRecipients, complainedRecipients, deliveredRecipients, + sandboxDeliveries, } = await this.sesWebhookHandler.handleWebhook({ body: parseJSON(req.body), headers: req.headers, @@ -577,6 +584,8 @@ export class EmailTrackingService { emailTrackingErrorsCounter.inc({ error_type: 'suppression_update_failed', source: 'ses' }) } + await this.captureSandboxDeliveries(sandboxDeliveries || []) + return { status, message: body as string } } catch (error) { emailTrackingErrorsCounter.inc({ error_type: error.name || 'unknown' }) @@ -585,6 +594,17 @@ export class EmailTrackingService { } } + private async captureSandboxDeliveries( + deliveries: { teamId: string; isTest: boolean; recipientCount: number }[] + ): Promise { + for (const { teamId, isTest, recipientCount } of deliveries) { + const parsedTeamId = Number(teamId) + if (Number.isSafeInteger(parsedTeamId) && parsedTeamId > 0) { + await this.sandboxEmailSender.capture(parsedTeamId, isTest, { type: 'delivered', recipientCount }) + } + } + } + private parseTrackingParams(query: Record): { functionId?: string invocationId?: string diff --git a/nodejs/src/cdp/services/messaging/helpers/ses.ts b/nodejs/src/cdp/services/messaging/helpers/ses.ts index 04a6ced57eac..e98fedec8b24 100644 --- a/nodejs/src/cdp/services/messaging/helpers/ses.ts +++ b/nodejs/src/cdp/services/messaging/helpers/ses.ts @@ -396,7 +396,8 @@ export class SesWebhookHandler { private trackingCodeSigner: EmailTrackingCodeSigner, // Empty set means no restriction (dev/test); prod is expected to configure the workflow SES // topic ARN. - allowedTopicArns: string[] = [] + allowedTopicArns: string[] = [], + private sandboxConfigurationSetName = '' ) { this.allowedTopicArns = new Set(allowedTopicArns.map((arn) => arn.trim()).filter(Boolean)) } @@ -612,6 +613,7 @@ export class SesWebhookHandler { emailAddresses: string[] timestamp?: string }[] + sandboxDeliveries?: { teamId: string; isTest: boolean; recipientCount: number }[] }> { // Payloads carry recipients and clicked links, including reset links from sensitive emails, // so the logs keep identifiers only. @@ -714,6 +716,7 @@ export class SesWebhookHandler { emailAddresses: string[] timestamp?: string }[] = [] + const sandboxDeliveries: { teamId: string; isTest: boolean; recipientCount: number }[] = [] for (const rec of records) { logger.info('[SesWebhookHandler] processing record', { @@ -877,6 +880,20 @@ export class SesWebhookHandler { // carrier is only used for engagement metrics/log entries above. const codeIsTrusted = parsedCode?.format === 'signed' + if ( + rec.eventType === 'Delivery' && + codeIsTrusted && + teamId && + this.sandboxConfigurationSetName && + rec.mail.tags?.['ses:configuration-set']?.[0] === this.sandboxConfigurationSetName + ) { + sandboxDeliveries.push({ + teamId, + isTest: isTest ?? false, + recipientCount: rec.delivery.recipients?.length ?? 0, + }) + } + // Suppression writes (below) skip test sends — editor "Run test" traffic must not be // able to perturb production suppression state by targeting a bad recipient. const suppressionAllowed = teamId && codeIsTrusted && !isTest @@ -937,6 +954,7 @@ export class SesWebhookHandler { hardBounceRecipients, complainedRecipients, deliveredRecipients, + sandboxDeliveries, } } } diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts index d165c1a972ab..234bcae3a756 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -51,6 +51,7 @@ export type SandboxDailyCapClaim = type SandboxEmailOutcome = | { type: 'sent'; recipientCount: number } + | { type: 'delivered'; recipientCount: number } | { type: 'blocked' reason: @@ -63,6 +64,17 @@ type SandboxEmailOutcome = blockedRecipientCount: number } +function outcomeProperties(outcome: SandboxEmailOutcome, isTest: boolean): Record { + switch (outcome.type) { + case 'sent': + return { recipient_count: outcome.recipientCount, source: isTest ? 'test' : 'workflow' } + case 'blocked': + return { reason: outcome.reason, blocked_recipient_count: outcome.blockedRecipientCount } + case 'delivered': + return { recipient_count: outcome.recipientCount } + } +} + function dailyBucket(key: string, requested: number, capacity: number): ClaimRequest { return { key, requested, capacity, refillPerSecond: capacity / SECONDS_PER_DAY, ttlSeconds: DAILY_CAP_TTL_SECONDS } } @@ -292,11 +304,10 @@ export class SandboxEmailSender { } const team = await this.teamManager.getTeam(teamId) if (team) { - const properties = - outcome.type === 'sent' - ? { recipient_count: outcome.recipientCount, source: isTest ? 'test' : 'workflow' } - : { reason: outcome.reason, blocked_recipient_count: outcome.blockedRecipientCount } - captureTeamEvent(team, `workflows sandbox email ${outcome.type}`, { ...properties, is_test: isTest }) + captureTeamEvent(team, `workflows sandbox email ${outcome.type}`, { + ...outcomeProperties(outcome, isTest), + is_test: isTest, + }) } } catch (error) { logger.warn('Could not capture sandbox email event', { teamId, error }) diff --git a/products/workflows/backend/services/email_health.py b/products/workflows/backend/services/email_health.py index 35d86d68a037..4988f1516735 100644 --- a/products/workflows/backend/services/email_health.py +++ b/products/workflows/backend/services/email_health.py @@ -22,6 +22,7 @@ from products.workflows.backend.models.hog_flow.hog_flow import HogFlow from products.workflows.backend.providers.ses import SESProvider from products.workflows.backend.services.email_sending_attribution import fold_email_totals_by_flow +from products.workflows.backend.services.email_sending_tier import SANDBOX_SENT_METRIC, SENT_METRIC, own_sends from products.workflows.backend.utils.email_sending_tiers import max_email_sending_tier, resolve_team_email_sending_tier logger = structlog.get_logger(__name__) @@ -89,9 +90,9 @@ def team_email_sending_allowance(team_id: int) -> EmailSendingAllowance: def _team_email_sends_since(team_id: int, after: datetime) -> int: totals = fetch_app_metric_totals_by_team_and_source( - app_source="hog_flow", name=["email_sent"], after=after, team_ids=[team_id] + app_source="hog_flow", name=[SENT_METRIC, SANDBOX_SENT_METRIC], after=after, team_ids=[team_id] ) - return sum(counts.get("email_sent", 0) for counts in totals.get(team_id, {}).values()) + return sum(own_sends(counts) for counts in totals.get(team_id, {}).values()) AWS_TENANT_REPUTATION_CACHE_SECONDS = 5 * 60 diff --git a/products/workflows/backend/services/email_sending_tier.py b/products/workflows/backend/services/email_sending_tier.py index 4f3b83feae6b..ee8a3126eac4 100644 --- a/products/workflows/backend/services/email_sending_tier.py +++ b/products/workflows/backend/services/email_sending_tier.py @@ -25,6 +25,7 @@ APP_SOURCE = "hog_flow" SENT_METRIC = "email_sent" +SANDBOX_SENT_METRIC = "email_sandbox_sent" HARD_BOUNCE_METRIC = "email_bounced_hard" # SES complaint events are recorded under this name (see the SES webhook handler in the email # worker), so a complaint rate reads `email_blocked`, not a metric called "complaint". @@ -330,7 +331,7 @@ def _fetch_daily_metrics( workload: Workload = Workload.DEFAULT, ) -> dict[int, dict[str, dict[str, int]]]: auto_pause_metrics = [name for name in settings.WORKFLOWS_EMAIL_TIER_AUTO_PAUSE_METRIC_NAMES if name] - metric_names = [SENT_METRIC, HARD_BOUNCE_METRIC, COMPLAINT_METRIC, *auto_pause_metrics] + metric_names = [SENT_METRIC, SANDBOX_SENT_METRIC, HARD_BOUNCE_METRIC, COMPLAINT_METRIC, *auto_pause_metrics] # Tag the query so it is attributable in query-cost analysis and does not trip the untagged-query # guard in local dev. Celery adds only task identity, not a product or feature. This derives tier # state in the background, so it is enrichment, not a customer-facing query. One context here @@ -346,17 +347,22 @@ def _fetch_daily_metrics( ) +def own_sends(counts: dict[str, int]) -> int: + return max(0, counts.get(SENT_METRIC, 0) - counts.get(SANDBOX_SENT_METRIC, 0)) + + def _history_from_daily(team_id: int, days: dict[str, dict[str, int]]) -> TeamSendingHistory: # The tier decision only reads team totals, so metrics recorded under a batch job id and under # its parent workflow id sum the same either way and need no per-source resolution. auto_pause_metrics = [name for name in settings.WORKFLOWS_EMAIL_TIER_AUTO_PAUSE_METRIC_NAMES if name] + daily_sends = {day: sent for day, counts in days.items() if (sent := own_sends(counts)) > 0} return TeamSendingHistory( team_id=team_id, - sent=sum(counts.get(SENT_METRIC, 0) for counts in days.values()), + sent=sum(daily_sends.values()), hard_bounced=sum(counts.get(HARD_BOUNCE_METRIC, 0) for counts in days.values()), complained=sum(counts.get(COMPLAINT_METRIC, 0) for counts in days.values()), auto_paused=any(counts.get(metric, 0) > 0 for counts in days.values() for metric in auto_pause_metrics), - daily_sends={day: counts.get(SENT_METRIC, 0) for day, counts in days.items() if counts.get(SENT_METRIC, 0)}, + daily_sends=daily_sends, ) diff --git a/products/workflows/backend/test/test_email_sending_tier.py b/products/workflows/backend/test/test_email_sending_tier.py index 6509c8d0b0d4..dc564e1824d1 100644 --- a/products/workflows/backend/test/test_email_sending_tier.py +++ b/products/workflows/backend/test/test_email_sending_tier.py @@ -1,6 +1,7 @@ from datetime import timedelta -from posthog.test.base import BaseTest +import time_machine +from posthog.test.base import BaseTest, ClickhouseTestMixin from unittest.mock import patch from django.test import SimpleTestCase, override_settings @@ -8,6 +9,8 @@ from parameterized import parameterized +from posthog.test.fixtures import create_app_metric2 + from products.workflows.backend.management.commands.backfill_workflows_email_sending_tiers import ( Command as BackfillCommand, ) @@ -18,8 +21,10 @@ TeamSendingHistory, TierDecision, apply_tier_decision, + build_sending_histories, decide_tier, highest_qualifying_tier, + recompute_email_sending_tier_for_team, recompute_email_sending_tiers, ) from products.workflows.backend.utils.email_sending_tiers import get_email_sending_tier_limits @@ -446,6 +451,96 @@ def test_pinned_team_never_moves(self, _name: str, tier: int) -> None: assert TeamWorkflowsConfig.objects.get(team=self.team).email_sending_tier == tier +@time_machine.travel("2026-10-05T12:00:00Z", tick=False) +@override_settings(**TIER_SETTINGS) +class TestSandboxSendsInTierHistory(ClickhouseTestMixin, BaseTest): + def _record_metrics(self, *, days_ago: int, **counts: int) -> None: + for metric_name, count in counts.items(): + if count: + create_app_metric2( + team_id=self.team.pk, + app_source="hog_flow", + metric_kind="email", + metric_name=metric_name, + count=count, + timestamp=timezone.now() - timedelta(days=days_ago), + ) + + def _record_sends(self, *, days_ago: int, sent: int, sandbox_sent: int) -> None: + self._record_metrics(days_ago=days_ago, email_sent=sent, email_sandbox_sent=sandbox_sent) + + @parameterized.expand( + [ + ("no sends decay", 1, 0, 0, 0, "inactive"), + ("only sandbox sends decay like no sends", 1, 600, 600, 0, "inactive"), + ("sandbox sends do not lift own sends over the use bar", 0, 600, 300, 0, "tier_not_used_enough"), + ("own sends over the use bar still promote", 0, 900, 300, 1, "clean_and_used"), + ] + ) + def test_only_own_sends_count_toward_tier_history( + self, _name: str, current_tier: int, sent: int, sandbox_sent: int, expected_tier: int, expected_reason: str + ) -> None: + TeamWorkflowsConfig.objects.update_or_create( + team=self.team, + defaults={ + "email_sending_tier": current_tier, + "email_sending_tier_updated_at": timezone.now() - timedelta(days=40), + }, + ) + for days_ago in (1, 2): + self._record_sends(days_ago=days_ago, sent=sent, sandbox_sent=sandbox_sent) + + decision = recompute_email_sending_tier_for_team(self.team.id) + + assert decision is not None + assert (decision.new_tier, decision.reason) == (expected_tier, expected_reason) + assert TeamWorkflowsConfig.objects.get(team=self.team).email_sending_tier == expected_tier + + @parameterized.expand( + [ + ( + "sandbox excess on one day does not erase own sends of another day", + { + 1: {"email_sent": 100, "email_sandbox_sent": 300}, + 2: {"email_sent": 200, "email_sandbox_sent": 50}, + }, + {2: 150}, + 0, + 0, + ), + ( + "bounces and complaints of sandbox sends still count", + {1: {"email_sent": 50, "email_sandbox_sent": 50, "email_bounced_hard": 2, "email_blocked": 1}}, + {}, + 2, + 1, + ), + ] + ) + def test_sandbox_subtraction_is_per_day_and_keeps_feedback( + self, + _name: str, + metrics_by_days_ago: dict[int, dict[str, int]], + expected_daily_sends_by_days_ago: dict[int, int], + expected_hard_bounced: int, + expected_complained: int, + ) -> None: + for days_ago, counts in metrics_by_days_ago.items(): + self._record_metrics(days_ago=days_ago, **counts) + + history = build_sending_histories(after=timezone.now() - timedelta(days=30), team_ids=[self.team.pk])[ + self.team.pk + ] + + expected_daily_sends = { + (timezone.now() - timedelta(days=days_ago)).strftime("%Y-%m-%d"): count + for days_ago, count in expected_daily_sends_by_days_ago.items() + } + assert history.daily_sends == expected_daily_sends + assert history.sent == sum(expected_daily_sends.values()) + assert (history.hard_bounced, history.complained) == (expected_hard_bounced, expected_complained) + + @override_settings(**TIER_SETTINGS) class TestBackfillEmailSendingTiers(BaseTest): def test_suspended_team_is_not_promoted_despite_qualifying_history(self) -> None: diff --git a/products/workflows/backend/tests/api/test_email_reputation.py b/products/workflows/backend/tests/api/test_email_reputation.py index 6d88e11e306d..d9e373088c6a 100644 --- a/products/workflows/backend/tests/api/test_email_reputation.py +++ b/products/workflows/backend/tests/api/test_email_reputation.py @@ -91,6 +91,28 @@ def test_reputation_endpoint_returns_empty_shape_when_nothing_was_sent(self): "email_sending_suspension_reason": "", } + def test_sending_allowance_leaves_out_sandbox_sends(self): + sends_by_source = { + self.team.id: {"flow-a": {"email_sent": 5, "email_sandbox_sent": 3}, "flow-b": {"email_sent": 2}} + } + + def requested_metrics(*, name: list[str], **_kwargs: object) -> dict[int, dict[str, dict[str, int]]]: + return { + team_id: { + source: {metric: count for metric, count in counts.items() if metric in name} + for source, counts in sources.items() + } + for team_id, sources in sends_by_source.items() + } + + with patch( + "products.workflows.backend.services.email_health.fetch_app_metric_totals_by_team_and_source", + side_effect=requested_metrics, + ): + allowance = self._get_reputation({})["sending_allowance"] + assert allowance["emails_sent_last_hour"] == 4 + assert allowance["emails_sent_last_day"] == 4 + @parameterized.expand( [ ("no_findings", "ENABLED", None, "healthy"),