diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index fa96812f88c0..9ce0d07ccebc 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -39,6 +39,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. @@ -92,5 +93,7 @@ The worker checks the pause again after the daily allowance claim and gives the Django mirrors the tenant named by `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` from the SES tenant-state webhook and the daily reconciliation, without emailing project admins. Set the same `SES_SANDBOX_TENANT_NAME` on Django web, Celery, and the worker. A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Own-sender sends do not read this state. Each delivered send emits a `workflows sandbox email sent` event. -Successful sandbox sends still count as `email_sent`, which the sending tier reads, until sandbox sends get their own metric. -Keep the switch off in production until that metric is deployed. +A successful sandbox send records `email_sandbox_sent` next to `email_sent`, so the workflow's metrics still show it. +The sending tier subtracts `email_sandbox_sent` from `email_sent` per day, and the team's sending allowance per workflow, both floored at zero. Sandbox bounces and complaints still count. +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 this metric to the worker and Django before you enable the switch in production. diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts index 649511b14f71..078f2b387690 100644 --- a/nodejs/src/cdp/cdp-api.ts +++ b/nodejs/src/cdp/cdp-api.ts @@ -246,7 +246,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 c9e73bfa3f7a..18bf9d36fd76 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, + redis, + sandboxEmailRateLimiter, + 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, - redis, - sandboxEmailRateLimiter, - 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..bfe56f14e91b 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.captureDelivered(parsedTeamId, isTest, recipientCount) + } + } + } + private parseTrackingParams(query: Record): { functionId?: string invocationId?: string diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index 92870928aa44..5e0f68c486e2 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -509,7 +509,7 @@ describe('EmailService', () => { }), ]) expect(result.metrics.map((metric) => metric.metric_name)).toEqual( - isTest ? [] : ['email_sent', 'email_untracked'] + isTest ? [] : ['email_sent', 'email_untracked', 'email_sandbox_sent'] ) expect(capture).toHaveBeenCalledWith( expect.objectContaining({ id: team.id }), diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index be25934f7f80..679651ceaf4d 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -897,6 +897,17 @@ export class EmailService { }) } + if (success && isSandbox) { + result.metrics.push({ + team_id: invocation.teamId, + app_source_id: invocation.parentRunId ?? invocation.functionId, + instance_id: invocation.state.actionId || invocation.id, + metric_kind: 'email', + metric_name: 'email_sandbox_sent', + count: 1, + }) + } + if (success && assetRow) { result.messageAssets.push(assetRow) } 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 0bb188858895..f3c7b6e5ad40 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -266,6 +266,13 @@ export class SandboxEmailSender { }) } + public async captureDelivered(teamId: number, isTest: boolean, recipientCount: number): Promise { + await this.captureEvent(teamId, 'workflows sandbox email delivered', { + recipient_count: recipientCount, + is_test: isTest, + }) + } + public async captureBlocked( teamId: number, isTest: boolean, diff --git a/nodejs/src/cdp/types.ts b/nodejs/src/cdp/types.ts index 79aa6c4a11f3..53869cefe606 100644 --- a/nodejs/src/cdp/types.ts +++ b/nodejs/src/cdp/types.ts @@ -269,6 +269,7 @@ export type MinimalAppMetric = { | 'email_blocked' | 'email_unsubscribed' | 'email_untracked' + | 'email_sandbox_sent' | 'push_sent' | 'push_failed' | 'push_skipped' diff --git a/products/workflows/backend/services/email_health.py b/products/workflows/backend/services/email_health.py index 5be6acff5ff5..ab19f2293176 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 6f4a65ffa139..8a2567b75df7 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"),