diff --git a/projects/hutch/src/infra/index.ts b/projects/hutch/src/infra/index.ts index 8f19ebf82..82983319a 100644 --- a/projects/hutch/src/infra/index.ts +++ b/projects/hutch/src/infra/index.ts @@ -6,9 +6,7 @@ import { HutchLambda, HutchAPIGateway, HutchDynamoDBAccess, HutchEventBus, Hutch import { BLOG_SITE_LOG_GROUP, CancelSubscriptionCommand, - CrawlEmailLinkPreview, DeleteAccountCommand, - EmailReceivedEvent, ExportUserDataCommand, SendTrialFeedbackEmailCommand, ReaderViewLoadingSucceeded, @@ -51,12 +49,6 @@ const alertEmail = config.require("alertEmail"); const rawEmailBucketName = config.require("rawEmailBucketName"); const inboxMailParentZone = config.require("inboxMailParentZone"); -// The content-media CDN over the shared article-content bucket is owned by save-link; -// the link-preview crawler uploads lead images there and needs the CDN base URL. That -// URL is the CDN's custom domain (a config constant), so derive it here the way save-link -// does instead of a cross-stack requireOutput, which would couple deploy order (see the -// infrastructure-design skill). -const imagesCdnBaseUrl = `https://${config.require("contentMediaCdnDomain")}`; const tableNames = { articles: config.require("dynamodbArticlesTable"), userArticles: config.require("dynamodbUserArticlesTable"), @@ -186,8 +178,6 @@ const dynamodb = new HutchDynamoDBAccess("hutch-dynamodb-access", { { arn: storage.pendingSignupsTable.arn, includeIndexes: false }, { arn: storage.importSessionsTable.arn, includeIndexes: false }, { arn: storage.inboxAddressesTable.arn, includeIndexes: true }, - { arn: storage.inboxEmailsTable.arn, includeIndexes: false }, - { arn: storage.inboxEmailLinksTable.arn, includeIndexes: false }, { arn: storage.subscriptionProvidersTable.arn, includeIndexes: true }, { arn: storage.onboardingTable.arn, includeIndexes: false }, { arn: storage.rateLimitsTable.arn, includeIndexes: false }, @@ -322,8 +312,6 @@ const lambda = new HutchLambda(LAMBDA_NAMES.hutchHandler, { DYNAMODB_PENDING_SIGNUPS_TABLE: storage.pendingSignupsTable.name, DYNAMODB_IMPORT_SESSIONS_TABLE: storage.importSessionsTable.name, DYNAMODB_INBOX_ADDRESSES_TABLE: storage.inboxAddressesTable.name, - DYNAMODB_INBOX_EMAILS_TABLE: storage.inboxEmailsTable.name, - DYNAMODB_INBOX_EMAIL_LINKS_TABLE: storage.inboxEmailLinksTable.name, INBOX_ADDRESS_DOMAIN: inboxAddressDomain, DYNAMODB_SUBSCRIPTION_PROVIDERS_TABLE: storage.subscriptionProvidersTable.name, DYNAMODB_ONBOARDING_TABLE: storage.onboardingTable.name, @@ -645,257 +633,6 @@ const deleteAccountLambdaWithSQS = new HutchSQSBackedLambda("delete-account", { eventBus.subscribe(DeleteAccountCommand, deleteAccountLambdaWithSQS); -// --- Inbound email receive worker Lambda --- -// Drains the SES→SNS receipt notifications: fetches the raw .eml from the raw -// bucket, resolves each recipient, parses + sanitizes the body into the content -// bucket, writes a row per recipient, and publishes EmailReceivedEvent. Expected -// catch-all-MX conditions (unknown or disabled recipient) record an audit row and -// ACK — never paging. Oversize / unparseable mail also records an audit row, but -// only pages (fails to the DLQ so the HutchSQSBackedLambda alarm fires) when a -// real, enabled recipient is affected; junk to guessed addresses just ACKs. The -// immutable raw .eml is kept forever as the audit trail (★15 degrade-with-alert). -const receiveEmailDynamodb = new HutchDynamoDBAccess("receive-email-dynamodb", { - tables: [ - { arn: storage.inboxEmailsTable.arn, includeIndexes: false }, - { arn: storage.inboxAddressesTable.arn, includeIndexes: false }, - ], - // findByAddress is a GetItem and putEmail a (conditional) PutItem — no Query. - actions: ["dynamodb:GetItem", "dynamodb:PutItem"], -}); - -const receiveEmailQueue = new HutchSQS("receive-email", { - // Matches the worker timeout so an in-flight parse cannot be redelivered. - visibilityTimeoutSeconds: 120, -}); - -const receiveEmailLambda = new HutchLambda("receive-email", { - entryPoint: "./src/runtime/receive-email.main.ts", - outputDir: ".lib/receive-email", - assetDir: "./src/runtime", - memorySize: 1024, - timeout: 120, - environment: { - DYNAMODB_INBOX_EMAILS_TABLE: storage.inboxEmailsTable.name, - DYNAMODB_INBOX_ADDRESSES_TABLE: storage.inboxAddressesTable.name, - RAW_EMAIL_BUCKET_NAME: rawEmailBucketName, - CONTENT_BUCKET_NAME: contentBucketName, - EVENT_BUS_NAME: eventBus.eventBusName, - // 20 MiB — half SES's ~40 MB hard inbound cap; bounds parse memory. - INBOX_MAX_EMAIL_BYTES: String(20 * 1024 * 1024), - }, - policies: [ - ...receiveEmailDynamodb.policies, - // Reads the raw bucket (the .eml) and only writes the content bucket (the - // sanitized body) — no content-bucket read. - ...HutchS3ReadWrite.readPoliciesForBucket("receive-email-raw-read", rawEmailBucketName), - ...HutchS3ReadWrite.writePoliciesForBucket("receive-email-content-write", contentBucketName), - ], -}); - -eventBus.grantPublish(receiveEmailLambda); - -new HutchSQSBackedLambda("receive-email", { - lambda: receiveEmailLambda, - queue: receiveEmailQueue, - alertEmailDLQEntry: alertEmail, - batchSize: 1, -}); - -// SES → SNS → SQS bridge: SES cannot target SQS directly, so the receipt rule -// publishes to inboxMail's topic and the receive queue subscribes here with raw -// message delivery, so the SQS body is the SES notification JSON the handler -// parses. The queue policy lets only this topic enqueue. -const receiveEmailQueuePolicy = new aws.sqs.QueuePolicy("receive-email-sns-policy", { - queueUrl: receiveEmailQueue.queueUrl, - policy: pulumi - .all([receiveEmailQueue.queueArn, inboxMail.notificationTopicArn]) - .apply(([queueArn, topicArn]) => - JSON.stringify({ - Version: "2012-10-17", - Statement: [ - { - Effect: "Allow", - Principal: { Service: "sns.amazonaws.com" }, - Action: "sqs:SendMessage", - Resource: queueArn, - Condition: { ArnEquals: { "aws:SourceArn": topicArn } }, - }, - ], - }), - ), -}); - -new aws.sns.TopicSubscription( - "receive-email-subscription", - { - topic: inboxMail.notificationTopicArn, - protocol: "sqs", - endpoint: receiveEmailQueue.queueArn, - rawMessageDelivery: true, - }, - { dependsOn: [receiveEmailQueuePolicy] }, -); - -// --- Inbox link previews (extract → crawl, M3) --- -// EmailReceivedEvent → extract-email-links re-derives the body from the raw .eml, -// extracts links, writes pending rows, and fans out one CrawlEmailLinkPreview per -// link (★14 cap + truncate-degrade-with-dedicated-alert-queue). Each CrawlEmailLinkPreview → -// crawl-email-link-preview crawls a preview WITHOUT saving to /queue (★16 SSRF -// guard inherited from crawlAndFinalize). Neither Lambda is granted the -// articles/user-articles tables — the "nothing saved to the queue" invariant is -// enforced at the IAM boundary. -const extractEmailLinksDynamodb = new HutchDynamoDBAccess("extract-email-links-dynamodb", { - tables: [ - { arn: storage.inboxEmailsTable.arn, includeIndexes: false }, - { arn: storage.inboxEmailLinksTable.arn, includeIndexes: false }, - ], - // getEmail (GetItem); putLink/putLinksMeta (PutItem); the conditional put needs - // no Query, and listing is the web layer's job. - actions: ["dynamodb:GetItem", "dynamodb:PutItem"], -}); - -const extractEmailLinksQueue = new HutchSQS("extract-email-links", { - visibilityTimeoutSeconds: 120, -}); - -// Truncation is a successful degradation (the first N previews still shipped), not -// a processing failure — so it must NOT land in the consumer's own failure DLQ. -// There it would (a) make the DLQ depth alarm ambiguous between a genuine "email -// never extracted" and a benign "email truncated", and (b) re-enter the source -// queue on redrive and bounce straight back (the synthetic body has no `.detail`). -// It gets a dedicated sink whose send-rate alarm → SNS email instead, so the -// failure DLQ's "messages awaiting redrive" contract stays unambiguous. -const truncationAlertQueue = new aws.sqs.Queue("extract-email-links-truncated-alert", { - name: "extract-email-links-truncated-alert", - // Nothing consumes this queue — it is an audit trail of truncated emails. The - // alarm below fires on send RATE, not depth, so these retained messages no - // longer pin it in ALARM; 14 days is just how long the audit trail survives. - messageRetentionSeconds: 1209600, -}); - -const truncationAlertTopic = new aws.sns.Topic("extract-email-links-truncated-alert-topic", { - name: "extract-email-links-truncated-alert-topic", -}); - -new aws.sns.TopicSubscription("extract-email-links-truncated-alert-email", { - topic: truncationAlertTopic.arn, - protocol: "email", - endpoint: alertEmail, -}); - -// Alarm on the SEND RATE (a count metric), not queue DEPTH (a gauge). Nothing -// consumes this queue, so an ApproximateNumberOfMessagesVisible>=1 depth alarm -// would stay in ALARM until the message expires 14 days later (or an operator -// purges it), and a second truncation while already in ALARM would never -// re-notify. NumberOfMessagesSent has a datapoint only in the period a message is -// enqueued, so with treatMissingData "notBreaching" the alarm fires on each -// truncation burst and auto-resets the next empty period — re-firing for the next. -new aws.cloudwatch.MetricAlarm("extract-email-links-truncated-alarm", { - name: "extract-email-links-truncated-alarm", - comparisonOperator: "GreaterThanOrEqualToThreshold", - evaluationPeriods: 1, - metricName: "NumberOfMessagesSent", - namespace: "AWS/SQS", - period: 300, - statistic: "Sum", - threshold: 1, - treatMissingData: "notBreaching", - alarmDescription: - "extract-email-links hit the per-email link cap and truncated an email (the first N previews still shipped)", - dimensions: { QueueName: truncationAlertQueue.name }, - alarmActions: [truncationAlertTopic.arn], -}); - -const extractEmailLinksLambda = new HutchLambda("extract-email-links", { - entryPoint: "./src/runtime/extract-email-links.main.ts", - outputDir: ".lib/extract-email-links", - assetDir: "./src/runtime", - memorySize: 1024, - timeout: 120, - environment: { - DYNAMODB_INBOX_EMAILS_TABLE: storage.inboxEmailsTable.name, - DYNAMODB_INBOX_EMAIL_LINKS_TABLE: storage.inboxEmailLinksTable.name, - RAW_EMAIL_BUCKET_NAME: rawEmailBucketName, - EVENT_BUS_NAME: eventBus.eventBusName, - // A typical newsletter has < 30 links; 200 is generous headroom before the - // per-email cap truncates and the working path still ships the first 200. - INBOX_MAX_LINKS_PER_EMAIL: String(200), - EXTRACT_LINKS_TRUNCATION_ALERT_QUEUE_URL: truncationAlertQueue.url, - }, - policies: [ - ...extractEmailLinksDynamodb.policies, - // Reads the raw .eml to re-derive the body; never writes any bucket. - ...HutchS3ReadWrite.readPoliciesForBucket("extract-email-links-raw-read", rawEmailBucketName), - // The truncated-degrade alert is an explicit SendMessage to the dedicated - // alert queue (NOT the failure DLQ), whose own send-rate alarm pages the operator. - { - name: "extract-email-links-truncated-alert-send-pol", - policy: pulumi.output(truncationAlertQueue.arn).apply((arn) => - JSON.stringify({ - Version: "2012-10-17", - Statement: [{ Effect: "Allow", Action: ["sqs:SendMessage"], Resource: [arn] }], - }), - ), - }, - ], -}); - -eventBus.grantPublish(extractEmailLinksLambda); - -const extractEmailLinksWithSQS = new HutchSQSBackedLambda("extract-email-links", { - lambda: extractEmailLinksLambda, - queue: extractEmailLinksQueue, - alertEmailDLQEntry: alertEmail, - batchSize: 1, -}); - -eventBus.subscribe(EmailReceivedEvent, extractEmailLinksWithSQS); - -const crawlEmailLinkPreviewDynamodb = new HutchDynamoDBAccess("crawl-email-link-preview-dynamodb", { - tables: [{ arn: storage.inboxEmailLinksTable.arn, includeIndexes: false }], - // setLinkOutcome is an UpdateItem; getLink is not used by the worker. - actions: ["dynamodb:UpdateItem"], -}); - -const crawlEmailLinkPreviewQueue = new HutchSQS("crawl-email-link-preview", { - // A single dead/slow origin retries and DLQs in isolation; matches the worker - // timeout so an in-flight crawl cannot be redelivered. - visibilityTimeoutSeconds: 120, -}); - -const crawlEmailLinkPreviewLambda = new HutchLambda("crawl-email-link-preview", { - entryPoint: "./src/runtime/crawl-email-link-preview.main.ts", - outputDir: ".lib/crawl-email-link-preview", - assetDir: "./src/runtime", - memorySize: 1024, - timeout: 120, - environment: { - DYNAMODB_INBOX_EMAIL_LINKS_TABLE: storage.inboxEmailLinksTable.name, - // Uploads each lead image to the shared content bucket, fronted by the - // save-link content-media CDN (read cross-stack above). - CONTENT_BUCKET_NAME: contentBucketName, - IMAGES_CDN_BASE_URL: imagesCdnBaseUrl, - }, - policies: [ - ...crawlEmailLinkPreviewDynamodb.policies, - // Only writes the content bucket (the lead-image thumbnail) — no read, and - // no access to the articles/user-articles tables (nothing saved to /queue). - ...HutchS3ReadWrite.writePoliciesForBucket( - "crawl-email-link-preview-content-write", - contentBucketName, - ), - ], -}); - -const crawlEmailLinkPreviewWithSQS = new HutchSQSBackedLambda("crawl-email-link-preview", { - lambda: crawlEmailLinkPreviewLambda, - queue: crawlEmailLinkPreviewQueue, - alertEmailDLQEntry: alertEmail, - batchSize: 1, -}); - -eventBus.subscribe(CrawlEmailLinkPreview, crawlEmailLinkPreviewWithSQS); - // --- Reader-ready digest (fan-out + 6h flush + send) --- // When an article's clean reader view reaches the successful terminal state, // save-link publishes ReaderViewLoadingSucceeded (per-URL, global). The fan-out diff --git a/projects/hutch/src/runtime/app.ts b/projects/hutch/src/runtime/app.ts index 89bfd90ee..14f8b1401 100644 --- a/projects/hutch/src/runtime/app.ts +++ b/projects/hutch/src/runtime/app.ts @@ -92,7 +92,6 @@ import { initInMemoryPendingPdf } from "@packages/test-fixtures/providers/pendin import { initInMemoryImportSession } from "@packages/test-fixtures/providers/import-session"; import { initDynamoDbImportSession } from "./providers/import-session/dynamodb-import-session"; import { initInMemoryInboxAddress } from "@packages/test-fixtures/providers/inbox-address"; -import { initInMemoryInboxEmail, initInMemoryInboxEmailLink } from "@packages/test-fixtures/providers/inbox-email"; import { initExchangeGoogleCode } from "./providers/google-auth/google-token"; import { initExchangeAppleCode } from "./providers/apple-auth/apple-token"; import { initCreateAppleClientSecret } from "./providers/apple-auth/apple-client-secret"; @@ -116,7 +115,9 @@ import { httpErrorMessageMapping } from "./web/pages/queue/queue.error"; import { initFoundingAllocation } from "./web/shared/founding-progress/founding-allocation"; import { initCachedUserCount } from "./web/auth/cached-user-count"; import { getEnv, requireEnv } from "@packages/require-env"; -import { initDynamoDbInboxAddress, initDynamoDbInboxEmail, initDynamoDbInboxEmailLink } from "@packages/inbox-store"; +import { initDynamoDbInboxAddress } from "@packages/inbox-store"; +import { DEFAULT_INBOX_ALIAS } from "@packages/domain/inbox"; +import type { UserId } from "@packages/domain/user"; /** * Hutch SSR does not run PDF extraction in-process — the @@ -175,8 +176,6 @@ function initProviders() { const pendingPdfBucketName = requireEnv("PENDING_PDF_BUCKET_NAME"); const importSessionsTable = requireEnv("DYNAMODB_IMPORT_SESSIONS_TABLE"); const inboxAddressesTable = requireEnv("DYNAMODB_INBOX_ADDRESSES_TABLE"); - const inboxEmailsTable = requireEnv("DYNAMODB_INBOX_EMAILS_TABLE"); - const inboxEmailLinksTable = requireEnv("DYNAMODB_INBOX_EMAIL_LINKS_TABLE"); const inboxAddressDomain = requireEnv("INBOX_ADDRESS_DOMAIN"); const subscriptionProvidersTable = requireEnv("DYNAMODB_SUBSCRIPTION_PROVIDERS_TABLE"); const onboardingTable = requireEnv("DYNAMODB_ONBOARDING_TABLE"); @@ -318,12 +317,6 @@ function initProviders() { tableName: inboxAddressesTable, now: () => new Date(), }); - const inboxEmailStore = initDynamoDbInboxEmail({ client, tableName: inboxEmailsTable }); - const inboxEmailLinkStore = initDynamoDbInboxEmailLink({ client, tableName: inboxEmailLinksTable }); - const readEmailContent = initS3ReadContent({ - send: (cmd) => s3Client.send(cmd), - bucketName: contentBucketName, - }); const { consumeRateLimit } = initDynamoDbRateLimit({ client, tableName: rateLimitsTable, @@ -350,11 +343,13 @@ function initProviders() { readArticleContent, importSessionStore, extractLinksFromPageUrl, - inboxAddressStore, - inboxEmailStore, - inboxEmailLinkStore, - readEmailContent, - inboxAddressDomain, + provisionInboxAddress: async (userId: UserId) => { + await inboxAddressStore.createAddress({ + userId, + domain: inboxAddressDomain, + name: DEFAULT_INBOX_ALIAS, + }); + }, subscriptionProviders, trialScheduler, createSubscriptionOnExistingCustomer: stripeSubscriptions.createSubscriptionOnExistingCustomer, @@ -584,8 +579,6 @@ function initProviders() { const importSessionStore = initInMemoryImportSession({ now: () => new Date() }); const inboxAddressStore = initInMemoryInboxAddress({ now: () => new Date() }); - const inboxEmailStore = initInMemoryInboxEmail(); - const inboxEmailLinkStore = initInMemoryInboxEmailLink(); const inboxAddressDomain = requireEnv("INBOX_ADDRESS_DOMAIN"); // In-process counters are valid here because dev runs a single long-lived @@ -613,11 +606,13 @@ function initProviders() { }), importSessionStore, extractLinksFromPageUrl, - inboxAddressStore, - inboxEmailStore, - inboxEmailLinkStore, - readEmailContent: articleStore.readContent, - inboxAddressDomain, + provisionInboxAddress: async (userId: UserId) => { + await inboxAddressStore.createAddress({ + userId, + domain: inboxAddressDomain, + name: DEFAULT_INBOX_ALIAS, + }); + }, subscriptionProviders: devSubscriptionProviders, trialScheduler: devTrialScheduler, createSubscriptionOnExistingCustomer: devStripeSubscriptions.createSubscriptionOnExistingCustomer, diff --git a/projects/hutch/src/runtime/crawl-email-link-preview.main.ts b/projects/hutch/src/runtime/crawl-email-link-preview.main.ts deleted file mode 100644 index 2a6941adf..000000000 --- a/projects/hutch/src/runtime/crawl-email-link-preview.main.ts +++ /dev/null @@ -1,93 +0,0 @@ -import { S3Client } from "@aws-sdk/client-s3"; -import { - initReadabilityParser, - linkedinSiteRules, - mediumSiteRules, - theInformationSiteRules, -} from "@packages/article-parser"; -import { - CRAWL_PERSONAS, - initCrawlArticle, - initCrawlFetch, - initFetchThumbnailImage, - initXTwitterSiteRules, -} from "@packages/crawl-article"; -import { isBlockedIpAddress } from "@packages/domain/article"; -import { initCrawlAndFinalizeArticle, initFinalizeArticle } from "@packages/finalize-article"; -import { HutchLogger, consoleLogger } from "@packages/hutch-logger"; -import { createDynamoDocumentClient } from "@packages/hutch-storage-client"; -import { requireEnv } from "@packages/require-env"; -import { initCrawlEmailLinkPreviewHandler } from "./domain/inbox/crawl-email-link-preview-handler"; -import { initS3PutImageObject } from "./providers/article-store/s3-put-image-object"; -import { initDynamoDbInboxEmailLink } from "@packages/inbox-store"; - -const inboxEmailLinksTable = requireEnv("DYNAMODB_INBOX_EMAIL_LINKS_TABLE"); -const contentBucketName = requireEnv("CONTENT_BUCKET_NAME"); -const imagesCdnBaseUrl = requireEnv("IMAGES_CDN_BASE_URL"); - -const s3Client = new S3Client({}); -const dynamoClient = createDynamoDocumentClient(); -const logger = HutchLogger.from(consoleLogger); -const logError = (message: string, error?: Error) => - logger.error( - JSON.stringify({ - level: "ERROR", - timestamp: new Date().toISOString(), - message, - stack: error?.stack, - }), - ); -const logInfo = (message: string) => - logger.info( - JSON.stringify({ - level: "INFO", - timestamp: new Date().toISOString(), - message, - }), - ); - -// The same SSRF-guarded crawlFetch the save pipeline uses: every connect and -// redirect hop runs isBlockedIpAddress, so a link that DNS-rebinds to a private -// or metadata address is refused at connect time. -const crawlFetch = initCrawlFetch({ - fetch: globalThis.fetch, - personas: CRAWL_PERSONAS, - isBlocked: isBlockedIpAddress, -}); -const siteRules = [ - theInformationSiteRules, - mediumSiteRules, - linkedinSiteRules, - initXTwitterSiteRules({ crawlFetch, logError }), -]; -const crawlArticle = initCrawlArticle({ crawlFetch, siteRules, logError, logInfo }); -const { parseHtml } = initReadabilityParser({ - crawlArticle, - siteRules, - logError, -}); -const fetchThumbnailImage = initFetchThumbnailImage({ crawlFetch, logError, logInfo }); -const { putImageObject } = initS3PutImageObject({ client: s3Client, bucketName: contentBucketName }); -// A preview keeps only the article metadata and discards the body, so media -// rewriting is a no-op here; the lead-image upload to the content-bucket CDN is -// the only media side effect a preview needs. -const finalizeArticle = initFinalizeArticle({ - parseHtml, - downloadMedia: async () => [], - processContent: async ({ html }) => html, - fetchThumbnailImage, - putImageObject, - imagesCdnBaseUrl, -}); -const crawlAndFinalize = initCrawlAndFinalizeArticle({ crawlArticle, finalizeArticle }); - -const inboxEmailLinkStore = initDynamoDbInboxEmailLink({ - client: dynamoClient, - tableName: inboxEmailLinksTable, -}); - -export const handler = initCrawlEmailLinkPreviewHandler({ - crawlAndFinalize, - setLinkOutcome: inboxEmailLinkStore.setLinkOutcome, - logger, -}); diff --git a/projects/hutch/src/runtime/domain/inbox/crawl-email-link-preview-handler.test.ts b/projects/hutch/src/runtime/domain/inbox/crawl-email-link-preview-handler.test.ts deleted file mode 100644 index e61a78bf2..000000000 --- a/projects/hutch/src/runtime/domain/inbox/crawl-email-link-preview-handler.test.ts +++ /dev/null @@ -1,224 +0,0 @@ -import assert from "node:assert/strict"; -import type { CrawlArticle } from "@packages/crawl-article"; -import { - type CrawlAndFinalizeArticle, - type CrawlAndFinalizeResult, - type FinalizeArticle, - initCrawlAndFinalizeArticle, -} from "@packages/finalize-article"; -import { HutchLogger, noopLogger } from "@packages/hutch-logger"; -import { EmailLinkOrdinalSchema, type InboxEmailLinkStore } from "@packages/domain/inbox"; -import { UserIdSchema } from "@packages/domain/user"; -import { buildLambdaContext } from "@packages/test-fixtures/lambda-context"; -import { initInMemoryInboxEmailLink } from "@packages/test-fixtures/providers/inbox-email"; -import { buildSqsEvent } from "@packages/test-fixtures/sqs"; -import { initCrawlEmailLinkPreviewHandler } from "./crawl-email-link-preview-handler"; - -const USER = UserIdSchema.parse("00000000000000000000000000000001"); -const RAM = "2026-06-24T09:00:00.000Z#"; - -function commandBody(over: Partial<{ url: string; ordinal: string }> = {}): string { - return JSON.stringify({ - detail: { - userId: USER, - receivedAtMessageId: RAM, - ordinal: over.ordinal ?? "0000", - url: over.url ?? "https://example.com/post", - }, - }); -} - -function fetched( - metadata: { title: string; siteName: string; excerpt: string; imageUrl?: string }, -): CrawlAndFinalizeResult { - return { - status: "fetched", - bodyHash: "hash", - article: { - html: "

body that is discarded

", - metadata: { - wordCount: 100, - estimatedReadTime: 1, - ...metadata, - }, - }, - }; -} - -async function seedPending(store: InboxEmailLinkStore, ordinal = "0000") { - await store.putLink({ - userId: USER, - receivedAtMessageId: RAM, - ordinal: EmailLinkOrdinalSchema.parse(ordinal), - url: "https://example.com/post", - status: "pending", - title: undefined, - excerpt: undefined, - siteName: undefined, - imageUrl: undefined, - failureReason: undefined, - }); -} - -function makeHandler(deps: { - crawlAndFinalize: CrawlAndFinalizeArticle; - setLinkOutcome: InboxEmailLinkStore["setLinkOutcome"]; -}) { - const handler = initCrawlEmailLinkPreviewHandler({ - crawlAndFinalize: deps.crawlAndFinalize, - setLinkOutcome: deps.setLinkOutcome, - logger: HutchLogger.from(noopLogger), - }); - return (body: string) => handler(buildSqsEvent([{ messageId: "rec-1", body }]), buildLambdaContext(), () => {}); -} - -async function getLink(store: InboxEmailLinkStore, ordinal = "0000") { - const link = await store.getLink({ - userId: USER, - receivedAtMessageId: RAM, - ordinal: EmailLinkOrdinalSchema.parse(ordinal), - }); - assert(link, "expected the seeded link to resolve"); - return link; -} - -describe("initCrawlEmailLinkPreviewHandler", () => { - it("stamps a crawled preview from the fetched metadata and discards the body", async () => { - const store = initInMemoryInboxEmailLink(); - await seedPending(store); - const run = makeHandler({ - crawlAndFinalize: async () => - fetched({ - title: "A title", - siteName: "Example", - excerpt: "An excerpt", - imageUrl: "https://cdn.test/x.jpg", - }), - setLinkOutcome: store.setLinkOutcome, - }); - - const result = await run(commandBody()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - const link = await getLink(store); - expect(link.status).toBe("crawled"); - expect(link.title).toBe("A title"); - expect(link.siteName).toBe("Example"); - expect(link.excerpt).toBe("An excerpt"); - expect(link.imageUrl).toBe("https://cdn.test/x.jpg"); - }); - - it("maps a failed crawl to a failed preview and ACKs the record", async () => { - const store = initInMemoryInboxEmailLink(); - await seedPending(store); - const run = makeHandler({ - crawlAndFinalize: async () => ({ status: "failed", reason: "crawl-failed" }), - setLinkOutcome: store.setLinkOutcome, - }); - - const result = await run(commandBody()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - const link = await getLink(store); - expect(link.status).toBe("failed"); - expect(link.failureReason).toBe("crawl-failed"); - }); - - it("maps an unsupported crawl to a failed preview", async () => { - const store = initInMemoryInboxEmailLink(); - await seedPending(store); - const run = makeHandler({ - crawlAndFinalize: async () => ({ status: "unsupported", reason: "pdf" }), - setLinkOutcome: store.setLinkOutcome, - }); - - await run(commandBody()); - - expect((await getLink(store)).failureReason).toBe("pdf"); - }); - - it("maps a not-modified result to a failed preview for totality", async () => { - const store = initInMemoryInboxEmailLink(); - await seedPending(store); - const run = makeHandler({ - crawlAndFinalize: async () => ({ status: "not-modified" }), - setLinkOutcome: store.setLinkOutcome, - }); - - await run(commandBody()); - - expect((await getLink(store)).failureReason).toBe("not-modified"); - }); - - it("maps a permanently-dead (404) link to a failed preview with the not-found reason and ACKs the record", async () => { - const store = initInMemoryInboxEmailLink(); - await seedPending(store); - const run = makeHandler({ - crawlAndFinalize: async () => ({ status: "not-found", httpStatus: 404 }), - setLinkOutcome: store.setLinkOutcome, - }); - - const result = await run(commandBody()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - const link = await getLink(store); - expect(link.status).toBe("failed"); - expect(link.failureReason).toBe("not-found"); - }); - - it("fails the record for a malformed command envelope", async () => { - const store = initInMemoryInboxEmailLink(); - const run = makeHandler({ - crawlAndFinalize: async () => fetched({ title: "x", siteName: "y", excerpt: "z" }), - setLinkOutcome: store.setLinkOutcome, - }); - - const result = await run(JSON.stringify({ detail: { wrong: "shape" } })); - - assert(result); - expect(result.batchItemFailures).toEqual([{ itemIdentifier: "rec-1" }]); - }); - - it("fails the record when the outcome write throws", async () => { - const run = makeHandler({ - crawlAndFinalize: async () => fetched({ title: "x", siteName: "y", excerpt: "z" }), - setLinkOutcome: async () => { - throw new Error("ddb unavailable"); - }, - }); - - const result = await run(commandBody()); - - assert(result); - expect(result.batchItemFailures).toEqual([{ itemIdentifier: "rec-1" }]); - }); - - it("blocks an SSRF URL via the inherited guard, never fetching it (unsafe-url)", async () => { - const store = initInMemoryInboxEmailLink(); - await seedPending(store); - let crawlCalled = false; - const crawlArticle: CrawlArticle = async () => { - crawlCalled = true; - throw new Error("must not fetch an SSRF URL"); - }; - const finalizeArticle: FinalizeArticle = async () => { - throw new Error("must not finalize an SSRF URL"); - }; - const run = makeHandler({ - crawlAndFinalize: initCrawlAndFinalizeArticle({ crawlArticle, finalizeArticle }), - setLinkOutcome: store.setLinkOutcome, - }); - - const result = await run(commandBody({ url: "http://169.254.169.254/latest/meta-data" })); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - expect(crawlCalled).toBe(false); - const link = await getLink(store); - expect(link.status).toBe("failed"); - expect(link.failureReason).toBe("unsafe-url"); - }); -}); diff --git a/projects/hutch/src/runtime/domain/inbox/crawl-email-link-preview-handler.ts b/projects/hutch/src/runtime/domain/inbox/crawl-email-link-preview-handler.ts deleted file mode 100644 index 871bd108a..000000000 --- a/projects/hutch/src/runtime/domain/inbox/crawl-email-link-preview-handler.ts +++ /dev/null @@ -1,92 +0,0 @@ -import type { - Handler, - SQSBatchItemFailure, - SQSBatchResponse, - SQSEvent, -} from "aws-lambda"; -import { CrawlEmailLinkPreview } from "@packages/hutch-infra-components"; -import type { HutchLogger } from "@packages/hutch-logger"; -import { EmailLinkOrdinalSchema, type InboxEmailLinkStore } from "@packages/domain/inbox"; -import { UserIdSchema } from "@packages/domain/user"; -import type { CrawlAndFinalizeArticle } from "@packages/finalize-article"; - -/** - * Consumes one `CrawlEmailLinkPreview` per message and crawls a preview of the - * single URL WITHOUT saving it to the reading queue: it keeps only the metadata - * and discards the article body, so nothing lands in the articles / user-articles - * tables. A dead/blocked/paywalled link is an expected `failed` preview (the SQS - * record is ACKed); only a genuine store-write fault or a malformed envelope - * fails the record to its DLQ. The SSRF guard is inherited from - * `crawlAndFinalize`, which fails closed on localhost/metadata/private-range URLs - * before any network call. - */ -export function initCrawlEmailLinkPreviewHandler(deps: { - crawlAndFinalize: CrawlAndFinalizeArticle; - setLinkOutcome: InboxEmailLinkStore["setLinkOutcome"]; - logger: HutchLogger; -}): Handler { - const { crawlAndFinalize, setLinkOutcome, logger } = deps; - - return async (event): Promise => { - const batchItemFailures: SQSBatchItemFailure[] = []; - - for (const record of event.Records) { - try { - const envelope = JSON.parse(record.body); - const parsed = CrawlEmailLinkPreview.detailSchema.safeParse(envelope.detail); - if (!parsed.success) { - logger.error("[crawl-email-link-preview] malformed command", { - messageId: record.messageId, - }); - batchItemFailures.push({ itemIdentifier: record.messageId }); - continue; - } - const userId = UserIdSchema.parse(parsed.data.userId); - const ordinal = EmailLinkOrdinalSchema.parse(parsed.data.ordinal); - const { receivedAtMessageId, url } = parsed.data; - - const result = await crawlAndFinalize({ url }); - if (result.status === "fetched") { - const { metadata } = result.article; - await setLinkOutcome({ - userId, - receivedAtMessageId, - ordinal, - outcome: { - status: "crawled", - title: metadata.title, - excerpt: metadata.excerpt, - siteName: metadata.siteName, - imageUrl: metadata.imageUrl, - }, - }); - logger.info("[crawl-email-link-preview] crawled", { receivedAtMessageId, ordinal }); - continue; - } - const failureReason = - result.status === "not-modified" || result.status === "not-found" - ? result.status - : result.reason; - await setLinkOutcome({ - userId, - receivedAtMessageId, - ordinal, - outcome: { status: "failed", failureReason }, - }); - logger.info("[crawl-email-link-preview] preview unavailable", { - receivedAtMessageId, - ordinal, - failureReason, - }); - } catch (error) { - logger.error("[crawl-email-link-preview] record failed", { - messageId: record.messageId, - error, - }); - batchItemFailures.push({ itemIdentifier: record.messageId }); - } - } - - return { batchItemFailures }; - }; -} diff --git a/projects/hutch/src/runtime/domain/inbox/email-content-id.test.ts b/projects/hutch/src/runtime/domain/inbox/email-content-id.test.ts deleted file mode 100644 index 2f177394f..000000000 --- a/projects/hutch/src/runtime/domain/inbox/email-content-id.test.ts +++ /dev/null @@ -1,41 +0,0 @@ -import assert from "node:assert/strict"; -import { UserIdSchema } from "@packages/domain/user"; -import { emailContentResourceId } from "./email-content-id"; - -const RECEIVED_AT_MESSAGE_ID = "2026-06-24T09:00:00.000Z#"; - -describe("emailContentResourceId", () => { - it("derives different content keys for two users that share a receivedAtMessageId", () => { - const alice = UserIdSchema.parse("00000000000000000000000000000001"); - const bob = UserIdSchema.parse("00000000000000000000000000000002"); - - const aliceKey = emailContentResourceId({ - userId: alice, - receivedAtMessageId: RECEIVED_AT_MESSAGE_ID, - }).toS3ContentKey(); - const bobKey = emailContentResourceId({ - userId: bob, - receivedAtMessageId: RECEIVED_AT_MESSAGE_ID, - }).toS3ContentKey(); - - // Same sender Message-ID at the same receipt instant must NOT collide across - // users: each body is partitioned by its owner just like the row is. - assert.notEqual(aliceKey, bobKey); - assert.match(aliceKey, /00000000000000000000000000000001/); - }); - - it("is stable for one user and sort key so the read resolves the write's key", () => { - const user = UserIdSchema.parse("00000000000000000000000000000001"); - - const write = emailContentResourceId({ - userId: user, - receivedAtMessageId: RECEIVED_AT_MESSAGE_ID, - }).toS3ContentKey(); - const read = emailContentResourceId({ - userId: user, - receivedAtMessageId: RECEIVED_AT_MESSAGE_ID, - }).toS3ContentKey(); - - assert.equal(write, read); - }); -}); diff --git a/projects/hutch/src/runtime/domain/inbox/email-content-id.ts b/projects/hutch/src/runtime/domain/inbox/email-content-id.ts deleted file mode 100644 index 430fa6f24..000000000 --- a/projects/hutch/src/runtime/domain/inbox/email-content-id.ts +++ /dev/null @@ -1,25 +0,0 @@ -import { ArticleResourceUniqueId } from "@packages/article-resource-unique-id"; -import type { UserId } from "@packages/domain/user"; - -/** - * The resource id a received email's sanitized body is stored under in the - * content bucket. Scoped by the owning `userId` exactly like the `inbox_emails` - * row is partitioned, so two users whose rows happen to share a - * `receivedAtMessageId` (same sender Message-ID at the same receipt instant) - * never read or overwrite each other's private body. Both segments are - * percent-encoded into single `email://inbox//` path segments so the - * sort key's `#`/`<`/`>` characters can't be mistaken for a URL fragment (which - * `normalizeUrl` would drop), keeping the id 1:1 with the row. Shared by the - * receive path (write) and the detail page (read) so both resolve the same S3 - * key. - */ -export function emailContentResourceId(input: { - userId: UserId; - receivedAtMessageId: string; -}): ArticleResourceUniqueId { - return ArticleResourceUniqueId.parse( - `email://inbox/${encodeURIComponent(input.userId)}/${encodeURIComponent( - input.receivedAtMessageId, - )}`, - ); -} diff --git a/projects/hutch/src/runtime/domain/inbox/extract-email-links-handler.test.ts b/projects/hutch/src/runtime/domain/inbox/extract-email-links-handler.test.ts deleted file mode 100644 index 7b4268412..000000000 --- a/projects/hutch/src/runtime/domain/inbox/extract-email-links-handler.test.ts +++ /dev/null @@ -1,215 +0,0 @@ -import assert from "node:assert/strict"; -import { HutchLogger, noopLogger } from "@packages/hutch-logger"; -import { - type EmailLinkOrdinal, - type InboxEmailEntry, - InboxAddressSchema, - MessageIdSchema, - type ParseEmailResult, -} from "@packages/domain/inbox"; -import { type UserId, UserIdSchema } from "@packages/domain/user"; -import { buildLambdaContext } from "@packages/test-fixtures/lambda-context"; -import { initInMemoryInboxEmailLink } from "@packages/test-fixtures/providers/inbox-email"; -import { buildSqsEvent } from "@packages/test-fixtures/sqs"; -import { initExtractEmailLinksHandler } from "./extract-email-links-handler"; - -const USER = UserIdSchema.parse("00000000000000000000000000000001"); -const RECEIVED_AT = "2026-06-24T09:00:00.000Z"; -const RAM = `${RECEIVED_AT}#`; -const RAW_KEY = "inbound/ses-msg-1"; - -function makeEmail(overrides: Partial = {}): InboxEmailEntry { - return { - userId: USER, - receivedAtMessageId: RAM, - messageId: MessageIdSchema.parse(""), - recipientAddress: InboxAddressSchema.parse("in-3f9a2c@read.place"), - senderEmail: "news@example.com", - subject: "Digest", - status: "received", - receivedAt: RECEIVED_AT, - rawEmailS3Key: RAW_KEY, - bodyS3Key: "content/m/content.html", - ...overrides, - }; -} - -function parsedOk(html: string): ParseEmailResult { - return { - ok: true, - email: { - from: "news@example.com", - subject: "Digest", - text: "", - html, - messageId: MessageIdSchema.parse(""), - receivedAt: RECEIVED_AT, - inlineImages: [], - }, - }; -} - -function eventBody(over: Partial<{ userId: string; receivedAtMessageId: string }> = {}): string { - return JSON.stringify({ - detail: { - userId: over.userId ?? USER, - receivedAtMessageId: over.receivedAtMessageId ?? RAM, - recipientAddress: "in-3f9a2c@read.place", - }, - }); -} - -function makeHarness(opts?: { - getEmail?: (input: { userId: UserId; receivedAtMessageId: string }) => Promise; - readRawEmail?: (s3Key: string) => Promise; - parseEmail?: () => Promise; - derivedHtml?: string; - maxLinks?: number; -}) { - const linkStore = initInMemoryInboxEmailLink(); - const published: { ordinal: EmailLinkOrdinal; url: string }[] = []; - const alerts: { found: number }[] = []; - - const handler = initExtractEmailLinksHandler({ - getEmail: opts?.getEmail ?? (async () => makeEmail()), - readRawEmail: opts?.readRawEmail ?? (async () => Buffer.from("raw eml")), - parseEmail: opts?.parseEmail ?? (async () => parsedOk("

body

")), - deriveSanitizedBody: () => opts?.derivedHtml ?? "", - putLink: linkStore.putLink, - putLinksMeta: linkStore.putLinksMeta, - publishCrawlPreview: async ({ ordinal, url }) => { - published.push({ ordinal, url }); - }, - alertTruncated: async ({ found }) => { - alerts.push({ found }); - }, - logger: HutchLogger.from(noopLogger), - maxLinks: opts?.maxLinks ?? 200, - }); - - const run = (body: string) => - handler(buildSqsEvent([{ messageId: "rec-1", body }]), buildLambdaContext(), () => {}); - - return { linkStore, published, alerts, run }; -} - -describe("initExtractEmailLinksHandler", () => { - it("writes one pending row per link and fans out a crawl command for each", async () => { - const harness = makeHarness({ - derivedHtml: "https://a.test/x https://b.test/y https://c.test/z", - }); - - const result = await harness.run(eventBody()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - const { links, meta } = await harness.linkStore.listLinksByEmail({ - userId: USER, - receivedAtMessageId: RAM, - }); - expect(links.map((l) => [l.ordinal, l.url, l.status])).toEqual([ - ["0000", "https://a.test/x", "pending"], - ["0001", "https://b.test/y", "pending"], - ["0002", "https://c.test/z", "pending"], - ]); - // Meta is always written once extraction finishes (the "extraction ran" - // barrier the detail view polls against); only `truncated` differs. - expect(meta).toEqual({ truncated: false }); - expect(harness.published).toEqual([ - { ordinal: "0000", url: "https://a.test/x" }, - { ordinal: "0001", url: "https://b.test/y" }, - { ordinal: "0002", url: "https://c.test/z" }, - ]); - expect(harness.alerts).toHaveLength(0); - }); - - it("caps the fan-out, writes a truncated meta item, and raises one alert", async () => { - const harness = makeHarness({ - derivedHtml: "https://a.test/x https://b.test/y https://c.test/z", - maxLinks: 2, - }); - - const result = await harness.run(eventBody()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - const { links, meta } = await harness.linkStore.listLinksByEmail({ - userId: USER, - receivedAtMessageId: RAM, - }); - expect(links).toHaveLength(2); - expect(meta).toEqual({ truncated: true }); - expect(harness.published).toHaveLength(2); - expect(harness.alerts).toEqual([{ found: 3 }]); - }); - - it("skips an email that is not in the received state", async () => { - const harness = makeHarness({ - getEmail: async () => makeEmail({ status: "unparsed", bodyS3Key: undefined }), - derivedHtml: "https://a.test/x", - }); - - await harness.run(eventBody()); - - const { links } = await harness.linkStore.listLinksByEmail({ - userId: USER, - receivedAtMessageId: RAM, - }); - expect(links).toHaveLength(0); - expect(harness.published).toHaveLength(0); - }); - - it("skips when the email row is not yet visible", async () => { - const harness = makeHarness({ getEmail: async () => undefined, derivedHtml: "https://a.test/x" }); - - const result = await harness.run(eventBody()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - expect(harness.published).toHaveLength(0); - }); - - it("retries when the raw .eml is not yet readable", async () => { - const harness = makeHarness({ readRawEmail: async () => undefined }); - - const result = await harness.run(eventBody()); - - assert(result); - expect(result.batchItemFailures).toEqual([{ itemIdentifier: "rec-1" }]); - }); - - it("acks (no rows) when the raw no longer parses", async () => { - const harness = makeHarness({ parseEmail: async () => ({ ok: false, reason: "unparseable" }) }); - - const result = await harness.run(eventBody()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - expect(harness.published).toHaveLength(0); - }); - - it("fails the record for a malformed event envelope", async () => { - const harness = makeHarness({ derivedHtml: "https://a.test/x" }); - - const result = await harness.run(JSON.stringify({ detail: { wrong: "shape" } })); - - assert(result); - expect(result.batchItemFailures).toEqual([{ itemIdentifier: "rec-1" }]); - }); - - it("is idempotent under re-delivery: no duplicate rows, re-publishes the fan-out", async () => { - const harness = makeHarness({ derivedHtml: "https://a.test/x https://b.test/y" }); - - await harness.run(eventBody()); - await harness.run(eventBody()); - - const { links, meta } = await harness.linkStore.listLinksByEmail({ - userId: USER, - receivedAtMessageId: RAM, - }); - expect(links).toHaveLength(2); - // Meta is an idempotent overwrite — re-delivery leaves a single barrier row. - expect(meta).toEqual({ truncated: false }); - expect(harness.published).toHaveLength(4); - }); -}); diff --git a/projects/hutch/src/runtime/domain/inbox/extract-email-links-handler.ts b/projects/hutch/src/runtime/domain/inbox/extract-email-links-handler.ts deleted file mode 100644 index 86426eca8..000000000 --- a/projects/hutch/src/runtime/domain/inbox/extract-email-links-handler.ts +++ /dev/null @@ -1,165 +0,0 @@ -import type { - Handler, - SQSBatchItemFailure, - SQSBatchResponse, - SQSEvent, -} from "aws-lambda"; -import { EmailReceivedEvent } from "@packages/hutch-infra-components"; -import type { HutchLogger } from "@packages/hutch-logger"; -import { - capEmailLinks, - type EmailLinkOrdinal, - formatEmailLinkOrdinal, - type InboxEmailLinkStore, - type InboxEmailStore, - type ParseEmailResult, - type ParsedEmailInlineImage, -} from "@packages/domain/inbox"; -import { extractUrls } from "@packages/domain/import-session"; -import { type UserId, UserIdSchema } from "@packages/domain/user"; - -/** - * Consumes `EmailReceivedEvent` and turns the links found inside the email into - * `pending` preview rows, then fans out one `CrawlEmailLinkPreview` command per - * link (mirroring how /queue fans each import URL into its own SaveLinkCommand). - * - * The body is RE-DERIVED from the immutable raw `.eml` on every run — read raw → - * re-parse → re-sanitize — so a future parse/sanitize change applies to - * extraction too, never a body sanitized by stale logic. The per-email soft cap - * bounds the fan-out; a truncated email still delivers the first N previews (the - * working path) while writing a truncated meta item and raising a DLQ alert. - */ -export function initExtractEmailLinksHandler(deps: { - getEmail: InboxEmailStore["getEmail"]; - readRawEmail: (s3Key: string) => Promise; - parseEmail: (input: { raw: Buffer; receivedAt: string }) => Promise; - deriveSanitizedBody: (input: { html: string; inlineImages: ParsedEmailInlineImage[] }) => string; - putLink: InboxEmailLinkStore["putLink"]; - putLinksMeta: InboxEmailLinkStore["putLinksMeta"]; - publishCrawlPreview: (input: { - userId: UserId; - receivedAtMessageId: string; - ordinal: EmailLinkOrdinal; - url: string; - }) => Promise; - alertTruncated: (input: { - userId: UserId; - receivedAtMessageId: string; - found: number; - }) => Promise; - logger: HutchLogger; - maxLinks: number; -}): Handler { - const { - getEmail, - readRawEmail, - parseEmail, - deriveSanitizedBody, - putLink, - putLinksMeta, - publishCrawlPreview, - alertTruncated, - logger, - maxLinks, - } = deps; - - return async (event): Promise => { - const batchItemFailures: SQSBatchItemFailure[] = []; - - for (const record of event.Records) { - try { - const envelope = JSON.parse(record.body); - const parsed = EmailReceivedEvent.detailSchema.safeParse(envelope.detail); - if (!parsed.success) { - logger.error("[extract-email-links] malformed event", { messageId: record.messageId }); - batchItemFailures.push({ itemIdentifier: record.messageId }); - continue; - } - const userId = UserIdSchema.parse(parsed.data.userId); - const { receivedAtMessageId } = parsed.data; - - const email = await getEmail({ userId, receivedAtMessageId }); - if (email === undefined || email.status !== "received") { - // Only `received` emails have a renderable body to extract from; - // rejected/unparsed rows (and a row not yet visible) are skipped. - logger.info("[extract-email-links] nothing to extract", { - receivedAtMessageId, - status: email?.status, - }); - continue; - } - - const raw = await readRawEmail(email.rawEmailS3Key); - if (raw === undefined) { - // S3 can be eventually consistent at receipt; retry. - logger.warn("[extract-email-links] raw .eml not yet readable, retrying", { - s3Key: email.rawEmailS3Key, - }); - batchItemFailures.push({ itemIdentifier: record.messageId }); - continue; - } - - const parsedEmail = await parseEmail({ raw, receivedAt: email.receivedAt }); - if (!parsedEmail.ok) { - // A body that does not parse has no links to extract. - logger.warn("[extract-email-links] raw no longer parseable", { receivedAtMessageId }); - continue; - } - - const sanitizedHtml = deriveSanitizedBody({ - html: parsedEmail.email.html, - inlineImages: parsedEmail.email.inlineImages, - }); - const extracted = extractUrls(Buffer.from(sanitizedHtml, "utf8")); - const { urls, truncated } = capEmailLinks(extracted, { maxLinks }); - - for (const [index, url] of urls.entries()) { - const ordinal = formatEmailLinkOrdinal(index); - // Put the pending row BEFORE publishing so the Articles tab shows N - // pending cards immediately; a re-delivery hits the conditional put as - // a no-op duplicate, then re-publishes (the crawl consumer is idempotent). - await putLink({ - userId, - receivedAtMessageId, - ordinal, - url, - status: "pending", - title: undefined, - excerpt: undefined, - siteName: undefined, - imageUrl: undefined, - failureReason: undefined, - }); - await publishCrawlPreview({ userId, receivedAtMessageId, ordinal, url }); - } - - // Write the per-email meta row LAST, after every link row is in place, so - // its presence is an "extraction finished" barrier: the detail view reads - // no-meta as "still extracting, keep polling" and meta-present-with-zero-rows - // as the genuinely terminal "no links found" — never collapsing the two. - await putLinksMeta({ userId, receivedAtMessageId, meta: { truncated } }); - - if (truncated) { - await alertTruncated({ userId, receivedAtMessageId, found: extracted.totalFound }); - logger.error("[extract-email-links] link cap hit, truncated", { - receivedAtMessageId, - found: extracted.totalFound, - kept: urls.length, - }); - } - logger.info("[extract-email-links] extracted", { - receivedAtMessageId, - links: urls.length, - }); - } catch (error) { - logger.error("[extract-email-links] record failed", { - messageId: record.messageId, - error, - }); - batchItemFailures.push({ itemIdentifier: record.messageId }); - } - } - - return { batchItemFailures }; - }; -} diff --git a/projects/hutch/src/runtime/domain/inbox/receive-email-handler.test.ts b/projects/hutch/src/runtime/domain/inbox/receive-email-handler.test.ts deleted file mode 100644 index f4e985e68..000000000 --- a/projects/hutch/src/runtime/domain/inbox/receive-email-handler.test.ts +++ /dev/null @@ -1,389 +0,0 @@ -import assert from "node:assert/strict"; -import { HutchLogger, noopLogger } from "@packages/hutch-logger"; -import { - DEFAULT_INBOX_ALIAS, - type InboxEmailStore, - MessageIdSchema, - type ParseEmailResult, -} from "@packages/domain/inbox"; -import { type UserId, UserIdSchema } from "@packages/domain/user"; -import { buildLambdaContext } from "@packages/test-fixtures/lambda-context"; -import { initInMemoryInboxAddress } from "@packages/test-fixtures/providers/inbox-address"; -import { initInMemoryInboxEmail } from "@packages/test-fixtures/providers/inbox-email"; -import { buildSqsEvent } from "@packages/test-fixtures/sqs"; -import { initReceiveEmailHandler } from "./receive-email-handler"; -import type { StoreEmailBody } from "./store-email-body"; - -const OWNER = UserIdSchema.parse("00000000000000000000000000000001"); -const SECOND = UserIdSchema.parse("00000000000000000000000000000002"); -const UNROUTED = UserIdSchema.parse("__unrouted__"); -const RECEIVED_AT = "2026-06-24T09:00:00.000Z"; -const RAW_KEY = "inbound/ses-msg-1"; - -function sesNotification(recipients: string[]): string { - return JSON.stringify({ - mail: { messageId: "ses-msg-1" }, - receipt: { - timestamp: RECEIVED_AT, - recipients, - action: { objectKey: RAW_KEY }, - }, - }); -} - -function parsedOk(): ParseEmailResult { - return { - ok: true, - email: { - from: "news@example.com", - subject: "Digest", - text: "text", - html: "

hi

", - messageId: MessageIdSchema.parse(""), - receivedAt: RECEIVED_AT, - inlineImages: [], - }, - }; -} - -function makeHarness(opts?: { - parseEmail?: () => Promise; - storeBody?: StoreEmailBody; - maxEmailBytes?: number; -}) { - const addressStore = initInMemoryInboxAddress({ now: () => new Date() }); - const emailStore = initInMemoryInboxEmail(); - const rawMap = new Map(); - const published: { detail: { receivedAtMessageId: string; userId: string } }[] = []; - - const handler = initReceiveEmailHandler({ - readRawEmail: async (key) => rawMap.get(key), - findByAddress: addressStore.findByAddress, - putEmail: emailStore.putEmail, - parseEmail: opts?.parseEmail ?? (async () => parsedOk()), - storeBody: opts?.storeBody ?? (async () => "content/email/content.html"), - publishEvent: async (_event, detail) => { - published.push({ detail: detail as { receivedAtMessageId: string; userId: string } }); - }, - logger: HutchLogger.from(noopLogger), - maxEmailBytes: opts?.maxEmailBytes ?? 20 * 1024 * 1024, - }); - - const runMany = (recipients: string[]) => - handler( - buildSqsEvent([{ messageId: "rec-1", body: sesNotification(recipients) }]), - buildLambdaContext(), - () => {}, - ); - const run = (recipient: string) => runMany([recipient]); - - return { addressStore, emailStore, rawMap, published, handler, run, runMany }; -} - -async function listEmails(emailStore: InboxEmailStore, userId: UserId) { - const { emails } = await emailStore.listEmailsByUserId({ userId, page: 1, pageSize: 100 }); - return emails; -} - -async function mintAddress(addressStore: ReturnType) { - const entry = await addressStore.createAddress({ - userId: OWNER, - domain: "read.place", - name: DEFAULT_INBOX_ALIAS, - }); - return entry.address; -} - -describe("initReceiveEmailHandler", () => { - it("fails the record for a structurally invalid SES notification, writing no row", async () => { - const { emailStore, published, handler } = makeHarness(); - - const result = await handler( - buildSqsEvent([{ messageId: "rec-1", body: JSON.stringify({ wrong: "shape" }) }]), - buildLambdaContext(), - () => {}, - ); - - assert(result); - expect(result.batchItemFailures).toHaveLength(1); - expect(await listEmails(emailStore, OWNER)).toHaveLength(0); - expect(published).toHaveLength(0); - }); - - it("retries (no row) when the raw .eml is not yet readable", async () => { - const { addressStore, emailStore, published, run } = makeHarness(); - const address = await mintAddress(addressStore); - - const result = await run(address); - - assert(result); - expect(result.batchItemFailures).toHaveLength(1); - expect(await listEmails(emailStore, OWNER)).toHaveLength(0); - expect(published).toHaveLength(0); - }); - - it("ACKs a non-forwarding recipient (raw kept, no row, no operator page)", async () => { - const { emailStore, rawMap, published, run } = makeHarness(); - rawMap.set(RAW_KEY, Buffer.from("raw")); - - const result = await run("postmaster@read.place"); - - assert(result); - // Expected on a public catch-all MX — ACK rather than page the operator. - expect(result.batchItemFailures).toHaveLength(0); - expect(await listEmails(emailStore, OWNER)).toHaveLength(0); - expect(published).toHaveLength(0); - }); - - it("rejects an oversize email with an audit row under the owner and a DLQ failure", async () => { - const { addressStore, emailStore, rawMap, published, run } = makeHarness({ maxEmailBytes: 8 }); - const address = await mintAddress(addressStore); - rawMap.set(RAW_KEY, Buffer.from("this is definitely longer than eight bytes")); - - const result = await run(address); - - assert(result); - expect(result.batchItemFailures).toHaveLength(1); - const [row] = await listEmails(emailStore, OWNER); - expect(row.status).toBe("rejected"); - expect(row.bodyS3Key).toBeUndefined(); - expect(published).toHaveLength(0); - }); - - it("ACKs an oversize email addressed only to unknown recipients (no page)", async () => { - const { emailStore, rawMap, published, run } = makeHarness({ maxEmailBytes: 8 }); - rawMap.set(RAW_KEY, Buffer.from("this is definitely longer than eight bytes")); - - const result = await run("in-zzzzzz@read.place"); - - assert(result); - // Oversize spam to a guessed address on the public MX has no victim — audit - // under the unrouted partition and ACK rather than page the operator. - expect(result.batchItemFailures).toHaveLength(0); - const [row] = await listEmails(emailStore, UNROUTED); - expect(row.status).toBe("rejected"); - expect(published).toHaveLength(0); - }); - - it("records an unknown recipient under the unrouted partition and ACKs (no page)", async () => { - const { emailStore, rawMap, published, run } = makeHarness(); - rawMap.set(RAW_KEY, Buffer.from("raw")); - - const result = await run("in-zzzzzz@read.place"); - - assert(result); - // A guessed/mistyped address is expected on a public MX — audit, don't page. - expect(result.batchItemFailures).toHaveLength(0); - expect(await listEmails(emailStore, OWNER)).toHaveLength(0); - const [row] = await listEmails(emailStore, UNROUTED); - expect(row.status).toBe("rejected"); - expect(published).toHaveLength(0); - }); - - it("records a disabled recipient under the unrouted partition and ACKs (no page)", async () => { - const { addressStore, emailStore, rawMap, published, run } = makeHarness(); - const address = await mintAddress(addressStore); - await addressStore.disableAddress({ userId: OWNER, address }); - rawMap.set(RAW_KEY, Buffer.from("raw")); - - const result = await run(address); - - assert(result); - // Mail to a turned-off address recurs while senders still have it — audit, - // don't page. The owner opted out, so the row lands under the unrouted - // partition rather than cluttering their list with "Rejected" rows. - expect(result.batchItemFailures).toHaveLength(0); - expect(await listEmails(emailStore, OWNER)).toHaveLength(0); - const [row] = await listEmails(emailStore, UNROUTED); - expect(row.status).toBe("rejected"); - expect(row.recipientAddress).toBe(address); - expect(published).toHaveLength(0); - }); - - it("records an unparseable email as status=unparsed and fails to the DLQ", async () => { - const { addressStore, emailStore, rawMap, published, run } = makeHarness({ - parseEmail: async () => ({ ok: false, reason: "unparseable" }), - }); - const address = await mintAddress(addressStore); - rawMap.set(RAW_KEY, Buffer.from("raw")); - - const result = await run(address); - - assert(result); - expect(result.batchItemFailures).toHaveLength(1); - const [row] = await listEmails(emailStore, OWNER); - expect(row.status).toBe("unparsed"); - expect(row.bodyS3Key).toBeUndefined(); - expect(published).toHaveLength(0); - }); - - it("ACKs an unparseable email addressed only to unknown recipients (no page)", async () => { - const { emailStore, rawMap, published, run } = makeHarness({ - parseEmail: async () => ({ ok: false, reason: "unparseable" }), - }); - rawMap.set(RAW_KEY, Buffer.from("raw")); - - const result = await run("in-zzzzzz@read.place"); - - assert(result); - // Malformed spam to a guessed address is not a parser gap worth paging on — - // audit under the unrouted partition and ACK. - expect(result.batchItemFailures).toHaveLength(0); - const [row] = await listEmails(emailStore, UNROUTED); - expect(row.status).toBe("unparsed"); - expect(published).toHaveLength(0); - }); - - it("stores a received email with a body pointer and publishes EmailReceived once", async () => { - const { addressStore, emailStore, rawMap, published, run } = makeHarness({ - storeBody: async () => "content/email/content.html", - }); - const address = await mintAddress(addressStore); - rawMap.set(RAW_KEY, Buffer.from("raw")); - - const result = await run(address); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - const [row] = await listEmails(emailStore, OWNER); - expect(row.status).toBe("received"); - expect(row.bodyS3Key).toBe("content/email/content.html"); - expect(row.senderEmail).toBe("news@example.com"); - expect(published).toHaveLength(1); - expect(published[0].detail.receivedAtMessageId).toBe(`${RECEIVED_AT}#`); - expect(published[0].detail.userId).toBe(OWNER); - }); - - it("matches a recipient case-insensitively when an MTA upper-cases the local part", async () => { - const { addressStore, emailStore, rawMap, published, run } = makeHarness(); - const address = await mintAddress(addressStore); - rawMap.set(RAW_KEY, Buffer.from("raw")); - - // Minted addresses are lowercase and the lookup is exact; an MTA that - // preserves a differently-cased local part must still reach the owner. - const result = await run(address.toUpperCase()); - - assert(result); - expect(result.batchItemFailures).toHaveLength(0); - const [row] = await listEmails(emailStore, OWNER); - expect(row.status).toBe("received"); - expect(published).toHaveLength(1); - }); - - it("persists 'unparsed' (no body, no event) and ACKs when the body sanitizes to nothing", async () => { - const { addressStore, emailStore, rawMap, published, run } = makeHarness({ - // An all-", - inlineImages: [], - }); - - // A body the sanitizer empties (only stripped tags) must NOT become a - // zero-byte object that reads back as "" and renders a blank iframe — signal - // "no body" so the caller persists `unparsed` and shows the unavailable panel. - expect(key).toBeUndefined(); - expect(writes).toHaveLength(0); - }); -}); diff --git a/projects/hutch/src/runtime/domain/inbox/store-email-body.ts b/projects/hutch/src/runtime/domain/inbox/store-email-body.ts deleted file mode 100644 index 09d6c6685..000000000 --- a/projects/hutch/src/runtime/domain/inbox/store-email-body.ts +++ /dev/null @@ -1,40 +0,0 @@ -import { deriveSanitizedBody } from "@packages/domain/inbox"; -import type { ParsedEmailInlineImage } from "@packages/domain/inbox"; -import type { UserId } from "@packages/domain/user"; -import { emailContentResourceId } from "./email-content-id"; - -export type StoreEmailBody = (input: { - userId: UserId; - receivedAtMessageId: string; - html: string; - inlineImages: ParsedEmailInlineImage[]; -}) => Promise; - -/** - * Turn a parsed newsletter into the safe HTML the View tab renders, and write it - * to the content bucket. Each `cid:` inline image (already resolved to an - * `email://cid/` URL by the preparser) is inlined as a `data:` URI carrying - * its bytes, then the allowlist sanitizer keeps those — and strips every remote - * image so tracking beacons can't fire. Inlining (rather than rehosting to a - * cross-origin CDN) is what lets the images render inside the View tab's - * sandboxed, opaque-origin iframe. - * - * Returns the S3 key the body was written to, or `undefined` when sanitizing - * leaves nothing renderable — a body composed entirely of tags the sanitizer - * strips wholesale (`