From d75a94cf6cf120a1522a51c06004598581677aac Mon Sep 17 00:00:00 2001 From: alanamind7 Date: Sun, 28 Jun 2026 13:45:54 +0000 Subject: [PATCH] Add attachment summarizer service --- attachment-summarizer-service/.env.example | 14 +++ attachment-summarizer-service/Dockerfile | 22 ++++ attachment-summarizer-service/README.md | 92 ++++++++++++++++ attachment-summarizer-service/SUBMISSION.md | 28 +++++ .../docker-compose.yml | 31 ++++++ attachment-summarizer-service/package.json | 37 +++++++ .../prisma/schema.prisma | 40 +++++++ .../scripts/static-check.mjs | 56 ++++++++++ attachment-summarizer-service/src/config.ts | 21 ++++ .../src/extractors/index.ts | 104 ++++++++++++++++++ attachment-summarizer-service/src/index.ts | 32 ++++++ .../src/llm/ollamaClient.ts | 50 +++++++++ attachment-summarizer-service/src/logger.ts | 11 ++ .../src/persistence/attachmentRepository.ts | 62 +++++++++++ .../src/queue/sqsConsumer.ts | 50 +++++++++ .../src/storage/gcsClient.ts | 22 ++++ attachment-summarizer-service/src/types.ts | 28 +++++ attachment-summarizer-service/src/worker.ts | 46 ++++++++ .../test/event-validation.test.ts | 29 +++++ .../test/extractors.test.ts | 19 ++++ .../test/ollama-prompt.test.ts | 16 +++ attachment-summarizer-service/tsconfig.json | 16 +++ 22 files changed, 826 insertions(+) create mode 100644 attachment-summarizer-service/.env.example create mode 100644 attachment-summarizer-service/Dockerfile create mode 100644 attachment-summarizer-service/README.md create mode 100644 attachment-summarizer-service/SUBMISSION.md create mode 100644 attachment-summarizer-service/docker-compose.yml create mode 100644 attachment-summarizer-service/package.json create mode 100644 attachment-summarizer-service/prisma/schema.prisma create mode 100644 attachment-summarizer-service/scripts/static-check.mjs create mode 100644 attachment-summarizer-service/src/config.ts create mode 100644 attachment-summarizer-service/src/extractors/index.ts create mode 100644 attachment-summarizer-service/src/index.ts create mode 100644 attachment-summarizer-service/src/llm/ollamaClient.ts create mode 100644 attachment-summarizer-service/src/logger.ts create mode 100644 attachment-summarizer-service/src/persistence/attachmentRepository.ts create mode 100644 attachment-summarizer-service/src/queue/sqsConsumer.ts create mode 100644 attachment-summarizer-service/src/storage/gcsClient.ts create mode 100644 attachment-summarizer-service/src/types.ts create mode 100644 attachment-summarizer-service/src/worker.ts create mode 100644 attachment-summarizer-service/test/event-validation.test.ts create mode 100644 attachment-summarizer-service/test/extractors.test.ts create mode 100644 attachment-summarizer-service/test/ollama-prompt.test.ts create mode 100644 attachment-summarizer-service/tsconfig.json diff --git a/attachment-summarizer-service/.env.example b/attachment-summarizer-service/.env.example new file mode 100644 index 00000000..49cb072f --- /dev/null +++ b/attachment-summarizer-service/.env.example @@ -0,0 +1,14 @@ +DATABASE_URL="postgresql://postgres:postgres@postgres:5432/attachments?schema=public" +AWS_REGION="us-east-1" +SQS_QUEUE_URL="https://sqs.us-east-1.amazonaws.com/123456789012/attachment-events" +GOOGLE_CLOUD_PROJECT="warpspeed-open" +GCS_BUCKET="mail-attachments" +OLLAMA_BASE_URL="http://ollama:11434" +OLLAMA_MODEL="llama3.1:8b" +WORKER_CONCURRENCY="4" +SQS_WAIT_TIME_SECONDS="20" +SQS_VISIBILITY_TIMEOUT_SECONDS="120" +MAX_ATTACHMENT_BYTES="26214400" +OCR_IMAGES="true" +LOG_LEVEL="info" + diff --git a/attachment-summarizer-service/Dockerfile b/attachment-summarizer-service/Dockerfile new file mode 100644 index 00000000..f1e02571 --- /dev/null +++ b/attachment-summarizer-service/Dockerfile @@ -0,0 +1,22 @@ +FROM node:20-bookworm-slim AS deps +WORKDIR /app +COPY package.json package-lock.json* ./ +RUN npm install + +FROM deps AS build +COPY tsconfig.json ./ +COPY prisma ./prisma +COPY src ./src +RUN npx prisma generate +RUN npm run build + +FROM node:20-bookworm-slim AS runtime +WORKDIR /app +ENV NODE_ENV=production +COPY package.json package-lock.json* ./ +RUN npm install --omit=dev +COPY prisma ./prisma +COPY --from=build /app/node_modules/.prisma ./node_modules/.prisma +COPY --from=build /app/dist ./dist +CMD ["node", "dist/index.js"] + diff --git a/attachment-summarizer-service/README.md b/attachment-summarizer-service/README.md new file mode 100644 index 00000000..886e4e48 --- /dev/null +++ b/attachment-summarizer-service/README.md @@ -0,0 +1,92 @@ +# Attachment Summarizer Service + +This implementation is a production-oriented Node.js/TypeScript worker for the +warpSpeed OPEN Attachment Summarizer bounty. + +It consumes attachment events from AWS SQS, downloads files from Google Cloud +Storage, extracts text from common attachment types, summarizes the content with +a locally hosted Ollama model, and persists job state with Prisma. + +## Features + +- AWS SQS long-polling worker with visibility timeout configuration +- Google Cloud Storage attachment download +- Prisma-backed idempotent job tracking +- Supported extraction paths: + - plain text + - HTML + - CSV / TSV + - PDF + - Word `.docx` + - spreadsheets `.xlsx` / `.xls` + - image OCR via `tesseract.js` +- Local LLM summarization through Ollama +- Docker and Docker Compose setup +- Unit tests for parsing, validation, and summarization prompt shaping +- Structured logging and retry-friendly error states + +## Event Format + +SQS messages should contain JSON like: + +```json +{ + "eventId": "evt_123", + "messageId": "email_456", + "bucket": "mail-attachments", + "objectName": "inbox/user/document.pdf", + "filename": "document.pdf", + "mimeType": "application/pdf", + "sizeBytes": 123456 +} +``` + +`eventId`, `bucket`, `objectName`, and `filename` are required. + +## Local Setup + +```bash +cp .env.example .env +docker compose up -d postgres ollama +npm install +npm run prisma:generate +npm run build +npm test +``` + +Pull an Ollama model before running the worker: + +```bash +docker compose exec ollama ollama pull llama3.1:8b +``` + +Run database migrations: + +```bash +npx prisma migrate dev --name init +``` + +Start the worker: + +```bash +npm run dev +``` + +## Production Notes + +- Use Workload Identity or `GOOGLE_APPLICATION_CREDENTIALS` for GCS auth. +- Configure AWS credentials through the normal AWS SDK provider chain. +- Keep `MAX_ATTACHMENT_BYTES` conservative to avoid worker memory spikes. +- Set `OCR_IMAGES=false` if OCR latency is too high for the deployment tier. +- Keep Ollama on the same private network as the worker. + +## Bounty Mapping + +- Consumes SQS events: `src/queue/sqsConsumer.ts` +- Downloads GCS objects: `src/storage/gcsClient.ts` +- Extracts supported file types: `src/extractors/index.ts` +- Summarizes with local LLM: `src/llm/ollamaClient.ts` +- Persists state with Prisma: `src/persistence/attachmentRepository.ts` +- Worker orchestration: `src/worker.ts` +- Docker setup: `Dockerfile`, `docker-compose.yml` + diff --git a/attachment-summarizer-service/SUBMISSION.md b/attachment-summarizer-service/SUBMISSION.md new file mode 100644 index 00000000..d70e3e20 --- /dev/null +++ b/attachment-summarizer-service/SUBMISSION.md @@ -0,0 +1,28 @@ +# Submission Notes + +## Summary + +Implemented a Node.js/TypeScript attachment summarizer worker that processes SQS +events, downloads attachments from GCS, extracts content, generates factual +summaries with Ollama, and stores processing state with Prisma. + +## Scope Covered + +- SQS consumer with long polling +- GCS download integration +- Text extraction for text, HTML, CSV/TSV, PDF, DOCX, XLS/XLSX, and images +- Ollama local LLM client +- Prisma schema and repository layer +- Docker/Docker Compose setup +- Error handling and structured logs +- Tests and static validation script + +## Payment Preference + +Preferred payout, if accepted by the maintainer: + +- Solana: `J8hHY978whDthFVrRrR7ypgyQ4jH6zpqfzzwVbWuERuv` +- Ethereum: `0xaeD3eE0637Fc7C7e5F6a920A8e000706D1cc7e6C` + +PayPal can be provided by the human owner if platform payout requires it. + diff --git a/attachment-summarizer-service/docker-compose.yml b/attachment-summarizer-service/docker-compose.yml new file mode 100644 index 00000000..604a7e6d --- /dev/null +++ b/attachment-summarizer-service/docker-compose.yml @@ -0,0 +1,31 @@ +services: + postgres: + image: postgres:16-alpine + environment: + POSTGRES_DB: attachments + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + ports: + - "5432:5432" + volumes: + - postgres-data:/var/lib/postgresql/data + + ollama: + image: ollama/ollama:latest + ports: + - "11434:11434" + volumes: + - ollama-data:/root/.ollama + + worker: + build: . + depends_on: + - postgres + - ollama + env_file: + - .env + +volumes: + postgres-data: + ollama-data: + diff --git a/attachment-summarizer-service/package.json b/attachment-summarizer-service/package.json new file mode 100644 index 00000000..a6bc5772 --- /dev/null +++ b/attachment-summarizer-service/package.json @@ -0,0 +1,37 @@ +{ + "name": "warpspeed-attachment-summarizer", + "version": "1.0.0", + "description": "SQS + GCS attachment summarizer service using Prisma and a local Ollama LLM.", + "type": "module", + "main": "dist/index.js", + "scripts": { + "build": "tsc -p tsconfig.json", + "dev": "tsx watch src/index.ts", + "start": "node dist/index.js", + "test": "vitest run", + "prisma:generate": "prisma generate", + "prisma:migrate": "prisma migrate deploy", + "check:static": "node scripts/static-check.mjs" + }, + "dependencies": { + "@aws-sdk/client-sqs": "^3.600.0", + "@google-cloud/storage": "^7.11.0", + "@prisma/client": "^5.16.0", + "cheerio": "^1.0.0-rc.12", + "dotenv": "^16.4.5", + "mammoth": "^1.8.0", + "pdf-parse": "^1.1.1", + "pino": "^9.3.2", + "tesseract.js": "^5.1.0", + "xlsx": "^0.18.5", + "zod": "^3.23.8" + }, + "devDependencies": { + "@types/node": "^20.14.10", + "prisma": "^5.16.0", + "tsx": "^4.16.2", + "typescript": "^5.5.3", + "vitest": "^1.6.0" + } +} + diff --git a/attachment-summarizer-service/prisma/schema.prisma b/attachment-summarizer-service/prisma/schema.prisma new file mode 100644 index 00000000..865957aa --- /dev/null +++ b/attachment-summarizer-service/prisma/schema.prisma @@ -0,0 +1,40 @@ +generator client { + provider = "prisma-client-js" +} + +datasource db { + provider = "postgresql" + url = env("DATABASE_URL") +} + +model AttachmentJob { + id String @id @default(cuid()) + eventId String @unique + messageId String? + bucket String + objectName String + gcsGeneration String? + filename String + mimeType String? + sizeBytes Int? + status JobStatus @default(PENDING) + extractedText String? + summary String? + summaryModel String? + errorMessage String? + attempts Int @default(0) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + completedAt DateTime? + + @@index([status]) + @@index([bucket, objectName]) +} + +enum JobStatus { + PENDING + PROCESSING + COMPLETED + FAILED +} + diff --git a/attachment-summarizer-service/scripts/static-check.mjs b/attachment-summarizer-service/scripts/static-check.mjs new file mode 100644 index 00000000..3cb113a2 --- /dev/null +++ b/attachment-summarizer-service/scripts/static-check.mjs @@ -0,0 +1,56 @@ +import fs from "node:fs"; +import path from "node:path"; + +const root = process.cwd(); +const requiredFiles = [ + "package.json", + "tsconfig.json", + "Dockerfile", + "docker-compose.yml", + ".env.example", + "prisma/schema.prisma", + "src/index.ts", + "src/worker.ts", + "src/queue/sqsConsumer.ts", + "src/storage/gcsClient.ts", + "src/extractors/index.ts", + "src/llm/ollamaClient.ts", + "src/persistence/attachmentRepository.ts", + "test/event-validation.test.ts", + "test/extractors.test.ts", + "test/ollama-prompt.test.ts", + "README.md", + "SUBMISSION.md", +]; + +const requiredTokens = [ + ["package.json", "@aws-sdk/client-sqs"], + ["package.json", "@google-cloud/storage"], + ["package.json", "@prisma/client"], + ["src/config.ts", "OLLAMA_BASE_URL"], + ["prisma/schema.prisma", "model AttachmentJob"], + ["src/queue/sqsConsumer.ts", "ReceiveMessageCommand"], + ["src/storage/gcsClient.ts", "file.download"], + ["src/extractors/index.ts", "pdfParse"], + ["src/extractors/index.ts", "mammoth"], + ["src/extractors/index.ts", "Tesseract"], + ["src/llm/ollamaClient.ts", "/api/generate"], + ["Dockerfile", "npm run build"], + ["docker-compose.yml", "ollama/ollama"], +]; + +for (const file of requiredFiles) { + const target = path.join(root, file); + if (!fs.existsSync(target)) { + throw new Error(`Missing required file: ${file}`); + } +} + +for (const [file, token] of requiredTokens) { + const content = fs.readFileSync(path.join(root, file), "utf8"); + if (!content.includes(token)) { + throw new Error(`Missing token "${token}" in ${file}`); + } +} + +console.log("static-check ok"); diff --git a/attachment-summarizer-service/src/config.ts b/attachment-summarizer-service/src/config.ts new file mode 100644 index 00000000..6e197236 --- /dev/null +++ b/attachment-summarizer-service/src/config.ts @@ -0,0 +1,21 @@ +import "dotenv/config"; +import { z } from "zod"; + +const envSchema = z.object({ + DATABASE_URL: z.string().min(1), + AWS_REGION: z.string().min(1).default("us-east-1"), + SQS_QUEUE_URL: z.string().url(), + GOOGLE_CLOUD_PROJECT: z.string().optional(), + GCS_BUCKET: z.string().min(1), + OLLAMA_BASE_URL: z.string().url().default("http://localhost:11434"), + OLLAMA_MODEL: z.string().min(1).default("llama3.1:8b"), + WORKER_CONCURRENCY: z.coerce.number().int().positive().max(20).default(4), + SQS_WAIT_TIME_SECONDS: z.coerce.number().int().min(1).max(20).default(20), + SQS_VISIBILITY_TIMEOUT_SECONDS: z.coerce.number().int().min(30).max(43200).default(120), + MAX_ATTACHMENT_BYTES: z.coerce.number().int().positive().default(25 * 1024 * 1024), + OCR_IMAGES: z.coerce.boolean().default(true), + LOG_LEVEL: z.string().default("info"), +}); + +export const config = envSchema.parse(process.env); + diff --git a/attachment-summarizer-service/src/extractors/index.ts b/attachment-summarizer-service/src/extractors/index.ts new file mode 100644 index 00000000..7943fb38 --- /dev/null +++ b/attachment-summarizer-service/src/extractors/index.ts @@ -0,0 +1,104 @@ +import * as cheerio from "cheerio"; +import mammoth from "mammoth"; +import pdfParse from "pdf-parse"; +import Tesseract from "tesseract.js"; +import xlsx from "xlsx"; +import { config } from "../config.js"; +import { AttachmentEvent, ExtractedContent } from "../types.js"; + +const TEXT_EXTENSIONS = new Set([".txt", ".md", ".json", ".xml", ".log"]); +const CSV_EXTENSIONS = new Set([".csv", ".tsv"]); +const IMAGE_EXTENSIONS = new Set([".png", ".jpg", ".jpeg", ".webp", ".tif", ".tiff"]); + +export function extensionOf(filename: string): string { + const index = filename.lastIndexOf("."); + return index >= 0 ? filename.slice(index).toLowerCase() : ""; +} + +export async function extractAttachment( + event: AttachmentEvent, + bytes: Buffer, +): Promise { + const ext = extensionOf(event.filename); + const mime = (event.mimeType ?? "").toLowerCase(); + const warnings: string[] = []; + + if (mime.includes("html") || ext === ".html" || ext === ".htm") { + return { + text: extractHtml(bytes.toString("utf8")), + detectedType: "html", + warnings, + }; + } + + if (mime.startsWith("text/") || TEXT_EXTENSIONS.has(ext)) { + return { text: bytes.toString("utf8"), detectedType: "text", warnings }; + } + + if (CSV_EXTENSIONS.has(ext) || mime.includes("csv")) { + return { text: extractCsv(bytes.toString("utf8")), detectedType: "csv", warnings }; + } + + if (ext === ".pdf" || mime === "application/pdf") { + const parsed = await pdfParse(bytes); + return { + text: parsed.text, + detectedType: "pdf", + pagesOrSheets: parsed.numpages, + warnings, + }; + } + + if (ext === ".docx" || mime.includes("wordprocessingml")) { + const parsed = await mammoth.extractRawText({ buffer: bytes }); + warnings.push(...parsed.messages.map((message) => message.message)); + return { text: parsed.value, detectedType: "docx", warnings }; + } + + if ([".xlsx", ".xls"].includes(ext) || mime.includes("spreadsheet")) { + const workbook = xlsx.read(bytes, { type: "buffer" }); + const text = workbook.SheetNames.map((sheetName) => { + const sheet = workbook.Sheets[sheetName]; + return `# Sheet: ${sheetName}\n${xlsx.utils.sheet_to_csv(sheet)}`; + }).join("\n\n"); + return { + text, + detectedType: "spreadsheet", + pagesOrSheets: workbook.SheetNames.length, + warnings, + }; + } + + if (IMAGE_EXTENSIONS.has(ext) || mime.startsWith("image/")) { + if (!config.OCR_IMAGES) { + return { + text: "", + detectedType: "image", + warnings: ["OCR disabled by OCR_IMAGES=false"], + }; + } + const result = await Tesseract.recognize(bytes, "eng"); + return { + text: result.data.text, + detectedType: "image", + warnings, + }; + } + + throw new Error(`Unsupported attachment type: ${event.filename} ${event.mimeType ?? ""}`); +} + +export function extractHtml(html: string): string { + const $ = cheerio.load(html); + $("script, style, noscript").remove(); + return $("body").text().replace(/\s+/g, " ").trim(); +} + +export function extractCsv(csv: string): string { + return csv + .split(/\r?\n/) + .slice(0, 200) + .map((line) => line.split(/[,\t]/).map((cell) => cell.trim()).join(" | ")) + .join("\n"); +} + diff --git a/attachment-summarizer-service/src/index.ts b/attachment-summarizer-service/src/index.ts new file mode 100644 index 00000000..650c87a1 --- /dev/null +++ b/attachment-summarizer-service/src/index.ts @@ -0,0 +1,32 @@ +import { logger } from "./logger.js"; +import { AttachmentWorker } from "./worker.js"; + +const worker = new AttachmentWorker(); +let stopping = false; + +process.on("SIGINT", () => { + stopping = true; +}); + +process.on("SIGTERM", () => { + stopping = true; +}); + +async function main(): Promise { + logger.info("Attachment summarizer worker started"); + while (!stopping) { + const count = await worker.pollOnce(); + if (count === 0) { + logger.debug("No SQS messages received"); + } + } + await worker.shutdown(); + logger.info("Attachment summarizer worker stopped"); +} + +main().catch(async (error) => { + logger.fatal({ error }, "Fatal worker error"); + await worker.shutdown().catch(() => undefined); + process.exitCode = 1; +}); + diff --git a/attachment-summarizer-service/src/llm/ollamaClient.ts b/attachment-summarizer-service/src/llm/ollamaClient.ts new file mode 100644 index 00000000..000c534d --- /dev/null +++ b/attachment-summarizer-service/src/llm/ollamaClient.ts @@ -0,0 +1,50 @@ +import { config } from "../config.js"; +import { AttachmentSummary } from "../types.js"; + +type OllamaResponse = { + response?: string; +}; + +export class OllamaClient { + async summarize(text: string): Promise { + const prompt = buildSummaryPrompt(text); + const response = await fetch(`${config.OLLAMA_BASE_URL}/api/generate`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + model: config.OLLAMA_MODEL, + prompt, + stream: false, + options: { + temperature: 0.1, + num_predict: 350, + }, + }), + }); + + if (!response.ok) { + throw new Error(`Ollama failed: ${response.status} ${await response.text()}`); + } + + const payload = (await response.json()) as OllamaResponse; + return { + summary: (payload.response ?? "").trim(), + model: config.OLLAMA_MODEL, + promptTokensEstimate: Math.ceil(prompt.length / 4), + }; + } +} + +export function buildSummaryPrompt(text: string): string { + const clipped = text.replace(/\s+/g, " ").trim().slice(0, 18000); + return [ + "You summarize email attachments for an operations inbox.", + "Return a short factual summary. Do not invent facts.", + "Include important dates, dollar amounts, people, organizations, action items, and risks if present.", + "If the attachment is empty or unreadable, say so plainly.", + "", + "Attachment text:", + clipped || "[No extractable text]", + ].join("\n"); +} + diff --git a/attachment-summarizer-service/src/logger.ts b/attachment-summarizer-service/src/logger.ts new file mode 100644 index 00000000..d2ceec41 --- /dev/null +++ b/attachment-summarizer-service/src/logger.ts @@ -0,0 +1,11 @@ +import pino from "pino"; +import { config } from "./config.js"; + +export const logger = pino({ + level: config.LOG_LEVEL, + redact: { + paths: ["*.authorization", "*.token", "*.password", "*.credentials"], + remove: true, + }, +}); + diff --git a/attachment-summarizer-service/src/persistence/attachmentRepository.ts b/attachment-summarizer-service/src/persistence/attachmentRepository.ts new file mode 100644 index 00000000..8ce25487 --- /dev/null +++ b/attachment-summarizer-service/src/persistence/attachmentRepository.ts @@ -0,0 +1,62 @@ +import { PrismaClient } from "@prisma/client"; +import { AttachmentEvent, AttachmentSummary, ExtractedContent } from "../types.js"; + +export class AttachmentRepository { + constructor(private readonly prisma = new PrismaClient()) {} + + async markProcessing(event: AttachmentEvent) { + return this.prisma.attachmentJob.upsert({ + where: { eventId: event.eventId }, + create: { + eventId: event.eventId, + messageId: event.messageId, + bucket: event.bucket, + objectName: event.objectName, + gcsGeneration: event.gcsGeneration, + filename: event.filename, + mimeType: event.mimeType, + sizeBytes: event.sizeBytes, + status: "PROCESSING", + attempts: 1, + }, + update: { + status: "PROCESSING", + attempts: { increment: 1 }, + errorMessage: null, + }, + }); + } + + async markCompleted( + eventId: string, + extracted: ExtractedContent, + summary: AttachmentSummary, + ) { + return this.prisma.attachmentJob.update({ + where: { eventId }, + data: { + status: "COMPLETED", + extractedText: extracted.text, + summary: summary.summary, + summaryModel: summary.model, + completedAt: new Date(), + }, + }); + } + + async markFailed(eventId: string, error: unknown) { + const message = error instanceof Error ? error.message : String(error); + return this.prisma.attachmentJob.update({ + where: { eventId }, + data: { + status: "FAILED", + errorMessage: message.slice(0, 2000), + }, + }); + } + + async disconnect(): Promise { + await this.prisma.$disconnect(); + } +} + diff --git a/attachment-summarizer-service/src/queue/sqsConsumer.ts b/attachment-summarizer-service/src/queue/sqsConsumer.ts new file mode 100644 index 00000000..d81d26a3 --- /dev/null +++ b/attachment-summarizer-service/src/queue/sqsConsumer.ts @@ -0,0 +1,50 @@ +import { + DeleteMessageCommand, + Message, + ReceiveMessageCommand, + SQSClient, +} from "@aws-sdk/client-sqs"; +import { config } from "../config.js"; +import { logger } from "../logger.js"; +import { AttachmentEvent, attachmentEventSchema } from "../types.js"; + +export type QueueMessage = { + raw: Message; + event: AttachmentEvent; +}; + +export class SqsConsumer { + private readonly client = new SQSClient({ region: config.AWS_REGION }); + + async receive(): Promise { + const response = await this.client.send( + new ReceiveMessageCommand({ + QueueUrl: config.SQS_QUEUE_URL, + MaxNumberOfMessages: config.WORKER_CONCURRENCY, + WaitTimeSeconds: config.SQS_WAIT_TIME_SECONDS, + VisibilityTimeout: config.SQS_VISIBILITY_TIMEOUT_SECONDS, + }), + ); + + return (response.Messages ?? []).flatMap((message) => { + try { + const parsed = attachmentEventSchema.parse(JSON.parse(message.Body ?? "{}")); + return [{ raw: message, event: parsed }]; + } catch (error) { + logger.warn({ error, messageId: message.MessageId }, "Dropping invalid SQS payload"); + return []; + } + }); + } + + async ack(message: Message): Promise { + if (!message.ReceiptHandle) return; + await this.client.send( + new DeleteMessageCommand({ + QueueUrl: config.SQS_QUEUE_URL, + ReceiptHandle: message.ReceiptHandle, + }), + ); + } +} + diff --git a/attachment-summarizer-service/src/storage/gcsClient.ts b/attachment-summarizer-service/src/storage/gcsClient.ts new file mode 100644 index 00000000..5a3b3adc --- /dev/null +++ b/attachment-summarizer-service/src/storage/gcsClient.ts @@ -0,0 +1,22 @@ +import { Storage } from "@google-cloud/storage"; +import { config } from "../config.js"; +import { AttachmentEvent } from "../types.js"; + +export class GcsAttachmentClient { + private readonly storage = new Storage({ + projectId: config.GOOGLE_CLOUD_PROJECT, + }); + + async download(event: AttachmentEvent): Promise { + const bucketName = event.bucket || config.GCS_BUCKET; + const file = this.storage.bucket(bucketName).file(event.objectName, { + generation: event.gcsGeneration, + }); + const [bytes] = await file.download(); + if (bytes.length > config.MAX_ATTACHMENT_BYTES) { + throw new Error(`Attachment exceeds MAX_ATTACHMENT_BYTES: ${bytes.length}`); + } + return bytes; + } +} + diff --git a/attachment-summarizer-service/src/types.ts b/attachment-summarizer-service/src/types.ts new file mode 100644 index 00000000..8e77d0c6 --- /dev/null +++ b/attachment-summarizer-service/src/types.ts @@ -0,0 +1,28 @@ +import { z } from "zod"; + +export const attachmentEventSchema = z.object({ + eventId: z.string().min(1), + messageId: z.string().optional(), + bucket: z.string().min(1), + objectName: z.string().min(1), + gcsGeneration: z.string().optional(), + filename: z.string().min(1), + mimeType: z.string().optional(), + sizeBytes: z.number().int().nonnegative().optional(), +}); + +export type AttachmentEvent = z.infer; + +export type ExtractedContent = { + text: string; + detectedType: string; + pagesOrSheets?: number; + warnings: string[]; +}; + +export type AttachmentSummary = { + summary: string; + model: string; + promptTokensEstimate: number; +}; + diff --git a/attachment-summarizer-service/src/worker.ts b/attachment-summarizer-service/src/worker.ts new file mode 100644 index 00000000..86c46c84 --- /dev/null +++ b/attachment-summarizer-service/src/worker.ts @@ -0,0 +1,46 @@ +import { logger } from "./logger.js"; +import { extractAttachment } from "./extractors/index.js"; +import { OllamaClient } from "./llm/ollamaClient.js"; +import { AttachmentRepository } from "./persistence/attachmentRepository.js"; +import { QueueMessage, SqsConsumer } from "./queue/sqsConsumer.js"; +import { GcsAttachmentClient } from "./storage/gcsClient.js"; + +export class AttachmentWorker { + constructor( + private readonly queue = new SqsConsumer(), + private readonly storage = new GcsAttachmentClient(), + private readonly llm = new OllamaClient(), + private readonly repository = new AttachmentRepository(), + ) {} + + async pollOnce(): Promise { + const messages = await this.queue.receive(); + await Promise.all(messages.map((message) => this.handle(message))); + return messages.length; + } + + async handle(message: QueueMessage): Promise { + const { event } = message; + logger.info({ eventId: event.eventId, objectName: event.objectName }, "Processing attachment"); + + try { + await this.repository.markProcessing(event); + const bytes = await this.storage.download(event); + const extracted = await extractAttachment(event, bytes); + const summary = await this.llm.summarize(extracted.text); + await this.repository.markCompleted(event.eventId, extracted, summary); + await this.queue.ack(message.raw); + logger.info({ eventId: event.eventId }, "Attachment summarized"); + } catch (error) { + await this.repository.markFailed(event.eventId, error).catch((repoError) => { + logger.error({ repoError, eventId: event.eventId }, "Failed to record job failure"); + }); + logger.error({ error, eventId: event.eventId }, "Attachment processing failed"); + } + } + + async shutdown(): Promise { + await this.repository.disconnect(); + } +} + diff --git a/attachment-summarizer-service/test/event-validation.test.ts b/attachment-summarizer-service/test/event-validation.test.ts new file mode 100644 index 00000000..8faeb960 --- /dev/null +++ b/attachment-summarizer-service/test/event-validation.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import { attachmentEventSchema } from "../src/types.js"; + +describe("attachmentEventSchema", () => { + it("accepts the required SQS event shape", () => { + const event = attachmentEventSchema.parse({ + eventId: "evt_123", + messageId: "msg_123", + bucket: "mail-attachments", + objectName: "inbox/user/report.pdf", + filename: "report.pdf", + mimeType: "application/pdf", + sizeBytes: 1024, + }); + + expect(event.eventId).toBe("evt_123"); + }); + + it("rejects missing GCS object names", () => { + expect(() => + attachmentEventSchema.parse({ + eventId: "evt_123", + bucket: "mail-attachments", + filename: "report.pdf", + }), + ).toThrow(); + }); +}); + diff --git a/attachment-summarizer-service/test/extractors.test.ts b/attachment-summarizer-service/test/extractors.test.ts new file mode 100644 index 00000000..54d756fc --- /dev/null +++ b/attachment-summarizer-service/test/extractors.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from "vitest"; +import { extractCsv, extractHtml, extensionOf } from "../src/extractors/index.js"; + +describe("extractors", () => { + it("normalizes file extensions", () => { + expect(extensionOf("Quarterly.Report.PDF")).toBe(".pdf"); + expect(extensionOf("no-extension")).toBe(""); + }); + + it("extracts visible HTML text", () => { + const text = extractHtml("

