diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index 1fe69735e08b..7e39ca35213a 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -75,3 +75,21 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent. A send denied because the limiter itself failed keeps the shorter token bucket cadence, at most 5 to 10 minutes. - Gmail provides no per-message feedback loop, so complaint rates (ours and AWS's) cannot see Gmail complaints at all. - The `HOGFLOW_BATCH_TRIGGER_ELEVATED_TEAM_IDS` allowlist elevates a team to the top tier in Django only: the UI, the admin, and the batch audience cap show top-tier numbers, but the worker's send-time buckets read the stored tier. An allowlisted team below the top tier is therefore throttled at its stored tier while being told otherwise, and its shadow-mode delay counts overstate real impact. Before enforcing, pin each allowlisted team at the top tier from the Django admin and empty the allowlist, rather than plumbing the list into the worker. + +### Sandbox delivery + +The worker sends through a `sandbox` email integration only when `WORKFLOWS_SANDBOX_SENDER_ENABLED` is on and `SES_SANDBOX_TENANT_NAME`, `SES_SANDBOX_CONFIGURATION_SET`, and `SES_SANDBOX_FROM_ADDRESS` are set. +Otherwise every send through it skips with a reason in the run log and a `workflows sandbox email blocked` event with reason `switch_off`. +Every To, CC, and BCC address must belong to an active member of the team's active organization with a verified email address. +Membership snapshots expire within one minute, and the worker checks recipients again right before delivery. +A failed or stale membership check 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). +A send that passes every check goes out through the fixed SES tenant, configuration set, and From address, whatever the step's custom sender and Reply-To say. +Sandbox sends skip the team's tier budget, click and open tracking, and unsubscribe headers, because they reach only members of the sending organization. +The worker parses and normalizes the HTML, then appends a footer that names the sender and the organization ID. +It overrides root clipping and visibility styles, drops any root style declaration it cannot parse cleanly, and keeps the template's typography, so malformed markup may render differently from an own-domain send. +A template whose HTML cannot keep the footer fails the send. +The visible footer is best effort across email clients and template CSS; the signed tracking header still attributes the message. +Each delivered send emits a `workflows sandbox email sent` event. +Live workflow sends record `email_sent`, so workflow reports count them; test sends do not. +The worker has no daily sandbox caps yet, so only switch it on where Django provisions sandbox senders for a small set of projects. diff --git a/nodejs/package.json b/nodejs/package.json index ad8e54855e7f..33a8915a1ddc 100644 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -137,6 +137,7 @@ "node-schedule": "^2.1.0", "nodemailer": "^8.0.8", "p-limit": "3.1.0", + "parse5": "7.1.2", "pg": "^8.6.0", "pino": "^8.6.0", "posthog-node": "5.61.2", diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts index 6e3eb56bb824..96900b4476a1 100644 --- a/nodejs/src/cdp/cdp-services.ts +++ b/nodejs/src/cdp/cdp-services.ts @@ -5,7 +5,7 @@ import { PersonReadRepository } from '~/common/persons/repositories/person-repos import { RedisV2, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' import { UsageIngestionConfig, createUsageIngestionClient, usageReportTeamMatcher } from '~/common/usage-ingestion' import { PostgresRouter } from '~/common/utils/db/postgres' -import { getRedisHost } from '~/common/utils/db/redis' +import { RedisPoolConfig, getRedisHost } from '~/common/utils/db/redis' import { logger } from '~/common/utils/logger' import { PubSub } from '~/common/utils/pubsub' import { TeamManager } from '~/common/utils/team-manager' @@ -32,6 +32,7 @@ import { InvocationResultsService } from './services/invocation-results.service' import { HogFunctionManagerService } from './services/managers/hog-function-manager.service' import { HogFunctionTemplateManagerService } from './services/managers/hog-function-template-manager.service' import { IntegrationManagerService } from './services/managers/integration-manager.service' +import { OrganizationMembersService } from './services/managers/organization-members.service' import { PersonsManagerService } from './services/managers/persons-manager.service' import { RecipientsManagerService } from './services/managers/recipients-manager.service' import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service' @@ -44,6 +45,7 @@ import { MessageAssetsService } from './services/messaging/message-assets.servic import { PushNotificationService } from './services/messaging/push-notification.service' import { RecipientPreferencesService } from './services/messaging/recipient-preferences.service' import { RecipientTokensService } from './services/messaging/recipient-tokens.service' +import { SandboxEmailSender, sandboxBlockedEventRedisConfig } from './services/messaging/sandbox-email-sender' import { HogFunctionMonitoringService } from './services/monitoring/hog-function-monitoring.service' import { HogInvocationResultsService } from './services/monitoring/hog-invocation-results.service' import { HogWatcherService } from './services/monitoring/hog-watcher.service' @@ -175,6 +177,10 @@ export type CdpCoreServicesConfig = Pick< | 'SES_ENDPOINT' | 'SES_TRACKED_CONFIGURATION_SET' | 'SES_UNTRACKED_CONFIGURATION_SET' + | 'WORKFLOWS_SANDBOX_SENDER_ENABLED' + | 'SES_SANDBOX_TENANT_NAME' + | 'SES_SANDBOX_CONFIGURATION_SET' + | 'SES_SANDBOX_FROM_ADDRESS' | 'EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD' | 'EMAIL_TEAM_SENDING_CAP_MODE' | 'EMAIL_TEAM_SENDING_CAP_HOURLY_BY_TIER' @@ -390,17 +396,24 @@ export function createCdpCoreServices( deps: CdpCoreServicesDeps, redisName = 'cdp-redis' ): CdpCoreServices { - const redis = createRedisV2PoolFromConfig({ - connection: config.CDP_REDIS_HOST + const cdpRedisConnection = (name: string): RedisPoolConfig['connection'] => + config.CDP_REDIS_HOST ? { url: config.CDP_REDIS_HOST, options: { port: config.CDP_REDIS_PORT, password: config.CDP_REDIS_PASSWORD }, - name: redisName, + name, } - : { url: config.REDIS_URL, name: `${redisName}-fallback` }, + : { url: config.REDIS_URL, name: `${name}-fallback` } + const redis = createRedisV2PoolFromConfig({ + connection: cdpRedisConnection(redisName), poolMinSize: config.REDIS_POOL_MIN_SIZE, poolMaxSize: config.REDIS_POOL_MAX_SIZE, }) + // A stalled blocked-event call keeps its connection until Redis answers, so these calls get their own + // connections: on the shared pool they took the ones hog watcher and the duplicate observer wait for. + const sandboxBlockedEventRedis = createRedisV2PoolFromConfig( + sandboxBlockedEventRedisConfig(cdpRedisConnection(`${redisName}-sandbox-blocked-event`)) + ) const redisReader = createCdpReaderRedisPool(config, redis, redisName) const valkeyShadow = createCdpValkeyShadowPools(config, redisName) @@ -491,7 +504,18 @@ export function createCdpCoreServices( workflowsActivationReporter, messageAssetsService, workflowEmailRateLimiter, - teamEmailRateLimiter + 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, + }, + deps.teamManager, + sandboxBlockedEventRedis + ), + new OrganizationMembersService(deps.postgres) ) const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL) const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields) diff --git a/nodejs/src/cdp/config.ts b/nodejs/src/cdp/config.ts index 5e70db94dc93..aa64a59b1f88 100644 --- a/nodejs/src/cdp/config.ts +++ b/nodejs/src/cdp/config.ts @@ -173,6 +173,11 @@ export type CdpConfig = ClickhouseConfig & { // Configuration set without open/click tracking (same delivery/bounce/complaint event destination). // Empty means not provisioned: tracking-off sends fall back to the tracked set with a warning. SES_UNTRACKED_CONFIGURATION_SET: string + // The sandbox sender delivers only with the switch on and all three route settings set. + WORKFLOWS_SANDBOX_SENDER_ENABLED: boolean + SES_SANDBOX_TENANT_NAME: string + SES_SANDBOX_CONFIGURATION_SET: string + SES_SANDBOX_FROM_ADDRESS: string // Comma-separated allowlist of SNS Topic ARNs the SES webhook accepts events from. Empty string // means no restriction (dev/test); production should set this to the workflow SES topic ARN(s). SES_ALLOWED_SNS_TOPIC_ARNS: string @@ -423,6 +428,10 @@ export function getDefaultCdpConfig(): CdpConfig { SES_REGION: isTestEnv() || isDevEnv() ? 'us-east-1' : '', SES_TRACKED_CONFIGURATION_SET: 'posthog-messaging', SES_UNTRACKED_CONFIGURATION_SET: '', + WORKFLOWS_SANDBOX_SENDER_ENABLED: false, + SES_SANDBOX_TENANT_NAME: '', + SES_SANDBOX_CONFIGURATION_SET: '', + SES_SANDBOX_FROM_ADDRESS: '', SES_ALLOWED_SNS_TOPIC_ARNS: '', EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD: 5, diff --git a/nodejs/src/cdp/services/managers/organization-members.service.test.ts b/nodejs/src/cdp/services/managers/organization-members.service.test.ts new file mode 100644 index 000000000000..0fe7d7dc14e6 --- /dev/null +++ b/nodejs/src/cdp/services/managers/organization-members.service.test.ts @@ -0,0 +1,158 @@ +import { PostgresRouter } from '~/common/utils/db/postgres' + +import { OrganizationMembersService } from './organization-members.service' + +const routerFor = (select: jest.Mock): PostgresRouter => + ({ + transaction: (_usage: unknown, _tag: string, run: (tx: unknown) => Promise) => run({}), + query: (_tx: unknown, sql: string) => (sql.startsWith('SET LOCAL') ? Promise.resolve({ rows: [] }) : select()), + }) as unknown as PostgresRouter + +describe('OrganizationMembersService', () => { + const memberRow = (email: string): { team_id: number; email: string; is_email_verified: boolean } => ({ + team_id: 1, + email, + is_email_verified: true, + }) + + it('fails closed when every member lookup outlasts its one-minute decision', async () => { + let now = 0 + const query = jest.fn(() => { + now += 60_000 + return Promise.resolve({ rows: [memberRow('member@example.com')] }) + }) + const service = new OrganizationMembersService(routerFor(query), () => now) + + await expect(service.getBlockedRecipients(1, ['member@example.com'])).rejects.toThrow('expired') + expect(query).toHaveBeenCalledTimes(2) + }) + + it('fails closed when the fresh lookup outlasts its one-minute decision', async () => { + let now = 0 + const query = jest.fn(() => { + now += 60_000 + return Promise.resolve({ rows: [memberRow('member@example.com')] }) + }) + const service = new OrganizationMembersService(routerFor(query), () => now) + + await expect(service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).rejects.toThrow( + 'expired' + ) + expect(query).toHaveBeenCalledTimes(1) + }) + + it('fails closed after three seconds when the lookup cannot get a connection', async () => { + jest.useFakeTimers() + try { + const query = jest.fn() + const stalledPool = { + transaction: () => new Promise(() => {}), + query, + } as unknown as PostgresRouter + const service = new OrganizationMembersService(stalledPool) + + const check = service.getBlockedRecipients(1, ['member@example.com'], { fresh: true }) + const outcome = expect(check).rejects.toThrow('timed out') + await jest.advanceTimersByTimeAsync(3_000) + + await outcome + expect(query).not.toHaveBeenCalled() + } finally { + jest.useRealTimers() + } + }) + + it('keeps at most five lookups open while the pool has no connection, and runs no query for them once it recovers', async () => { + jest.useFakeTimers() + try { + let recovered = false + const waitingForConnection: (() => void)[] = [] + const query = jest.fn((_tx: unknown, sql: string) => + Promise.resolve({ rows: sql.startsWith('SET LOCAL') ? [] : [memberRow('member@example.com')] }) + ) + const pool = { + transaction: (_usage: unknown, _tag: string, run: (tx: unknown) => Promise) => + recovered + ? run({}) + : new Promise((resolve, reject) => + waitingForConnection.push(() => void run({}).then(resolve, reject)) + ), + query, + } as unknown as PostgresRouter + const service = new OrganizationMembersService(pool) + + const checks = Array.from({ length: 20 }, () => + service.getBlockedRecipients(1, ['member@example.com'], { fresh: true }) + ) + const outcomes = Promise.allSettled(checks) + await jest.advanceTimersByTimeAsync(3_000) + + const settled = await outcomes + expect(settled.every((outcome) => outcome.status === 'rejected')).toBe(true) + expect(waitingForConnection).toHaveLength(5) + + recovered = true + waitingForConnection.forEach((connect) => connect()) + await jest.advanceTimersByTimeAsync(0) + expect(query).not.toHaveBeenCalled() + + await expect(service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).resolves.toEqual([]) + expect(query).toHaveBeenCalledTimes(2) + } finally { + jest.useRealTimers() + } + }) + + it('shares one member lookup across sends and reloads it after a minute', async () => { + let now = 0 + let member = 'member@example.com' + const query = jest.fn(() => Promise.resolve({ rows: [memberRow(member)] })) + const service = new OrganizationMembersService(routerFor(query), () => now) + + expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([]) + expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([]) + expect(query).toHaveBeenCalledTimes(1) + + now = 60_001 + member = 'new-member@example.com' + expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([ + { address: 'member@example.com', unverifiedMember: false }, + ]) + expect(query).toHaveBeenCalledTimes(2) + }) + + it('reloads members for a fresh check inside the minute', async () => { + let member = 'member@example.com' + const query = jest.fn(() => Promise.resolve({ rows: [memberRow(member)] })) + const service = new OrganizationMembersService(routerFor(query), () => 0) + + expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([]) + member = 'new-member@example.com' + + expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).toEqual([ + { address: 'member@example.com', unverifiedMember: false }, + ]) + expect(query).toHaveBeenCalledTimes(2) + }) + + it('does not let a fresh check join a lookup that started before it', async () => { + let finishEarlierLookup: (rows: unknown) => void = () => {} + const query = jest + .fn() + .mockReturnValueOnce(new Promise((resolve) => (finishEarlierLookup = resolve))) + .mockResolvedValue({ rows: [memberRow('new-member@example.com')] }) + const service = new OrganizationMembersService(routerFor(query), () => 0) + + // Another send starts loading members, then the member is removed before the final check. + const earlierCheck = service.getBlockedRecipients(1, ['member@example.com']) + while (query.mock.calls.length === 0) { + await new Promise((resolve) => setImmediate(resolve)) + } + const freshCheck = service.getBlockedRecipients(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(query).toHaveBeenCalledTimes(2) + }) +}) diff --git a/nodejs/src/cdp/services/managers/organization-members.service.ts b/nodejs/src/cdp/services/managers/organization-members.service.ts new file mode 100644 index 000000000000..0688652ee05a --- /dev/null +++ b/nodejs/src/cdp/services/managers/organization-members.service.ts @@ -0,0 +1,152 @@ +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' +import { LazyLoader } from '~/common/utils/lazy-loader' + +import { DeadlineSlots } from '../../utils/deadline-slots' + +type OrganizationMemberSnapshot = { + verifiedEmails: ReadonlySet + unverifiedEmails: ReadonlySet + expiresAt: number +} + +export type BlockedRecipient = { readonly address: string; readonly unverifiedMember: boolean } + +const MAX_SNAPSHOT_AGE_MS = 60_000 +const REFRESH_AGE_MS = MAX_SNAPSHOT_AGE_MS - 1_000 +// The lookup is one indexed join. A stalled one fails the check here instead of holding up every other +// send in the worker's batch, which waits for this invocation to finish. The deadline counts the wait for +// a pooled connection too; the statement timeout also stops the query on the server. +const LOOKUP_TIMEOUT_MS = 3_000 +// A lookup past its deadline still waits in the common pool until it gets a connection, so at most this many +// are open at once. A check waiting for a slot holds nothing and fails at its deadline. +const MAX_OPEN_LOOKUPS = 5 + +type MemberRow = { team_id: number; email: string; is_email_verified: boolean } + +const normalizeEmail = (email: string): string => email.trim().toLowerCase() + +export class OrganizationMembersService { + private snapshots: LazyLoader + private lookupSlots = new DeadlineSlots(MAX_OPEN_LOOKUPS) + + constructor( + private postgres: PostgresRouter, + private now: () => number = Date.now + ) { + this.snapshots = new LazyLoader({ + name: 'sandbox_team_organization_members', + refreshAgeMs: REFRESH_AGE_MS, + refreshJitterMs: 0, + bufferMs: 0, + loader: (teamIds) => this.loadSnapshots(teamIds), + }) + } + + // `fresh` runs its own lookup, for the check right before delivery: a member removed while the + // send was prepared must not receive it, so neither the cached snapshot nor a lookup another + // send started earlier may answer it. + public async getBlockedRecipients( + teamId: number, + recipients: string[], + { fresh = false }: { fresh?: boolean } = {} + ): Promise { + const snapshot = fresh + ? await this.loadOwnSnapshot(String(teamId)) + : await this.getFreshSnapshot(String(teamId)) + return recipients + .filter((email) => !snapshot.verifiedEmails.has(normalizeEmail(email))) + .map((address) => ({ address, unverifiedMember: snapshot.unverifiedEmails.has(normalizeEmail(address)) })) + } + + private async loadOwnSnapshot(teamId: string): Promise { + const snapshot = (await this.loadSnapshots([teamId]))[teamId] + if (this.now() >= snapshot.expiresAt) { + throw new Error('The organization member check expired') + } + return snapshot + } + + private async loadSnapshots(teamIds: string[]): Promise> { + const expiresAt = this.now() + MAX_SNAPSHOT_AGE_MS + const rows = await this.lookupMembers(teamIds.map(Number)) + const members = new Map( + teamIds.map((id) => [id, { verified: new Set(), unverified: new Set() }]) + ) + for (const row of rows) { + const { verified, unverified } = members.get(String(row.team_id))! + ;(row.is_email_verified === true ? verified : unverified).add(normalizeEmail(row.email)) + } + return Object.fromEntries( + [...members].map(([id, { verified, unverified }]) => [ + id, + { verifiedEmails: verified, unverifiedEmails: unverified, expiresAt }, + ]) + ) + } + + private async lookupMembers(teamIds: number[]): Promise { + const tag = 'fetch-sandbox-team-organization-members' + let timedOut = false + let timer: NodeJS.Timeout | undefined + const deadline = new Promise((resolve) => { + timer = setTimeout(() => { + timedOut = true + resolve() + }, LOOKUP_TIMEOUT_MS) + }) + let rows: MemberRow[] | undefined + const lookup = this.lookupSlots.run(deadline, async () => { + rows = await this.postgres.transaction(PostgresUse.COMMON_WRITE, tag, async (tx) => { + // Nobody waits for a lookup that got its connection after its deadline, so it runs no query. + if (timedOut) { + throw new Error('The organization member lookup was abandoned at its deadline') + } + await this.postgres.query(tx, `SET LOCAL statement_timeout = ${LOOKUP_TIMEOUT_MS}`, undefined, tag) + // Only an explicit false deactivates an organization, matching the web middleware. + const result = await this.postgres.query( + tx, + `SELECT team.id AS team_id, users.email, users.is_email_verified + FROM posthog_team AS team + JOIN posthog_organization AS organization ON organization.id = team.organization_id + JOIN posthog_organizationmembership AS membership ON membership.organization_id = organization.id + JOIN posthog_user AS users ON users.id = membership.user_id + WHERE team.id = ANY($1::bigint[]) + AND organization.is_active IS NOT FALSE + AND users.is_active = true`, + [teamIds], + tag + ) + return result.rows + }) + }) + // A lookup that loses the race can still fail later, after nobody waits for it. + lookup.catch(() => undefined) + try { + await Promise.race([lookup, deadline]) + } finally { + clearTimeout(timer) + } + if (!rows) { + throw new Error('The organization member lookup timed out') + } + return rows + } + + // A snapshot older than a minute never decides a send: it is reloaded once, and a reload that + // itself outlasts the minute fails the check rather than trusting stale membership. + private async getFreshSnapshot(teamId: string): Promise { + let snapshot = await this.snapshots.get(teamId) + if (snapshot && this.now() >= snapshot.expiresAt) { + this.snapshots.markForRefresh(teamId) + snapshot = await this.snapshots.get(teamId) + } + if (!snapshot) { + throw new Error('Could not check organization members') + } + if (this.now() >= snapshot.expiresAt) { + this.snapshots.markForRefresh(teamId) + throw new Error('The organization member check expired') + } + return snapshot + } +} diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index f75e6ade4d29..34322db025d7 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -1,6 +1,11 @@ import { mockFetch } from '~/tests/helpers/mocks/request.mock' -import { MessageRejected, SendingPausedException, TooManyRequestsException } from '@aws-sdk/client-sesv2' +import { + MessageRejected, + SendEmailCommand, + SendingPausedException, + TooManyRequestsException, +} from '@aws-sdk/client-sesv2' import { DateTime } from 'luxon' import { register } from 'prom-client' @@ -12,18 +17,25 @@ import { CyclotronInvocationQueueParametersEmailType, } from '~/cdp/schema/cyclotron' import { TRANSACTIONAL_EMAIL_EVENT } from '~/cdp/services/hogflows/hogflow-sensitivity' -import { CyclotronJobInvocationHogFunction } from '~/cdp/types' +import { CyclotronJobInvocationHogFunction, CyclotronJobInvocationResult } from '~/cdp/types' import { IngestionOutputs } from '~/common/outputs/ingestion-outputs' -import { createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' +import { RedisClient, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' import { closeHub, createHub } from '~/common/utils/db/hub' -import { PostgresUse } from '~/common/utils/db/postgres' +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' import { parseJSON } from '~/common/utils/json-parse' +import * as posthog from '~/common/utils/posthog' import { UUIDT } from '~/common/utils/utils' import { waitForExpect } from '~/tests/helpers/expectations' -import { createTestTeamFixture } from '~/tests/helpers/sql' +import { + createOrganization, + createOrganizationMembership, + createTestTeamFixture, + createUser, +} from '~/tests/helpers/sql' import { INVALID_TEAM_ID_KEYS } from '~/tests/helpers/team' import { Hub, Team } from '../../../types' +import { OrganizationMembersService } from '../managers/organization-members.service' import { RecipientsManagerService } from '../managers/recipients-manager.service' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { RateLimiterService } from '../rate-limiter/rate-limiter.service' @@ -41,6 +53,7 @@ import { MailDevAPI } from './helpers/maildev' import { EmailTrackingCodeSigner } from './helpers/tracking-code' import { MessageAssetsService } from './message-assets.service' import { RecipientTokensService } from './recipient-tokens.service' +import { SandboxEmailSender, SandboxEmailSenderConfig } from './sandbox-email-sender' class ThrottlingException extends Error { constructor(message: string) { @@ -278,6 +291,712 @@ describe('EmailService', () => { sendEmailSpy = jest.spyOn(service.sesV2Client!, 'send') as any sendEmailSpy.mockResolvedValue({ MessageId: 'test-message-id' }) }) + describe('sandbox sender', () => { + const UNAVAILABLE_LOG = + 'Skipping send: the sandbox sender is unavailable right now. Verify your own domain to keep sending.' + let capture: jest.SpyInstance + let memberId: number + let memberIds: number[] + let extraOrganizationIds: string[] + const memberEmail = (name = 'test'): string => `${name}-${team.id}@example.com` + const createSandboxParams = ( + params: Partial = {} + ): CyclotronInvocationQueueParametersEmailType => + createEmailParams({ + from: { integrationId: 4 }, + to: { email: memberEmail(), name: 'Example member' }, + ...params, + }) + const createMember = async (email: string, organizationId = team.organization_id): Promise => { + const id = await createUser(hub.postgres, email) + memberIds.push(id) + await createOrganizationMembership(hub.postgres, organizationId, id) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET email = $1, is_active = true, is_email_verified = true WHERE id = $2', + [email, id], + 'test:verify-sandbox-member' + ) + return id + } + const createSandboxService = ( + enabled: boolean, + membersPostgres: PostgresRouter = hub.postgres, + routeSettings: Partial = {}, + teamEmailRateLimiter: RateLimiterService | null = null, + messageAssetsService?: MessageAssetsService, + blockedEventRedis?: ConstructorParameters[2] + ): EmailService => { + const workflowsConfig = new TeamWorkflowsConfigService(hub.postgres, hub.pubSub) + jest.spyOn(workflowsConfig, 'getEmailSendingTier').mockResolvedValue(0) + const sandboxService = new EmailService( + { + sesAccessKeyId: hub.SES_ACCESS_KEY_ID, + sesSecretAccessKey: hub.SES_SECRET_ACCESS_KEY, + sesRegion: hub.SES_REGION, + sesEndpoint: hub.SES_ENDPOINT, + sesTrackedConfigurationSet: hub.SES_TRACKED_CONFIGURATION_SET, + sesUntrackedConfigurationSet: hub.SES_UNTRACKED_CONFIGURATION_SET, + teamEmailCapMode: 'enforce', + teamEmailTierHourlyCaps: [1], + teamEmailTierDailyCaps: [1], + }, + hub.integrationManager, + workflowsConfig, + hub.ENCRYPTION_SALT_KEYS, + hub.SITE_URL, + new EmailTrackingCodeSigner(hub.ENCRYPTION_SALT_KEYS, hub.CDP_EMAIL_TRACKING_URL), + new EmailSuppressionService(hub.postgres, emailSuppressionConfigFromEnv()), + new RecipientsManagerService(hub.postgres), + null, + workflowsActivationReporter, + messageAssetsService, + null, + teamEmailRateLimiter, + new SandboxEmailSender( + { + enabled, + tenantName: 'sandbox-tenant', + configurationSetName: 'sandbox-email', + fromAddress: 'fixed-sandbox@example.com', + ...routeSettings, + }, + hub.teamManager, + blockedEventRedis ?? { + useClient: async (_options, callback) => { + const client = await hub.redisPool.acquire() + try { + return await callback(client as unknown as RedisClient) + } finally { + await hub.redisPool.release(client) + } + }, + } + ), + new OrganizationMembersService(membersPostgres) + ) + sendEmailSpy = jest.spyOn(sandboxService.sesV2Client!, 'send') as jest.SpyInstance + sendEmailSpy.mockResolvedValue({ MessageId: 'sandbox-message-id' }) + return sandboxService + } + const expectSkipped = (result: CyclotronJobInvocationResult): void => { + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.emailSendResult).toMatchObject({ status: 'skipped' }) + expect(result.invocation.state.vmState?.stack.at(-1)).toEqual({ success: false }) + } + const expectBlockedEvent = (reason: string, isTest: boolean, blockedRecipientCount: number): void => { + expect(capture).toHaveBeenLastCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason, is_test: isTest, blocked_recipient_count: blockedRecipientCount } + ) + } + + beforeEach(async () => { + capture = jest.spyOn(posthog, 'captureTeamEvent').mockImplementation(() => {}) + memberIds = [] + extraOrganizationIds = [] + memberId = await createMember(memberEmail()) + await createMember(memberEmail('cc')) + await createMember(memberEmail('bcc')) + await insertIntegration(hub.postgres, team.id, { + id: getIntegrationId(4), + kind: 'email', + config: { + provider: 'sandbox', + email: 'sandbox@example.com', + domain: 'example.com', + name: 'Example organization via PostHog', + verified: true, + }, + }) + invocation.queueParameters = createSandboxParams() + Object.assign(invocation, { hogFlow: { id: invocation.functionId, team_id: team.id } }) + }) + + afterEach(async () => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organizationmembership WHERE user_id = ANY($1::integer[])', + [memberIds], + 'test:delete-sandbox-memberships' + ) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_user WHERE id = ANY($1::integer[])', + [memberIds], + 'test:delete-sandbox-members' + ) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organization WHERE id = ANY($1::uuid[])', + [extraOrganizationIds], + 'test:delete-sandbox-organizations' + ) + const client = await hub.redisPool.acquire() + try { + const keys = await client.keys(`@posthog/workflows-sandbox-blocked/${team.id}/*`) + if (keys.length) { + await client.del(...keys) + } + } finally { + await hub.redisPool.release(client) + } + capture.mockRestore() + }) + + it.each([ + ['the switch is off', false, false, {}], + ['the switch is off', true, false, {}], + ['the tenant is empty', false, true, { tenantName: '' }], + ['the configuration set is blank', true, true, { configurationSetName: ' ' }], + ['the From address is empty', false, true, { fromAddress: '' }], + ] as const)( + 'skips when the sandbox sender is switched off because %s (isTest=%s)', + async (_reason, isTest, enabled, routeSettings: Partial) => { + service = createSandboxService(enabled, hub.postgres, routeSettings) + + const result = await service.executeSendEmail(invocation, isTest) + + expectSkipped(result) + expect(result.logs).toContainEqual( + expect.objectContaining({ level: 'info', message: UNAVAILABLE_LOG }) + ) + expectBlockedEvent('switch_off', isTest, 0) + } + ) + + it.each([false, true])( + 'sends own-domain email on the team route without member checks (switch on=%s)', + async (enabled) => { + const membersPostgres = { transaction: jest.fn() } as unknown as PostgresRouter + service = createSandboxService(enabled, membersPostgres) + invocation.queueParameters = createEmailParams({ + from: { integrationId: 1 }, + to: { email: 'outside@example.com', name: 'Outside' }, + }) + + const result = await service.executeSendEmail(invocation) + + expect(result.error).toBeUndefined() + expect(result.skipped).toBeFalsy() + expect(membersPostgres.transaction).not.toHaveBeenCalled() + expect(capture).not.toHaveBeenCalled() + const input = sendEmailSpy.mock.calls[0][0].input + expect(input).toMatchObject({ + TenantName: `team-${team.id}`, + FromEmailAddress: '"Test User" ', + Destination: { ToAddresses: ['"Outside" '] }, + }) + expect(input.ConfigurationSetName).not.toEqual('sandbox-email') + expect(input.Content?.Simple?.Headers?.map((header: { Name: string }) => header.Name)).toContain( + 'List-Unsubscribe' + ) + } + ) + + it.each([false, true])( + 'permits sandbox destination invocations only as explicit tests (isTest=%s)', + async (isTest) => { + service = createSandboxService(true) + const destination = createExampleInvocation({ team_id: team.id, id: 'email-destination' }) + destination.state.vmState = { stack: [] } as any + destination.queueParameters = createSandboxParams() + + const result = await service.executeSendEmail(destination, isTest) + + expect(result.error).toBeUndefined() + expect(sendEmailSpy).toHaveBeenCalledTimes(isTest ? 1 : 0) + expect(result.invocation.state.vmState?.stack).toEqual([{ success: isTest }]) + if (!isTest) { + expectSkipped(result) + expect(result.logs).toContainEqual( + expect.objectContaining({ + message: + 'Skipping send: the sandbox sender is only available for workflows and test sends.', + }) + ) + } + } + ) + + it.each([ + [false, 'Hello there.', 'Hello ', false], + [true, 'Hello there.', 'Hello ', false], + [false, '

Hello

', '

Hello

', true], + ] as const)( + 'sends untracked with the fixed identity and organization footer (isTest=%s, html=%s)', + async (isTest, html, expectedContent, htmlOnly) => { + const outputs = { + produce: jest.fn().mockResolvedValue(undefined), + } as unknown as IngestionOutputs<'message_assets'> + service = createSandboxService(true, hub.postgres, {}, null, new MessageAssetsService(outputs)) + invocation.state.actionId = 'send-email' + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4, email: 'override@example.com', name: 'Custom sender' }, + replyTo: 'reply@example.com', + cc: memberEmail('cc'), + bcc: memberEmail('bcc'), + text: htmlOnly ? undefined : 'Hello there.', + html, + }) + invocation.hogFunction.metadata = { + message_category_type: 'marketing', + tracking_enabled: true, + utm_tags_enabled: true, + } + + const result = await service.executeSendEmail(invocation, isTest) + + expect(result.error).toBeUndefined() + expect(result).toMatchObject({ finished: true, emailSendResult: { status: 'accepted' } }) + expect(result.skipped).not.toBe(true) + expect(result.invocation.state.vmState?.stack).toEqual([{ success: true }]) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + const input = (sendEmailSpy.mock.calls[0][0] as SendEmailCommand).input + const footer = `This email was sent by Example organization via PostHog with the PostHog sandbox sender. Organization ID: ${team.organization_id}.` + expect(input).toMatchObject({ + TenantName: 'sandbox-tenant', + ConfigurationSetName: 'sandbox-email', + FromEmailAddress: '"Example organization via PostHog" ', + FeedbackForwardingEmailAddress: 'fixed-sandbox@example.com', + Destination: { + ToAddresses: [`"Example member" <${memberEmail()}>`], + CcAddresses: [memberEmail('cc')], + BccAddresses: [memberEmail('bcc')], + }, + }) + expect(input.Content?.Simple?.Body?.Text).toEqual( + htmlOnly ? undefined : { Data: `Hello there.\n\n${footer}`, Charset: 'UTF-8' } + ) + const sentHtml = input.Content?.Simple?.Body?.Html?.Data + expect(sentHtml).toContain(expectedContent) + if (html.includes('${footer}`)).toBe(true) + expect(input.ReplyToAddresses).toBeUndefined() + const headerNames = input.Content?.Simple?.Headers?.map((header) => header.Name) + expect(headerNames).toContain('X-PostHog-Tracking-Code') + expect(headerNames).not.toContain('List-Unsubscribe') + expect(result.logs.filter((log) => log.message.startsWith('Ignoring custom sender'))).toEqual([ + expect.objectContaining({ + level: 'info', + message: + 'Ignoring custom sender and Reply-To settings: the sandbox sender uses a fixed identity.', + }), + ]) + expect(result.metrics.map((metric) => metric.metric_name)).toEqual( + isTest ? [] : ['email_sent', 'email_untracked'] + ) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email sent', + { is_test: isTest, recipient_count: 3, source: isTest ? 'test' : 'workflow' } + ) + expect(result.messageAssets).toEqual(isTest ? [] : [expect.objectContaining({ html: sentHtml })]) + } + ) + + it('never spends the team sending budget on a sandbox send', async () => { + const claimAllOrNothingPair = jest + .fn() + .mockResolvedValue({ granted: false, deniedIndex: 0, retryAfterMs: 60_000, reserved: false }) + service = createSandboxService(true, hub.postgres, {}, { + claimAllOrNothingPair, + } as unknown as RateLimiterService) + + const result = await service.executeSendEmail(invocation) + + expect(result.error).toBeUndefined() + expect(result.finished).toBe(true) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + expect(claimAllOrNothingPair).not.toHaveBeenCalled() + }) + + it('fails the send when the template cannot keep its identification footer', async () => { + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ html: '

Hello

', + '

Hello

', + null, + ], + [ + '

Hello

', + '

Hello

', + null, + ], + ['', null], + ['Hello </body>', '<pre>Hello &lt;/body&gt;</pre>', null], + ['<body><p>Hello</p><template><plaintext>Example', '<p>Hello</p><template><pre>Example</pre></template>', null], + ['<head><template><plaintext>Example', '<head><template><pre>Example</pre></template></head>', null], + [ + '<body><p>Hello</p><noscript><style>p {color:blue}', + '<p>Hello</p><div><style>p {color:blue}</style></div>', + null, + ], + [ + '<body><p>Hello</p><noscript>&lt;plaintext&gt;Example</noscript></body>', + '<p>Hello</p><div>&lt;plaintext&gt;Example</div>', + null, + ], + [ + '<body><table><tbody><tr><td>Hello</td></tr></tbody></table></body>', + '<td>Hello</td>', + '<textarea>Preview text', + ], + ['<body><p>Hello</p><noscript><style></noscript><!--', '<p>Hello</p>', null], + [ + '<body><p>Hello</p><noscript><style>/* </noscript><plaintext> */ p {color:blue}</style></noscript>', + '<style>/* </noscript><plaintext> */ p {color:blue}</style>', + null, + ], + [ + '<body><p>Hello</p><noscript><p title="</noscript><style>">Fallback</p></noscript></body>', + '<p title="</noscript><style>">Fallback</p>', + null, + ], + [ + '<body><noscript><style></noscript><script></style></noscript>Hello', + '<style></noscript><script></style>', + null, + ], + ['<body style="background:#525252;color:white"><p>Hello</p></body>', '<p>Hello</p>', null], + ])('retains the footer outside template content: %s', async (html, expectedContent, preheader) => { + const result = await sender.withIdentificationFooter( + params(html, undefined, preheader ?? undefined), + 'Example organization via PostHog', + 1 + ) + expect(result.html).toContain(expectedContent) + for (const scriptingEnabled of [false, true]) { + const root = parse(result.html!, { scriptingEnabled }).childNodes.find(defaultTreeAdapter.isElementNode) + const body = root?.childNodes + .filter(defaultTreeAdapter.isElementNode) + .find((node) => node.tagName === 'body') + expect(body?.childNodes.at(-1)).toMatchObject({ + tagName: 'div', + childNodes: [{ nodeName: '#text', value: footer }], + }) + } + expect(result.preheader).toBeUndefined() + }) + it.each([ + ['<body><p>Hello</p><script><!--<script>', null, null], + ['<body><p>Hello</p><template><script><!--<script>', null, null], + ['', '', 'The sandbox email template must include HTML or text content. Update the template and try again.'], + ])('rejects content that cannot retain the footer: %s', async (html, text, error) => { + await expect( + sender.withIdentificationFooter(params(html, text ?? undefined), 'Example organization via PostHog', 1) + ).rejects.toThrow( + error ?? + 'The sandbox email template could not retain its identification footer. Update the template and try again.' + ) + }) +}) diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts new file mode 100644 index 000000000000..b566a1ab510b --- /dev/null +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -0,0 +1,319 @@ +import { type DefaultTreeAdapterMap, defaultTreeAdapter, parse, parseFragment, serialize } from 'parse5' + +import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron' +import { RedisV2 } from '~/common/redis/redis-v2' +import { RedisConnectionConfig, RedisPoolConfig } from '~/common/utils/db/redis' +import { logger } from '~/common/utils/logger' +import { captureTeamEvent } from '~/common/utils/posthog' +import { TeamManager } from '~/common/utils/team-manager' + +import { DeadlineSlots } from '../../utils/deadline-slots' +import { maybeAddPreheaderToEmail } from './helpers/preheader' + +export interface SandboxEmailSenderConfig { + enabled: boolean + tenantName: string + configurationSetName: string + fromAddress: string +} + +const ROUTE_SETTING_NAMES = { + tenantName: 'SES_SANDBOX_TENANT_NAME', + configurationSetName: 'SES_SANDBOX_CONFIGURATION_SET', + fromAddress: 'SES_SANDBOX_FROM_ADDRESS', +} as const + +function emptyRouteSettings(config: SandboxEmailSenderConfig): string[] { + return (Object.keys(ROUTE_SETTING_NAMES) as (keyof typeof ROUTE_SETTING_NAMES)[]) + .filter((setting) => !config[setting].trim()) + .map((setting) => ROUTE_SETTING_NAMES[setting]) +} + +export type SandboxEmailBlockReason = 'switch_off' | 'recipient_not_member' | 'recipient_unverified' | 'check_failed' + +const BLOCKED_EVENT_WINDOW_SECONDS = 3600 +// The send is already decided when this runs, and the worker's batch waits for it, so a stalled +// store must not hold the invocation open. +const EVENT_CAPTURE_TIMEOUT_MS = 1000 +// The deadline cannot cancel a store call, so a stalled call keeps its connection until the store answers. +// At most this many run at once, on a pool of their own this size; a capture waiting for a slot holds +// nothing and gives up at its deadline, so a burst against a healthy store keeps every event. +export const MAX_OPEN_STORE_CALLS = 10 + +function htmlBody(document: DefaultTreeAdapterMap['document']): DefaultTreeAdapterMap['element'] | undefined { + const root = document.childNodes.find(defaultTreeAdapter.isElementNode) + return root?.childNodes.find( + (node): node is DefaultTreeAdapterMap['element'] => + defaultTreeAdapter.isElementNode(node) && node.tagName === 'body' && node.namespaceURI === root.namespaceURI + ) +} + +const BRACKET_CLOSERS: Record<string, string> = { '(': ')', '[': ']', '{': '}' } + +// Keeps only the template's complete declarations, so nothing it leaves open (a comment, string or +// bracket) can swallow the overrides appended after them. Scanning stops at an escape or at a line +// break inside a string, where a browser could tokenize the rest differently. +function completeDeclarations(style: string): string[] { + return scanDeclarations(style).filter((declaration) => declaration.includes(':') && !/[{}]/.test(declaration)) +} + +function scanDeclarations(style: string): string[] { + const declarations: string[] = [] + const closers: string[] = [] + let quote: string | null = null + let current = '' + for (let index = 0; index < style.length; index++) { + const char = style[index] + if (char === '\\' || (quote && /[\n\r\f]/.test(char))) { + return declarations + } + if (quote) { + current += char + quote = char === quote ? null : quote + } else if (char === '/' && style[index + 1] === '*') { + const end = style.indexOf('*/', index + 2) + if (end === -1) { + return declarations + } + // A comment separates tokens, so it becomes a space: dropping it could join a slash and a star. + current += ' ' + index = end + 1 + } else if (char === ';' && !closers.length) { + declarations.push(current) + current = '' + } else { + if (char === '"' || char === "'") { + quote = char + } else if (BRACKET_CLOSERS[char]) { + closers.push(BRACKET_CLOSERS[char]) + } else if (char === closers.at(-1)) { + closers.pop() + } + current += char + } + } + if (!quote && !closers.length) { + declarations.push(current) + } + return declarations +} + +function protectFooterAncestors( + document: DefaultTreeAdapterMap['document'], + body: DefaultTreeAdapterMap['element'] +): void { + const root = document.childNodes.find(defaultTreeAdapter.isElementNode)! + const visibleLayout = + 'display:block!important;position:static!important;margin:0!important;overflow:visible!important;height:auto!important;max-height:none!important;clip:auto!important;clip-path:none!important;transform:none!important;opacity:1!important;visibility:visible!important;filter:none!important;contain:none!important;content-visibility:visible!important;zoom:1!important' + for (const ancestor of [root, body]) { + const style = ancestor.attrs.find((attribute) => attribute.name === 'style') + if (style) { + style.value = [...completeDeclarations(style.value), visibleLayout].join(';') + } else { + ancestor.attrs.push({ name: 'style', value: visibleLayout }) + } + } +} + +function appendHtmlFooter(html: string, footer: string, preheader?: string): string { + const preheaderText = parseFragment(preheader ?? '', { scriptingEnabled: false }) + const document = parse(maybeAddPreheaderToEmail(html, serialize(preheaderText, { scriptingEnabled: false })), { + scriptingEnabled: false, + }) + const body = htmlBody(document) + if (!body) { + throw new Error('The sandbox email template must have an HTML body. Update the template and try again.') + } + const nodes: DefaultTreeAdapterMap['childNode'][] = [...document.childNodes] + for (const node of nodes) { + if (defaultTreeAdapter.isElementNode(node)) { + if (node.tagName === 'plaintext' && node.namespaceURI === body.namespaceURI) { + node.tagName = 'pre' + } + if (node.tagName === 'noscript' && node.namespaceURI === body.namespaceURI) { + node.tagName = 'div' + } + nodes.push(...node.childNodes) + if (node.tagName === 'template' && node.namespaceURI === body.namespaceURI) { + nodes.push( + ...defaultTreeAdapter.getTemplateContent(node as DefaultTreeAdapterMap['template']).childNodes + ) + } + } + } + protectFooterAncestors(document, body) + const paragraph = defaultTreeAdapter.createElement('div', body.namespaceURI, [ + { + name: 'style', + value: 'all:initial!important;display:block!important;visibility:visible!important;opacity:1!important;font:12px sans-serif!important;color:#525252!important;background:#fff!important;padding:16px 0!important', + }, + ]) + defaultTreeAdapter.insertText(paragraph, footer) + defaultTreeAdapter.appendChild(body, paragraph) + const serialized = serialize(document, { scriptingEnabled: false }) + for (const scriptingEnabled of [false, true]) { + const deliveredFooter = htmlBody(parse(serialized, { scriptingEnabled }))?.childNodes.at(-1) + if ( + !deliveredFooter || + !defaultTreeAdapter.isElementNode(deliveredFooter) || + deliveredFooter.tagName !== 'div' || + deliveredFooter.namespaceURI !== body.namespaceURI || + deliveredFooter.attrs.find((attribute) => attribute.name === 'style')?.value !== paragraph.attrs[0].value || + deliveredFooter.childNodes.length !== 1 || + !defaultTreeAdapter.isTextNode(deliveredFooter.childNodes[0]) || + deliveredFooter.childNodes[0].value !== footer + ) { + throw new Error( + 'The sandbox email template could not retain its identification footer. Update the template and try again.' + ) + } + } + return serialized +} + +export function sandboxBlockedEventRedisConfig(connection: RedisConnectionConfig): RedisPoolConfig { + return { + // Blocked events are optional telemetry, so a store that refuses these connections must not stop the worker. + connection: { ...connection, exitOnRepeatedErrors: false }, + poolMinSize: 0, + poolMaxSize: MAX_OPEN_STORE_CALLS, + } +} + +export class SandboxEmailSender { + public readonly canSend: boolean + private storeSlots = new DeadlineSlots(MAX_OPEN_STORE_CALLS) + + constructor( + public readonly config: SandboxEmailSenderConfig, + private teamManager: TeamManager, + private blockedEventRedis: Pick<RedisV2, 'useClient'> + ) { + this.config = { + ...config, + tenantName: config.tenantName.trim(), + configurationSetName: config.configurationSetName.trim(), + fromAddress: config.fromAddress.trim(), + } + const emptySettings = emptyRouteSettings(config) + if (config.enabled && emptySettings.length) { + logger.error('The sandbox sender is switched on with empty route settings, so it skips every send', { + emptySettings, + }) + } + this.canSend = config.enabled && !emptySettings.length + } + + public async withIdentificationFooter( + params: CyclotronInvocationQueueParametersEmailType, + senderName: string, + teamId: number + ): Promise<CyclotronInvocationQueueParametersEmailType> { + if (!params.text && !params.html) { + throw new Error( + 'The sandbox email template must include HTML or text content. Update the template and try again.' + ) + } + const team = await this.teamManager.getTeam(teamId) + if (!team) { + throw new Error('Could not identify the organization sending this sandbox email. Try again.') + } + const footer = `This email was sent by ${senderName} with the PostHog sandbox sender. Organization ID: ${team.organization_id}.` + return { + ...params, + from: { ...params.from, name: senderName, email: this.config.fromAddress }, + replyTo: undefined, + preheader: undefined, + ...(params.text ? { text: `${params.text}\n\n${footer}` } : {}), + ...(params.html ? { html: appendHtmlFooter(params.html, footer, params.preheader) } : {}), + } + } + + public async captureSent(teamId: number, isTest: boolean, recipientCount: number): Promise<void> { + await this.captureEvent(teamId, 'workflows sandbox email sent', { + recipient_count: recipientCount, + source: isTest ? 'test' : 'workflow', + is_test: isTest, + }) + } + + public async captureBlocked( + teamId: number, + isTest: boolean, + reason: SandboxEmailBlockReason, + blockedRecipientCount: number + ): Promise<void> { + await this.captureEvent( + teamId, + 'workflows sandbox email blocked', + { reason, blocked_recipient_count: blockedRecipientCount, is_test: isTest }, + () => this.isFirstBlockInWindow(teamId, isTest, reason) + ) + } + + private async captureEvent( + teamId: number, + event: string, + properties: Record<string, string | number | boolean>, + storeCheck?: () => Promise<boolean> + ): Promise<void> { + let timer: NodeJS.Timeout | undefined + const deadline = new Promise<void>((resolve) => { + timer = setTimeout(resolve, EVENT_CAPTURE_TIMEOUT_MS) + }) + const captureNow = (): Promise<void> => this.captureEventNow(teamId, event, properties, storeCheck) + const capture = storeCheck ? this.withStoreSlot(deadline, captureNow) : captureNow() + try { + if ((await Promise.race([capture.then(() => true), deadline.then(() => false)])) === false) { + logger.warn('Gave up capturing a sandbox email event: the event store did not answer', { + teamId, + event, + }) + } + } finally { + clearTimeout(timer) + } + } + + private async withStoreSlot(deadline: Promise<void>, work: () => Promise<void>): Promise<void> { + await this.storeSlots.run(deadline, work) + } + + private async captureEventNow( + teamId: number, + event: string, + properties: Record<string, string | number | boolean>, + storeCheck?: () => Promise<boolean> + ): Promise<void> { + try { + if (storeCheck && !(await storeCheck())) { + return + } + const team = await this.teamManager.getTeam(teamId) + if (team) { + captureTeamEvent(team, event, properties) + } + } catch (error) { + logger.warn('Could not capture sandbox email event', { teamId, event, error }) + } + } + + private async isFirstBlockInWindow( + teamId: number, + isTest: boolean, + reason: SandboxEmailBlockReason + ): Promise<boolean> { + const firstBlock = await this.blockedEventRedis.useClient( + { name: 'sandbox-blocked-event', timeout: 1000 }, + (client) => + client.set( + `@posthog/workflows-sandbox-blocked/${teamId}/${reason}/${isTest ? 'test' : 'live'}`, + '1', + 'EX', + BLOCKED_EVENT_WINDOW_SECONDS, + 'NX' + ) + ) + return firstBlock === 'OK' + } +} diff --git a/nodejs/src/cdp/utils/deadline-slots.test.ts b/nodejs/src/cdp/utils/deadline-slots.test.ts new file mode 100644 index 000000000000..4cbb79b1c6ac --- /dev/null +++ b/nodejs/src/cdp/utils/deadline-slots.test.ts @@ -0,0 +1,78 @@ +import { DeadlineSlots } from './deadline-slots' + +const settleSoon = (): Promise<void> => new Promise((resolve) => setImmediate(resolve)) + +describe('DeadlineSlots', () => { + it('runs at most its size at once and hands a freed slot to the next waiter', async () => { + const slots = new DeadlineSlots(2) + const finish: (() => void)[] = [] + const started: number[] = [] + const never = new Promise<void>(() => {}) + const runs = [0, 1, 2, 3].map((i) => + slots.run(never, () => { + started.push(i) + return new Promise<void>((resolve) => finish.push(resolve)) + }) + ) + + await settleSoon() + expect(started).toEqual([0, 1]) + + finish[0]() + await settleSoon() + expect(started).toEqual([0, 1, 2]) + + finish[1]() + finish[2]() + await settleSoon() + finish[3]() + await expect(Promise.all(runs)).resolves.toEqual([true, true, true, true]) + }) + + it('gives up without running when the deadline settles before a slot frees, and leaves no slot taken', async () => { + const slots = new DeadlineSlots(1) + let releaseFirst!: () => void + const first = slots.run( + new Promise<void>(() => {}), + () => new Promise<void>((resolve) => (releaseFirst = resolve)) + ) + let expire!: () => void + const deadline = new Promise<void>((resolve) => (expire = resolve)) + const late = jest.fn(() => Promise.resolve()) + const waiting = slots.run(deadline, late) + + expire() + await expect(waiting).resolves.toBe(false) + expect(late).not.toHaveBeenCalled() + + releaseFirst() + await expect(first).resolves.toBe(true) + const next = jest.fn(() => Promise.resolve()) + await expect(slots.run(new Promise<void>(() => {}), next)).resolves.toBe(true) + expect(next).toHaveBeenCalled() + }) + + it('keeps the slot of a running call past its deadline until the call settles', async () => { + const slots = new DeadlineSlots(1) + let expire!: () => void + let releaseStalled!: () => void + const stalled = slots.run( + new Promise<void>((resolve) => (expire = resolve)), + () => new Promise<void>((resolve) => (releaseStalled = resolve)) + ) + await settleSoon() + expire() + + let expireWaiter!: () => void + const waiter = jest.fn(() => Promise.resolve()) + const waiting = slots.run(new Promise<void>((resolve) => (expireWaiter = resolve)), waiter) + await settleSoon() + expect(waiter).not.toHaveBeenCalled() + + releaseStalled() + await stalled + await expect(waiting).resolves.toBe(true) + expect(waiter).toHaveBeenCalled() + expireWaiter() + }) +}) diff --git a/nodejs/src/cdp/utils/deadline-slots.ts b/nodejs/src/cdp/utils/deadline-slots.ts new file mode 100644 index 000000000000..753fbad3f629 --- /dev/null +++ b/nodejs/src/cdp/utils/deadline-slots.ts @@ -0,0 +1,43 @@ +// At most `size` calls run at once. A caller that finds every slot taken waits, holding nothing, and gives up +// when its deadline settles first. A running call keeps its slot until it settles, even past its deadline, +// because nothing can cancel it; a freed slot goes straight to the next waiter. +export class DeadlineSlots { + private open = 0 + private waiters: (() => void)[] = [] + + constructor(private readonly size: number) {} + + // Resolves true once `work` ran, or false without running it when `deadline` settled before a slot freed. + public async run(deadline: Promise<void>, work: () => Promise<void>): Promise<boolean> { + if (this.open < this.size) { + this.open++ + } else if (!(await this.waitForSlot(deadline))) { + return false + } + try { + await work() + return true + } finally { + const next = this.waiters.shift() + if (next) { + next() + } else { + this.open-- + } + } + } + + private waitForSlot(deadline: Promise<void>): Promise<boolean> { + return new Promise<boolean>((resolve) => { + const waiter = (): void => resolve(true) + this.waiters.push(waiter) + void deadline.then(() => { + const index = this.waiters.indexOf(waiter) + if (index !== -1) { + this.waiters.splice(index, 1) + resolve(false) + } + }) + }) + } +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7ee7092d65c3..72c325da98f5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1804,6 +1804,9 @@ importers: p-limit: specifier: 3.1.0 version: 3.1.0 + parse5: + specifier: 7.1.2 + version: 7.1.2 pg: specifier: ^8.6.0 version: 8.10.0