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/inbox/src/runtime/domain/inbox/crawl-email-link-preview-handler.ts b/projects/inbox/src/runtime/domain/inbox/crawl-email-link-preview-handler.ts new file mode 100644 index 000000000..871bd108a --- /dev/null +++ b/projects/inbox/src/runtime/domain/inbox/crawl-email-link-preview-handler.ts @@ -0,0 +1,92 @@ +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; +}): Handlerbody
")), + 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("fails the record when a dependency throws, so SQS redelivers it", async () => { + const harness = makeHarness({ + getEmail: async () => { + throw new Error("dynamo down"); + }, + }); + + const result = await harness.run(eventBody()); + + assert(result); + expect(result.batchItemFailures).toEqual([{ itemIdentifier: "rec-1" }]); + expect(harness.published).toHaveLength(0); + }); + + 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/inbox/src/runtime/domain/inbox/extract-email-links-handler.ts b/projects/inbox/src/runtime/domain/inbox/extract-email-links-handler.ts new file mode 100644 index 000000000..86426eca8 --- /dev/null +++ b/projects/inbox/src/runtime/domain/inbox/extract-email-links-handler.ts @@ -0,0 +1,165 @@ +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) => Promisehi
", + messageId: MessageIdSchema.parse("