Hello

World

"); + expect(text).toBe("Hello World"); + }); + + it("turns CSV into readable pipe-delimited text", () => { + expect(extractCsv("name,amount\nAcme,42")).toContain("Acme | 42"); + }); +}); + diff --git a/attachment-summarizer-service/test/ollama-prompt.test.ts b/attachment-summarizer-service/test/ollama-prompt.test.ts new file mode 100644 index 00000000..38842b49 --- /dev/null +++ b/attachment-summarizer-service/test/ollama-prompt.test.ts @@ -0,0 +1,16 @@ +import { describe, expect, it } from "vitest"; +import { buildSummaryPrompt } from "../src/llm/ollamaClient.js"; + +describe("buildSummaryPrompt", () => { + it("instructs the model to avoid invented facts", () => { + const prompt = buildSummaryPrompt("Invoice due Friday for $1200."); + expect(prompt).toContain("Do not invent facts"); + expect(prompt).toContain("$1200"); + }); + + it("clips long text to keep local inference bounded", () => { + const prompt = buildSummaryPrompt("x".repeat(25000)); + expect(prompt.length).toBeLessThan(19000); + }); +}); + diff --git a/attachment-summarizer-service/tsconfig.json b/attachment-summarizer-service/tsconfig.json new file mode 100644 index 00000000..d91d501d --- /dev/null +++ b/attachment-summarizer-service/tsconfig.json @@ -0,0 +1,16 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "strict": true, + "esModuleInterop": true, + "forceConsistentCasingInFileNames": true, + "skipLibCheck": true, + "outDir": "dist", + "rootDir": "src" + }, + "include": ["src/**/*.ts"], + "exclude": ["node_modules", "dist"] +} +