Data Management Lambdas
What is AWS Lambda? Lambda is a service that lets you run code without provisioning or managing a traditional server. Our data module system requires four small Lambda functions to receive commands from users and read/write data to S3 and DynamoDB.
STEP 3.1: Create Get Document Lambda
This function is responsible for: When a user clicks to view document details (API GET /documents/{documentId}), the function will fetch the detailed data and return it to them.
Access the Lambda service:
- In the search bar at the top of the AWS Console, enter Lambda and select the Lambda service.
- In the search bar at the top of the AWS Console, enter Lambda and select the Lambda service.
Initialize the function:
- Click the Create function button in the top right corner.
- Select the Author from scratch option (Default).
Configure basic information:
- Function name: Enter the exact source code standard name:
docuflow-dev-data-get-document-lambda. Or name it according to your requirements. - Runtime: Choose the standard programming language for the project (e.g.,
Node.js 24.x).
- Architecture: Select
arm64.
- Function name: Enter the exact source code standard name:
Configure Permissions:
- Scroll down and expand the Additional settings section.
- Under Custom execution role, select Use an existing role and choose the
docuflow-dev-data-lambda-rolecreated in the IAM section.
- Click Save.
- Scroll down and expand the Additional settings section.
Enter tags:
- Click on tags.
- Select the necessary tags according to the setup (e.g., Project: DocuFlowAI).
- Click Save.
Click the Create function button to create the lambda.
Once the function is created, you will see the management interface. In the Code source section, you can paste the programming code below and click the Deploy button to save the code.
Source Code (index.mjs):
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, GetCommand, ScanCommand } from "@aws-sdk/lib-dynamodb";
import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
const ddbClient = new DynamoDBClient({});
const docClient = DynamoDBDocumentClient.from(ddbClient);
const s3Client = new S3Client({});
const TABLE_NAME = process.env.DOCUFLOW_DEV_TABLE_NAME;
const RAW_BUCKET = process.env.DOCUFLOW_DEV_RAW_BUCKET || process.env.DOCUFLOW_DEV_RAW_BUCKET_NAME;
const PROCESSED_BUCKET = process.env.DOCUFLOW_DEV_PROCESSED_BUCKET;
const SOURCE_URL_EXPIRES_SECONDS = Number(process.env.SOURCE_URL_EXPIRES_SECONDS || 900);
let getSignedUrl;
export const handler = async (event) => {
try {
const method = event?.requestContext?.http?.method || event?.httpMethod || "";
if (method === "OPTIONS") {
return formatResponse(200, true, null);
}
if (method !== "GET") {
return formatResponse(
405,
false,
null,
"Only GET is supported.",
"API",
"METHOD_NOT_ALLOWED"
);
}
const envError = validateEnvironment();
if (envError) return envError;
const documentId = getDocumentId(event);
const userId = getUserId(event);
const admin = isAdmin(event);
if (!documentId) {
return formatResponse(400, false, null, "Missing documentId.", "VALIDATION", "INVALID_INPUT");
}
if (!userId) {
return formatResponse(401, false, null, "Missing authenticated Cognito user.", "AUTH", "UNAUTHORIZED");
}
logInfo("Get document request received", { documentId, userId, admin });
let dbRes = await docClient.send(new GetCommand({
TableName: TABLE_NAME,
Key: {
PK: `USER#${userId}`,
SK: `DOC#${documentId}`,
},
}));
if (!dbRes.Item && admin) {
const adminResult = await docClient.send(new ScanCommand({
TableName: TABLE_NAME,
ExpressionAttributeNames: { "#sk": "SK" },
ExpressionAttributeValues: { ":sk": `DOC#${documentId}` },
FilterExpression: "#sk = :sk",
Limit: 1,
}));
dbRes = { Item: adminResult.Items?.[0] };
}
if (!dbRes.Item) {
return formatResponse(404, false, null, "Document not found.", "DYNAMODB", "NOT_FOUND");
}
const safeItem = removeDynamoDbKeys(dbRes.Item);
const processedS3Key = resolveProcessedS3Key(safeItem);
if (!processedS3Key) {
return formatResponse(200, true, await normalizeDocumentResponse(safeItem, null));
}
let fullDocument;
try {
const s3Res = await s3Client.send(new GetObjectCommand({
Bucket: PROCESSED_BUCKET,
Key: normalizeS3Key(processedS3Key),
}));
fullDocument = await parseS3JsonBody(s3Res.Body);
logInfo("Processed result loaded from S3", {
documentId,
userId,
processedS3Bucket: PROCESSED_BUCKET,
processedS3Key,
});
} catch (s3Error) {
logError("Failed to read processed result from S3. Returning DynamoDB metadata only.", {
documentId,
userId,
processedS3Bucket: PROCESSED_BUCKET,
processedS3Key,
errorName: s3Error?.name,
errorMessage: s3Error?.message,
});
return formatResponse(200, true, {
...await normalizeDocumentResponse({ ...safeItem, processedS3Key }, null),
s3ReadWarning: {
errorCode: "S3_READ_FAILED",
errorMessage: s3Error?.message || "Could not read processed result from S3.",
errorStage: "S3_PROCESSED",
},
});
}
return formatResponse(
200,
true,
await normalizeDocumentResponse({ ...safeItem, processedS3Key }, fullDocument)
);
} catch (error) {
logError("Unhandled error", {
errorName: error?.name,
errorMessage: error?.message,
});
return formatResponse(
500,
false,
null,
error?.message || "Unknown error.",
"UNKNOWN",
"UNKNOWN_ERROR"
);
}
};
function validateEnvironment() {
if (!TABLE_NAME) {
return formatResponse(
500,
false,
null,
"DOCUFLOW_DEV_TABLE_NAME environment variable is missing.",
"DYNAMODB",
"UNKNOWN_ERROR"
);
}
if (!PROCESSED_BUCKET) {
return formatResponse(
500,
false,
null,
"DOCUFLOW_DEV_PROCESSED_BUCKET environment variable is missing.",
"S3_PROCESSED",
"UNKNOWN_ERROR"
);
}
if (
!Number.isInteger(SOURCE_URL_EXPIRES_SECONDS) ||
SOURCE_URL_EXPIRES_SECONDS < 60 ||
SOURCE_URL_EXPIRES_SECONDS > 3600
) {
return formatResponse(
500,
false,
null,
"SOURCE_URL_EXPIRES_SECONDS must be an integer between 60 and 3600.",
"CONFIGURATION",
"INVALID_ENVIRONMENT_VARIABLE"
);
}
return null;
}
function formatResponse(statusCode, success, data, errorMessage = null, errorStage = null, errorCode = "UNKNOWN_ERROR") {
const body = { success, data, error: null };
if (errorMessage) {
body.error = { errorCode, errorMessage, errorStage };
}
return {
statusCode,
headers: {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "Content-Type,Authorization",
"Access-Control-Allow-Methods": "OPTIONS,GET",
"Content-Type": "application/json",
},
body: JSON.stringify(body),
};
}
function getDocumentId(event) {
return (
event?.pathParameters?.documentId ||
event?.queryStringParameters?.documentId ||
event?.documentId ||
null
);
}
function getUserId(event) {
const claims =
event?.requestContext?.authorizer?.jwt?.claims ||
event?.requestContext?.authorizer?.claims ||
{};
return claims.sub || null;
}
function getGroups(event) {
const claims =
event?.requestContext?.authorizer?.jwt?.claims ||
event?.requestContext?.authorizer?.claims ||
{};
const groups = claims["cognito:groups"] || [];
return Array.isArray(groups) ? groups : String(groups).split(",");
}
function isAdmin(event) {
return getGroups(event).some((group) => group.trim().toLowerCase() === "admin");
}
function removeDynamoDbKeys(item = {}) {
const { PK, SK, GSI1PK, GSI1SK, ...safeItem } = item;
return safeItem;
}
function resolveProcessedS3Key(item = {}) {
return (
item.processedS3Key ||
item.storage?.processedS3Key ||
item.processed?.s3Key ||
item.resultS3Key ||
null
);
}
function resolveRawS3Key(item = {}) {
return (
item.rawS3Key ||
item.storage?.rawS3Key ||
item.raw?.s3Key ||
item.sourceS3Key ||
null
);
}
function parseS3Location(value, fallbackBucket = null) {
if (!value) return { bucket: null, key: null };
if (value.startsWith("s3://")) {
const withoutScheme = value.slice("s3://".length);
const slashIndex = withoutScheme.indexOf("/");
if (slashIndex < 0) return { bucket: withoutScheme, key: null };
return {
bucket: withoutScheme.slice(0, slashIndex),
key: withoutScheme.slice(slashIndex + 1),
};
}
return {
bucket: fallbackBucket,
key: normalizeS3Key(value),
};
}
async function loadGetSignedUrl() {
if (getSignedUrl) return getSignedUrl;
const presigner = await import("@aws-sdk/s3-request-presigner");
getSignedUrl = presigner.getSignedUrl;
return getSignedUrl;
}
async function createSourceUrl(rawS3Key) {
const { bucket, key } = parseS3Location(rawS3Key, RAW_BUCKET);
if (!bucket || !key) {
return null;
}
try {
const sign = await loadGetSignedUrl();
return await sign(
s3Client,
new GetObjectCommand({
Bucket: bucket,
Key: key,
}),
{
expiresIn: SOURCE_URL_EXPIRES_SECONDS,
}
);
} catch (error) {
logError("Failed to create source document presigned URL.", {
rawS3Bucket: bucket,
rawS3Key: key,
errorName: error?.name,
errorMessage: error?.message,
});
return null;
}
}
function normalizeS3Key(key) {
if (!key) return key;
if (key.startsWith("s3://")) {
const withoutScheme = key.slice("s3://".length);
const slashIndex = withoutScheme.indexOf("/");
return slashIndex >= 0 ? withoutScheme.slice(slashIndex + 1) : withoutScheme;
}
return key.replace(/^\/+/, "");
}
async function parseS3JsonBody(bodyStream) {
if (!bodyStream) {
throw new Error("S3 object Body is empty.");
}
const text = await bodyStream.transformToString();
return parsePossiblyDoubleStringifiedJson(text);
}
function parsePossiblyDoubleStringifiedJson(value) {
let parsed = typeof value === "string" ? JSON.parse(value) : value;
if (typeof parsed === "string") {
parsed = JSON.parse(parsed);
}
return parsed;
}
async function normalizeDocumentResponse(safeItem, fullDocument) {
const doc = unwrapProcessedDocument(fullDocument);
const invoice = firstRecord(
doc?.invoice,
doc?.normalized?.invoice,
doc?.extracted?.invoice,
doc?.data?.invoice,
{}
);
const confidence = firstRecord(doc?.confidence, doc?.normalized?.confidence, {});
const review = firstRecord(doc?.review, {});
const audit = firstRecord(doc?.audit, {});
const storage = firstRecord(doc?.storage, {});
const file = firstRecord(doc?.file, {});
const textractSummary = extractTextractSummaryFields(doc);
const lineItems = extractLineItems(doc, invoice);
const rawS3Key = pickValue(storage.rawS3Key, doc?.rawS3Key, safeItem.rawS3Key);
const sourceUrl = pickValue(doc?.sourceUrl, safeItem.sourceUrl, await createSourceUrl(rawS3Key), null);
return {
...safeItem,
documentId: pickValue(doc?.documentId, safeItem.documentId),
userId: pickValue(doc?.userId, safeItem.userId),
documentType: pickValue(doc?.documentType, safeItem.documentType, inferDocumentType(doc), "UNKNOWN"),
status: pickValue(doc?.status, safeItem.status),
originalFileName: pickValue(file.originalFileName, doc?.originalFileName, doc?.fileName, safeItem.originalFileName),
rawS3Key,
processedS3Key: pickValue(storage.processedS3Key, doc?.processedS3Key, safeItem.processedS3Key),
sourceUrl,
vendorName: pickValue(invoice.vendorName, doc?.vendorName, textractSummary.vendorName, safeItem.vendorName, ""),
invoiceNumber: pickValue(invoice.invoiceNumber, doc?.invoiceNumber, textractSummary.invoiceNumber, safeItem.invoiceNumber, ""),
invoiceDate: pickValue(invoice.invoiceDate, doc?.invoiceDate, textractSummary.invoiceDate, safeItem.invoiceDate, ""),
dueDate: pickValue(invoice.dueDate, doc?.dueDate, textractSummary.dueDate, safeItem.dueDate, ""),
currency: pickValue(invoice.currency, doc?.currency, safeItem.currency, inferCurrency(doc), "USD"),
subtotalAmount: toNumber(pickValue(invoice.subtotalAmount, doc?.subtotalAmount, textractSummary.subtotalAmount, safeItem.subtotalAmount), 0),
taxAmount: nullableNumber(pickValue(invoice.taxAmount, doc?.taxAmount, textractSummary.taxAmount, safeItem.taxAmount)),
discountAmount: toNumber(pickValue(invoice.discountAmount, doc?.discountAmount, safeItem.discountAmount), 0),
shippingAmount: toNumber(pickValue(invoice.shippingAmount, doc?.shippingAmount, safeItem.shippingAmount), 0),
totalAmount: toNumber(pickValue(invoice.totalAmount, doc?.totalAmount, textractSummary.totalAmount, safeItem.totalAmount), 0),
confidenceScore: normalizeConfidence(pickValue(
confidence.confidenceScore,
doc?.confidenceScore,
safeItem.confidenceScore
)),
hasLowConfidence: Boolean(pickValue(confidence.hasLowConfidence, doc?.hasLowConfidence, safeItem.hasLowConfidence, false)),
fieldConfidence: firstRecord(confidence.fieldConfidence, doc?.fieldConfidence, safeItem.fieldConfidence, {}),
reviewStatus: pickValue(review.reviewStatus, doc?.reviewStatus, safeItem.reviewStatus, "PENDING"),
reviewReasonCodes: firstArray(review.reviewReasonCodes, doc?.reviewReasonCodes, safeItem.reviewReasonCodes, []),
corrections: firstArray(review.corrections, doc?.corrections, []),
lineItems,
createdAt: pickValue(audit.createdAt, doc?.createdAt, safeItem.createdAt),
updatedAt: pickValue(audit.updatedAt, doc?.updatedAt, safeItem.updatedAt),
};
}
function unwrapProcessedDocument(value) {
let doc = value;
if (typeof doc === "string") {
doc = parsePossiblyDoubleStringifiedJson(doc);
}
if (!isRecord(doc)) return {};
return firstRecord(
doc.document,
doc.result,
doc.data?.document,
doc.data?.result,
doc.data,
doc
);
}
function extractLineItems(doc, invoice) {
const directItems = firstArray(
doc?.lineItems,
invoice?.lineItems,
doc?.normalized?.lineItems,
doc?.normalized?.invoice?.lineItems,
doc?.extracted?.lineItems,
doc?.data?.lineItems,
[]
);
if (directItems.length > 0) {
return directItems.map(normalizeLineItem).filter((item) => item.description || item.totalAmount > 0);
}
const textractItems = extractTextractLineItems(doc);
if (textractItems.length > 0) {
return textractItems;
}
return [];
}
function extractTextractLineItems(doc) {
const expenseDocuments = firstArray(
doc?.ExpenseDocuments,
doc?.expenseDocuments,
doc?.textract?.ExpenseDocuments,
doc?.textractResult?.ExpenseDocuments,
doc?.rawTextract?.ExpenseDocuments,
doc?.analyzeExpense?.ExpenseDocuments,
[]
);
const items = [];
for (const expenseDocument of expenseDocuments) {
const groups = firstArray(expenseDocument?.LineItemGroups, expenseDocument?.lineItemGroups, []);
for (const group of groups) {
const lineItems = firstArray(group?.LineItems, group?.lineItems, []);
for (const lineItem of lineItems) {
const normalized = normalizeTextractLineItem(lineItem);
if (normalized.description || normalized.totalAmount > 0) {
items.push(normalized);
}
}
}
}
return items;
}
function extractTextractSummaryFields(doc) {
const expenseDocuments = firstArray(
doc?.ExpenseDocuments,
doc?.expenseDocuments,
doc?.textract?.ExpenseDocuments,
doc?.textractResult?.ExpenseDocuments,
doc?.rawTextract?.ExpenseDocuments,
doc?.analyzeExpense?.ExpenseDocuments,
[]
);
const byType = {};
for (const expenseDocument of expenseDocuments) {
const summaryFields = firstArray(expenseDocument?.SummaryFields, expenseDocument?.summaryFields, []);
for (const field of summaryFields) {
const type = normalizeKey(field?.Type?.Text || field?.type?.text || field?.Type || field?.type);
const label = normalizeKey(field?.LabelDetection?.Text || field?.labelDetection?.text || "");
const key = type || label;
const value = field?.ValueDetection?.Text || field?.valueDetection?.text || field?.Value || field?.value || "";
if (key && value && !byType[key]) {
byType[key] = value;
}
}
}
return {
vendorName: pickSummaryValue(byType, ["VENDOR_NAME", "VENDOR", "FROM"]),
invoiceNumber: pickSummaryValue(byType, ["INVOICE_RECEIPT_ID", "INVOICE_NUMBER", "RECEIPT_NUMBER"]),
invoiceDate: pickSummaryValue(byType, ["INVOICE_RECEIPT_DATE", "INVOICE_DATE", "RECEIPT_DATE"]),
dueDate: pickSummaryValue(byType, ["DUE_DATE"]),
subtotalAmount: pickSummaryValue(byType, ["SUBTOTAL", "SUB_TOTAL"]),
taxAmount: pickSummaryValue(byType, ["TAX", "TAX_AMOUNT"]),
totalAmount: pickSummaryValue(byType, ["TOTAL", "TOTAL_DUE", "AMOUNT_DUE"]),
};
}
function normalizeTextractLineItem(lineItem) {
const fields = firstArray(lineItem?.LineItemExpenseFields, lineItem?.lineItemExpenseFields, []);
const byType = {};
for (const field of fields) {
const type = normalizeKey(field?.Type?.Text || field?.type?.text || field?.Type || field?.type);
const label = normalizeKey(field?.LabelDetection?.Text || field?.labelDetection?.text || "");
const key = type || label;
const value = field?.ValueDetection?.Text || field?.valueDetection?.text || field?.Value || field?.value || "";
const confidence = field?.ValueDetection?.Confidence ?? field?.valueDetection?.confidence ?? field?.Confidence ?? field?.confidence;
if (!key || !value) continue;
if (!byType[key]) {
byType[key] = { value, confidence };
}
}
const quantity = pickField(byType, ["QUANTITY", "QTY", "HRS_QTY", "HOURS", "HRS"]);
const description = pickField(byType, [
"ITEM",
"ITEM_DESCRIPTION",
"DESCRIPTION",
"SERVICE",
"PRODUCT",
"EXPENSE_ROW",
]);
const unitPrice = pickField(byType, [
"UNIT_PRICE",
"PRICE",
"RATE",
"RATE_PRICE",
"RATE/PRICE",
"UNIT_COST",
]);
const total = pickField(byType, [
"AMOUNT",
"TOTAL",
"LINE_TOTAL",
"SUBTOTAL",
"SUB_TOTAL",
"LINE_ITEM_TOTAL",
]);
const tax = pickField(byType, ["TAX", "TAX_AMOUNT"]);
const confidence = Math.min(
...[quantity, description, unitPrice, total, tax]
.map((field) => normalizeConfidence(field?.confidence))
.filter((value) => value > 0)
);
const normalizedTotal = toNumber(total?.value, toNumber(unitPrice?.value, 0));
return {
lineItemId: lineItem?.LineItemExpenseFields?.[0]?.Id || lineItem?.id || "",
description: cleanDescription(description?.value),
quantity: toNumber(quantity?.value, normalizedTotal > 0 ? 1 : 0),
unitPriceAmount: toNumber(unitPrice?.value, normalizedTotal),
taxAmount: toNumber(tax?.value, 0),
totalAmount: normalizedTotal,
confidenceScore: Number.isFinite(confidence) ? confidence : 0,
};
}
function pickSummaryValue(map, keys) {
for (const key of keys) {
const value = map[normalizeKey(key)];
if (value !== undefined && value !== null && value !== "") return value;
}
return undefined;
}
function normalizeLineItem(item) {
const description = pickValue(
item?.description,
item?.itemDescription,
item?.name,
item?.productName,
item?.service,
item?.Service,
""
);
const quantity = pickValue(item?.quantity, item?.qty, item?.hours, item?.hrsQty);
const unitPrice = pickValue(item?.unitPriceAmount, item?.unitPrice, item?.ratePrice, item?.rate, item?.price);
const total = pickValue(item?.totalAmount, item?.amount, item?.subTotal, item?.subtotal, item?.lineTotal);
const normalizedTotal = toNumber(total, toNumber(unitPrice, 0));
return {
lineItemId: String(pickValue(item?.lineItemId, item?.id, "")),
description: cleanDescription(description),
quantity: toNumber(quantity, normalizedTotal > 0 ? 1 : 0),
unitPriceAmount: toNumber(unitPrice, normalizedTotal),
taxAmount: toNumber(pickValue(item?.taxAmount, item?.tax), 0),
totalAmount: normalizedTotal,
confidenceScore: normalizeConfidence(pickValue(item?.confidenceScore, item?.confidence)),
};
}
function pickField(map, keys) {
for (const key of keys) {
const normalizedKey = normalizeKey(key);
if (map[normalizedKey]) return map[normalizedKey];
}
return null;
}
function cleanDescription(value) {
return String(value || "")
.replace(/\s+/g, " ")
.trim();
}
function inferDocumentType(doc) {
const type = String(doc?.documentType || doc?.type || "").toUpperCase();
if (type.includes("RECEIPT")) return "RECEIPT";
if (type.includes("INVOICE")) return "INVOICE";
return undefined;
}
function inferCurrency(doc) {
const text = JSON.stringify(doc || {}).toUpperCase();
if (text.includes("VND") || text.includes("₫")) return "VND";
if (text.includes("EUR") || text.includes("€")) return "EUR";
if (text.includes("GBP") || text.includes("£")) return "GBP";
if (text.includes("CNY") || text.includes("RMB") || text.includes("CN¥")) return "CNY";
if (text.includes("JPY") || text.includes("¥")) return "JPY";
if (text.includes("KRW") || text.includes("₩")) return "KRW";
if (text.includes("SGD") || text.includes("S$")) return "SGD";
if (text.includes("THB") || text.includes("฿")) return "THB";
if (text.includes("AUD") || text.includes("A$")) return "AUD";
if (text.includes("CAD") || text.includes("C$")) return "CAD";
if (text.includes("CHF")) return "CHF";
if (text.includes("HKD") || text.includes("HK$")) return "HKD";
if (text.includes("INR") || text.includes("₹")) return "INR";
if (text.includes("IDR") || text.includes("RP")) return "IDR";
if (text.includes("MYR") || text.includes("RM")) return "MYR";
if (text.includes("PHP") || text.includes("₱")) return "PHP";
if (text.includes("TWD") || text.includes("NT$")) return "TWD";
if (text.includes("USD") || text.includes("$")) return "USD";
return undefined;
}
function normalizeKey(value) {
return String(value || "")
.trim()
.toUpperCase()
.replace(/[^A-Z0-9]+/g, "_")
.replace(/^_+|_+$/g, "");
}
function pickValue(...values) {
for (const value of values) {
if (value !== undefined && value !== null && value !== "") {
if (isRecord(value) && "value" in value) return value.value;
return value;
}
}
return values[values.length - 1];
}
function firstRecord(...values) {
for (const value of values) {
if (isRecord(value)) return value;
}
return {};
}
function firstArray(...values) {
for (const value of values) {
if (Array.isArray(value)) return value;
}
return [];
}
function isRecord(value) {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function toNumber(value, fallback = 0) {
if (isRecord(value) && "value" in value) {
return toNumber(value.value, fallback);
}
if (value === null || value === undefined || value === "") {
return fallback;
}
if (typeof value === "number") {
return Number.isFinite(value) ? value : fallback;
}
const cleaned = String(value)
.replace(/[^\d.,-]/g, "")
.replace(/,/g, "");
const parsed = Number(cleaned);
return Number.isFinite(parsed) ? parsed : fallback;
}
function nullableNumber(value) {
if (value === null || value === undefined || value === "") return null;
return toNumber(value, null);
}
function normalizeConfidence(value) {
const number = toNumber(value, 0);
if (number <= 0) return 0;
if (number <= 1) return number;
return Math.min(number / 100, 1);
}
function logInfo(message, extra = {}) {
console.log(JSON.stringify({
level: "INFO",
service: "docuflow-dev-data-get-document-lambda",
message,
...extra,
}));
}
function logError(message, extra = {}) {
console.error(JSON.stringify({
level: "ERROR",
service: "docuflow-dev-data-get-document-lambda",
message,
...extra,
}));
}
STEP 3.2: CREATE LIST DOCUMENTS LAMBDA
This function helps retrieve the list of all user documents or filter those with errors (API GET /documents).
Return to the main interface of the Lambda service (click on Functions in the left menu).
Click the Create function button and choose Author from scratch.
Configure basic information:
- Function name: Enter exactly
docuflow-dev-data-list-documents-lambda. Or name it according to your requirements. - Runtime: Choose the standard programming language for the project (
Node.js 24.x).
- Architecture: Select
arm64.
- Function name: Enter exactly
Configure Permissions:
- Expand the Additional settings section.
- Under Custom execution role, select Use an existing role and choose the
docuflow-dev-data-lambda-role.
Enter tags:
- Click on tags, select the necessary tags according to the setup.
- Click Save.
- Click on tags, select the necessary tags according to the setup.
Click the Create function button.
Once the function is created, in the Code source section, paste the programming code below and click Deploy.
Source Code (index.mjs):
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, QueryCommand, ScanCommand } from "@aws-sdk/lib-dynamodb";
const ddbClient = new DynamoDBClient({});
const docClient = DynamoDBDocumentClient.from(ddbClient);
const TABLE_NAME = process.env.DOCUFLOW_DEV_TABLE_NAME;
const DEFAULT_PAGE_SIZE = 50;
const MAX_PAGE_SIZE = 100;
const ALLOWED_STATUSES = new Set([
"UPLOADED",
"QUEUED",
"PROCESSING",
"EXTRACTED",
"REVIEW_REQUIRED",
"FAILED",
"CORRECTED",
"APPROVED",
]);
export const handler = async (event) => {
try {
const method = event?.requestContext?.http?.method || event?.httpMethod || "";
if (method === "OPTIONS") {
return formatResponse(200, true, null);
}
if (method !== "GET") {
return formatResponse(
405,
false,
null,
"Only GET is supported.",
"API",
"METHOD_NOT_ALLOWED"
);
}
if (!TABLE_NAME) {
return formatResponse(
500,
false,
null,
"DOCUFLOW_DEV_TABLE_NAME environment variable is missing.",
"CONFIGURATION",
"MISSING_ENVIRONMENT_VARIABLE"
);
}
const userId = getUserId(event);
const admin = isAdmin(event);
if (!userId) {
return formatResponse(
401,
false,
null,
"Missing authenticated Cognito user.",
"AUTH",
"UNAUTHORIZED"
);
}
const query = event?.queryStringParameters || {};
const statusFilter = String(query.status || "").trim().toUpperCase();
if (statusFilter && !ALLOWED_STATUSES.has(statusFilter)) {
return formatResponse(
400,
false,
null,
"Unsupported document status.",
"VALIDATION",
"INVALID_STATUS"
);
}
let exclusiveStartKey;
try {
exclusiveStartKey = decodeNextToken(query.nextToken, userId, admin);
} catch {
return formatResponse(
400,
false,
null,
"Invalid nextToken.",
"VALIDATION",
"INVALID_NEXT_TOKEN"
);
}
const pageSize = normalizePageSize(query.limit);
const result = admin
? await scanAdminDocuments({ statusFilter, exclusiveStartKey, pageSize })
: await queryUserDocuments({ userId, statusFilter, exclusiveStartKey, pageSize });
const items = (result.Items || []).map(normalizeDocumentItem);
log("INFO", {
message: "Documents listed successfully",
userId,
admin,
statusFilter: statusFilter || null,
itemCount: items.length,
});
return formatResponse(200, true, {
items,
nextToken: encodeNextToken(result.LastEvaluatedKey),
});
} catch (error) {
log("ERROR", {
message: "Failed to list documents",
errorName: error?.name,
errorMessage: error?.message,
});
return formatResponse(
500,
false,
null,
error?.message || "Unknown error.",
"DYNAMODB",
"UNKNOWN_ERROR"
);
}
};
function getUserId(event) {
const claims =
event?.requestContext?.authorizer?.jwt?.claims ||
event?.requestContext?.authorizer?.claims ||
{};
return claims.sub || null;
}
function getGroups(event) {
const claims =
event?.requestContext?.authorizer?.jwt?.claims ||
event?.requestContext?.authorizer?.claims ||
{};
const groups = claims["cognito:groups"] || [];
return Array.isArray(groups) ? groups : String(groups).split(",");
}
function isAdmin(event) {
return getGroups(event).some((group) => group.trim().toLowerCase() === "admin");
}
function queryUserDocuments({ userId, statusFilter, exclusiveStartKey, pageSize }) {
const expressionAttributeNames = statusFilter
? { "#status": "status" }
: undefined;
const expressionAttributeValues = {
":pk": `USER#${userId}`,
":documentPrefix": "DOC#",
...(statusFilter ? { ":status": statusFilter } : {}),
};
return docClient.send(new QueryCommand({
TableName: TABLE_NAME,
KeyConditionExpression:
"PK = :pk AND begins_with(SK, :documentPrefix)",
ExpressionAttributeNames: expressionAttributeNames,
ExpressionAttributeValues: expressionAttributeValues,
FilterExpression: statusFilter ? "#status = :status" : undefined,
ExclusiveStartKey: exclusiveStartKey,
Limit: pageSize,
}));
}
function scanAdminDocuments({ statusFilter, exclusiveStartKey, pageSize }) {
const expressionAttributeNames = {
"#sk": "SK",
...(statusFilter ? { "#status": "status" } : {}),
};
const expressionAttributeValues = {
":documentPrefix": "DOC#",
...(statusFilter ? { ":status": statusFilter } : {}),
};
return docClient.send(new ScanCommand({
TableName: TABLE_NAME,
ExpressionAttributeNames: expressionAttributeNames,
ExpressionAttributeValues: expressionAttributeValues,
FilterExpression: statusFilter
? "begins_with(#sk, :documentPrefix) AND #status = :status"
: "begins_with(#sk, :documentPrefix)",
ExclusiveStartKey: exclusiveStartKey,
Limit: pageSize,
}));
}
function normalizePageSize(value) {
const numeric = Number(value);
if (!Number.isInteger(numeric) || numeric <= 0) return DEFAULT_PAGE_SIZE;
return Math.min(numeric, MAX_PAGE_SIZE);
}
function encodeNextToken(lastEvaluatedKey) {
if (!lastEvaluatedKey) return null;
return Buffer.from(JSON.stringify(lastEvaluatedKey), "utf8").toString(
"base64url"
);
}
function decodeNextToken(token, userId, admin = false) {
if (!token) return undefined;
const decoded = JSON.parse(
Buffer.from(String(token), "base64url").toString("utf8")
);
const expectedPk = admin ? /^USER#/ : new RegExp(`^USER#${escapeRegExp(userId)}$`);
if (!decoded || !expectedPk.test(String(decoded.PK || "")) || typeof decoded.SK !== "string" || !decoded.SK.startsWith("DOC#")) {
throw new Error("Invalid nextToken scope.");
}
return decoded;
}
function escapeRegExp(value) {
return String(value).replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}
function removeDynamoDbKeys(item = {}) {
const { PK, SK, GSI1PK, GSI1SK, ...safeItem } = item;
return safeItem;
}
function normalizeDocumentItem(item = {}) {
const safeItem = removeDynamoDbKeys(item);
const file = isRecord(safeItem.file) ? safeItem.file : {};
const storage = isRecord(safeItem.storage) ? safeItem.storage : {};
const rawS3Key = pickString(
safeItem.rawS3Key,
storage.rawS3Key,
safeItem.key
);
const processedS3Key = pickString(
safeItem.processedS3Key,
storage.processedS3Key
);
const originalFileName = pickDisplayFileName({
documentId: safeItem.documentId,
explicitFileName: pickString(
safeItem.originalFileName,
safeItem.fileName,
file.originalFileName,
file.fileName
),
rawS3Key,
});
return {
...safeItem,
originalFileName,
...(rawS3Key ? { rawS3Key } : {}),
...(processedS3Key ? { processedS3Key } : {}),
};
}
function pickDisplayFileName({ documentId, explicitFileName, rawS3Key }) {
if (explicitFileName && !isGenericRawObjectName(explicitFileName)) {
return explicitFileName;
}
const rawFileName = rawS3Key ? decodeS3KeySegment(rawS3Key.split("/").pop()) : "";
if (rawFileName && !isGenericRawObjectName(rawFileName)) {
return rawFileName;
}
if (explicitFileName) {
const extension = getFileExtension(explicitFileName || rawFileName);
return documentId ? `${documentId}${extension}` : explicitFileName;
}
return documentId ? `${documentId}.pdf` : "unknown";
}
function pickString(...values) {
for (const value of values) {
if (typeof value === "string" && value.trim()) return value.trim();
}
return "";
}
function decodeS3KeySegment(value = "") {
try {
return decodeURIComponent(value.replace(/\+/g, "%20"));
} catch {
return value;
}
}
function getFileExtension(fileName = "") {
const match = String(fileName).match(/(\.[A-Za-z0-9]+)$/);
return match ? match[1].toLowerCase() : "";
}
function isGenericRawObjectName(fileName = "") {
return /^original\.[A-Za-z0-9]+$/i.test(String(fileName).trim());
}
function isRecord(value) {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function formatResponse(
statusCode,
success,
data,
errorMessage = null,
errorStage = null,
errorCode = "UNKNOWN_ERROR"
) {
const body = { success, data, error: null };
if (errorMessage) {
body.error = { errorCode, errorMessage, errorStage };
}
return {
statusCode,
headers: {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "Authorization,Content-Type",
"Access-Control-Allow-Methods": "OPTIONS,GET",
"Content-Type": "application/json",
},
body: JSON.stringify(body),
};
}
function log(level, data) {
const writer = level === "ERROR" ? console.error : console.log;
writer(
JSON.stringify({
level,
service: "docuflow-dev-data-list-documents-lambda",
...data,
})
);
}
STEP 3.3: CREATE REVIEW UPDATE LAMBDA
This function is used to save the information that the user manually edited on the interface (e.g., AI recognized the wrong amount and the user corrected it).
Click Create function ➔ Author from scratch.
Configure basic information:
- Function name: Enter exactly
docuflow-dev-data-review-update-lambda. - Runtime: Choose the standard programming language for the project (
Node.js 24.x).
- Architecture: Select
arm64.
- Function name: Enter exactly
Configure Permissions:
- Expand the Additional settings section.
- Under Custom execution role, select Use an existing role and choose the
docuflow-dev-data-lambda-role.
Add tags:
- Click on tags, select the necessary tags according to the setup.
- Click Save.
- Click on tags, select the necessary tags according to the setup.
Click the Create function button.
This function is used to save the information that the user manually edited on the interface (e.g., AI recognized the wrong amount and the user corrected it)
Source Code (index.mjs):
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import {
DynamoDBDocumentClient,
GetCommand,
UpdateCommand,
} from "@aws-sdk/lib-dynamodb";
import {
S3Client,
GetObjectCommand,
PutObjectCommand,
} from "@aws-sdk/client-s3";
const docClient = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const s3Client = new S3Client({});
const TABLE_NAME = process.env.DOCUFLOW_DEV_TABLE_NAME;
const PROCESSED_BUCKET = process.env.DOCUFLOW_DEV_PROCESSED_BUCKET;
const ALLOWED_REVIEW_STATUSES = new Set(["CORRECTED", "APPROVED"]);
const REVIEWABLE_DOCUMENT_STATUSES = new Set([
"EXTRACTED",
"REVIEW_REQUIRED",
"CORRECTED",
"APPROVED",
]);
export function isReviewableDocumentStatus(status) {
return REVIEWABLE_DOCUMENT_STATUSES.has(status);
}
const ALLOWED_CORRECTION_FIELDS = new Set([
"documentType",
"invoiceNumber",
"vendorName",
"invoiceDate",
"dueDate",
"currency",
"subtotalAmount",
"discountAmount",
"shippingAmount",
"totalAmount",
"taxAmount",
"lineItems",
]);
const DYNAMODB_SUMMARY_FIELDS = new Set([
"documentType",
"invoiceNumber",
"vendorName",
"invoiceDate",
"dueDate",
"currency",
"subtotalAmount",
"discountAmount",
"shippingAmount",
"totalAmount",
"taxAmount",
]);
export const handler = async (event) => {
try {
const method = event?.requestContext?.http?.method || event?.httpMethod || "";
if (method === "OPTIONS") return formatResponse(200, true, null);
if (method !== "PATCH") {
return formatResponse(
405,
false,
null,
"Only PATCH is supported.",
"API",
"METHOD_NOT_ALLOWED"
);
}
const environmentError = validateEnvironment();
if (environmentError) return environmentError;
const documentId = event?.pathParameters?.documentId;
const userId = getUserId(event);
if (!userId) {
return formatResponse(
401,
false,
null,
"Missing authenticated Cognito user.",
"AUTH",
"UNAUTHORIZED"
);
}
if (!documentId) {
return formatResponse(
400,
false,
null,
"Missing documentId.",
"VALIDATION",
"INVALID_INPUT"
);
}
const body = parseBody(event);
if (!body) {
return formatResponse(
400,
false,
null,
"Request body must be valid JSON.",
"VALIDATION",
"INVALID_INPUT"
);
}
const reviewStatus = body.reviewStatus;
if (!ALLOWED_REVIEW_STATUSES.has(reviewStatus)) {
return formatResponse(
400,
false,
null,
"reviewStatus must be CORRECTED or APPROVED.",
"VALIDATION",
"INVALID_REVIEW_STATUS"
);
}
const corrections = normalizeCorrections(body.corrections);
if (corrections.error) {
return formatResponse(
400,
false,
null,
corrections.error,
"VALIDATION",
"INVALID_CORRECTIONS"
);
}
const key = { PK: `USER#${userId}`, SK: `DOC#${documentId}` };
const { Item: existingItem } = await docClient.send(
new GetCommand({ TableName: TABLE_NAME, Key: key })
);
if (!existingItem) {
return formatResponse(
404,
false,
null,
"Document not found.",
"DYNAMODB",
"NOT_FOUND"
);
}
if (!isReviewableDocumentStatus(existingItem.status)) {
return formatResponse(
409,
false,
null,
"Document is not in a reviewable status.",
"DYNAMODB",
"DOCUMENT_STATE_CONFLICT"
);
}
if (!existingItem.processedS3Key) {
return formatResponse(
409,
false,
null,
"Processed document is not available for review.",
"S3_PROCESSED",
"PROCESSED_DOCUMENT_MISSING"
);
}
const updatedAt = new Date().toISOString();
const reviewerNote = normalizeReviewerNote(body.reviewerNote);
let processedUpdate;
try {
processedUpdate = await updateProcessedDocument({
corrections: corrections.items,
documentId,
processedS3Key: existingItem.processedS3Key,
reviewStatus,
reviewedBy: userId,
reviewerNote,
updatedAt,
userId,
});
} catch (error) {
log("ERROR", {
documentId,
userId,
errorName: error?.name,
errorMessage: error?.message,
message: "Processed S3 document update failed",
});
return formatResponse(
error?.name === "PreconditionFailed" ? 409 : 502,
false,
null,
error?.name === "PreconditionFailed"
? "Processed document changed before the review could be saved."
: "Could not update the processed document in S3.",
"S3_PROCESSED",
error?.name === "PreconditionFailed"
? "DOCUMENT_STATE_CONFLICT"
: "S3_UPDATE_FAILED"
);
}
let updatedItem;
try {
const updateInput = buildDynamoDbUpdate({
corrections: corrections.items,
expectedStatus: existingItem.status,
key,
reviewStatus,
reviewedBy: userId,
reviewerNote,
updatedAt,
});
const result = await docClient.send(new UpdateCommand(updateInput));
updatedItem = result.Attributes;
} catch (error) {
const isConflict = error?.name === "ConditionalCheckFailedException";
if (processedUpdate) {
try {
await rollbackProcessedDocument(processedUpdate);
} catch (rollbackError) {
log("ERROR", {
documentId,
userId,
errorName: rollbackError?.name,
errorMessage: rollbackError?.message,
message: "Processed S3 document rollback failed",
});
}
}
log("ERROR", {
documentId,
userId,
errorName: error?.name,
errorMessage: error?.message,
message: "Review metadata update failed",
});
return formatResponse(
isConflict ? 409 : 500,
false,
null,
isConflict
? "Document status changed before the review could be saved."
: "Could not update review metadata.",
"DYNAMODB",
isConflict ? "DOCUMENT_STATE_CONFLICT" : "DYNAMODB_UPDATE_FAILED"
);
}
log("INFO", {
documentId,
userId,
reviewStatus,
correctionCount: corrections.items.length,
message: "Document review updated",
});
return formatResponse(200, true, {
documentId,
status: updatedItem?.status || reviewStatus,
reviewStatus: updatedItem?.reviewStatus || reviewStatus,
updatedAt: updatedItem?.updatedAt || updatedAt,
});
} catch (error) {
log("ERROR", {
errorName: error?.name,
errorMessage: error?.message,
message: "Unhandled review update error",
});
return formatResponse(
500,
false,
null,
error?.message || "Unknown error.",
"REVIEW",
"UNKNOWN_ERROR"
);
}
};
async function updateProcessedDocument({
corrections,
documentId,
processedS3Key,
reviewStatus,
reviewedBy,
reviewerNote,
updatedAt,
userId,
}) {
const location = parseS3Location(processedS3Key, PROCESSED_BUCKET);
if (
location.bucket !== PROCESSED_BUCKET ||
!location.key?.startsWith(`processed/${userId}/${documentId}/`)
) {
throw new Error("processedS3Key is invalid.");
}
const response = await s3Client.send(
new GetObjectCommand({ Bucket: location.bucket, Key: location.key })
);
if (!response.Body) throw new Error("Processed S3 document is empty.");
const originalBody = await response.Body.transformToString();
const document = parsePossiblyDoubleStringifiedJson(originalBody);
if (!isRecord(document)) {
throw new Error("Processed S3 document must contain a JSON object.");
}
document.status = reviewStatus;
document.review = isRecord(document.review) ? document.review : {};
document.review.reviewStatus = reviewStatus;
document.review.reviewedAt = updatedAt;
document.review.reviewedBy = reviewedBy;
document.review.reviewerNote = reviewerNote;
document.review.corrections = corrections;
document.review.reviewReasonCodes = [];
document.audit = isRecord(document.audit) ? document.audit : {};
document.audit.updatedAt = updatedAt;
for (const correction of corrections) {
applyCorrection(document, correction);
}
const putResult = await s3Client.send(
new PutObjectCommand({
Bucket: location.bucket,
Key: location.key,
Body: JSON.stringify(document),
ContentType: "application/json",
Metadata: {
...(response.Metadata || {}),
"document-id": documentId,
},
IfMatch: response.ETag,
})
);
return {
bucket: location.bucket,
key: location.key,
originalBody,
originalContentType: response.ContentType || "application/json",
originalMetadata: response.Metadata || {},
updatedETag: putResult.ETag,
};
}
async function rollbackProcessedDocument(update) {
if (!update.updatedETag) {
throw new Error("Updated S3 ETag is unavailable for safe rollback.");
}
await s3Client.send(
new PutObjectCommand({
Bucket: update.bucket,
Key: update.key,
Body: update.originalBody,
ContentType: update.originalContentType,
Metadata: update.originalMetadata,
IfMatch: update.updatedETag,
})
);
}
function buildDynamoDbUpdate({
corrections,
expectedStatus,
key,
reviewStatus,
reviewedBy,
reviewerNote,
updatedAt,
}) {
const names = {
"#status": "status",
"#reviewStatus": "reviewStatus",
};
const values = {
":status": reviewStatus,
":reviewStatus": reviewStatus,
":gsi1pk": `STATUS#${reviewStatus}`,
":gsi1sk": updatedAt,
":updatedAt": updatedAt,
":reviewedAt": updatedAt,
":reviewedBy": reviewedBy,
":reviewerNote": reviewerNote,
":reviewReasonCodes": [],
":expectedStatus": expectedStatus,
};
const assignments = [
"#status = :status",
"#reviewStatus = :reviewStatus",
"GSI1PK = :gsi1pk",
"GSI1SK = :gsi1sk",
"updatedAt = :updatedAt",
"reviewedAt = :reviewedAt",
"reviewedBy = :reviewedBy",
"reviewerNote = :reviewerNote",
"reviewReasonCodes = :reviewReasonCodes",
];
for (const correction of corrections) {
if (!DYNAMODB_SUMMARY_FIELDS.has(correction.fieldName)) continue;
const nameToken = `#correction${assignments.length}`;
const valueToken = `:correction${assignments.length}`;
names[nameToken] = correction.fieldName;
values[valueToken] = correction.newValue;
assignments.push(`${nameToken} = ${valueToken}`);
}
return {
TableName: TABLE_NAME,
Key: key,
UpdateExpression: `SET ${assignments.join(", ")}`,
ConditionExpression:
"attribute_exists(PK) AND attribute_exists(SK) AND #status = :expectedStatus",
ExpressionAttributeNames: names,
ExpressionAttributeValues: {
...values,
},
ReturnValues: "ALL_NEW",
};
}
function applyCorrection(document, correction) {
if (correction.fieldName === "documentType") {
document.documentType = correction.newValue;
return;
}
if (correction.fieldName === "lineItems") {
document.lineItems = correction.newValue;
return;
}
document.invoice = isRecord(document.invoice) ? document.invoice : {};
document.invoice[correction.fieldName] = correction.newValue;
}
function normalizeCorrections(value) {
if (value === undefined) return { items: [] };
if (!Array.isArray(value)) return { error: "corrections must be an array." };
if (value.length > 100) {
return { error: "A maximum of 100 corrections is allowed." };
}
if (Buffer.byteLength(JSON.stringify(value), "utf8") > 100_000) {
return { error: "Corrections payload is too large." };
}
const itemsByField = new Map();
for (const correction of value) {
if (
!isRecord(correction) ||
!ALLOWED_CORRECTION_FIELDS.has(correction.fieldName) ||
!("newValue" in correction)
) {
return { error: "Each correction must contain an allowed fieldName and newValue." };
}
if (correction.fieldName === "lineItems" && !Array.isArray(correction.newValue)) {
return { error: "The lineItems correction must contain an array." };
}
const valueError = validateCorrectionValue(
correction.fieldName,
correction.newValue
);
if (valueError) return { error: valueError };
itemsByField.set(correction.fieldName, {
fieldName: correction.fieldName,
oldValue: correction.oldValue ?? null,
newValue: normalizeCorrectionValue(
correction.fieldName,
correction.newValue
),
});
}
return { items: [...itemsByField.values()] };
}
function normalizeCorrectionValue(fieldName, value) {
if (fieldName === "lineItems") {
return value.map((item, index) => ({
lineItemId: String(item.lineItemId || `item-${String(index + 1).padStart(3, "0")}`),
description: String(item.description || "").trim(),
quantity: Number(item.quantity || 0),
unitPriceAmount: Number(item.unitPriceAmount || 0),
taxAmount: Number(item.taxAmount || 0),
totalAmount: Number(item.totalAmount || 0),
confidenceScore: Math.max(
0,
Math.min(1, Number(item.confidenceScore ?? 1))
),
}));
}
if (fieldName === "currency" || fieldName === "documentType") {
return String(value).trim().toUpperCase();
}
if (fieldName.endsWith("Amount")) return Number(value);
return String(value).trim();
}
export function validateCorrectionValue(fieldName, value) {
if (fieldName === "documentType") {
return ["INVOICE", "RECEIPT"].includes(String(value || "").trim().toUpperCase())
? null
: "documentType must be INVOICE or RECEIPT.";
}
if (fieldName === "lineItems") {
if (value.length > 200) return "A maximum of 200 line items is allowed.";
for (const item of value) {
if (!isRecord(item)) return "Each line item must be an object.";
if (String(item.description || "").trim().length > 500) {
return "Line item descriptions must not exceed 500 characters.";
}
for (const field of [
"quantity",
"unitPriceAmount",
"taxAmount",
"totalAmount",
"confidenceScore",
]) {
if (
item[field] !== undefined &&
item[field] !== null &&
!Number.isFinite(Number(item[field]))
) {
return `Line item ${field} must be numeric.`;
}
}
}
return null;
}
if (fieldName === "currency") {
return /^[A-Z]{3}$/.test(String(value || "").trim().toUpperCase())
? null
: "currency must be a three-letter ISO code.";
}
if (fieldName.endsWith("Amount")) {
return Number.isFinite(Number(value))
? null
: `${fieldName} must be numeric.`;
}
if (fieldName === "invoiceDate" || fieldName === "dueDate") {
return /^\d{4}-\d{2}-\d{2}$/.test(String(value || ""))
? null
: `${fieldName} must use YYYY-MM-DD format.`;
}
return typeof value === "string" && value.trim().length <= 500
? null
: `${fieldName} must be a string of at most 500 characters.`;
}
function normalizeReviewerNote(value) {
if (typeof value !== "string") return null;
const note = value.trim();
return note ? note.slice(0, 2000) : null;
}
function parseBody(event) {
if (!event?.body) return null;
if (isRecord(event.body)) return event.body;
try {
const value = event.isBase64Encoded
? Buffer.from(event.body, "base64").toString("utf8")
: event.body;
const parsed = JSON.parse(value);
return isRecord(parsed) ? parsed : null;
} catch {
return null;
}
}
function getUserId(event) {
return (
event?.requestContext?.authorizer?.jwt?.claims?.sub ||
event?.requestContext?.authorizer?.claims?.sub ||
null
);
}
function parseS3Location(value, fallbackBucket) {
if (typeof value !== "string" || !value.trim()) {
return { bucket: null, key: null };
}
if (!value.startsWith("s3://")) {
return { bucket: fallbackBucket, key: value.replace(/^\/+/, "") };
}
const location = value.slice(5);
const slashIndex = location.indexOf("/");
if (slashIndex < 1) return { bucket: location || null, key: null };
return {
bucket: location.slice(0, slashIndex),
key: location.slice(slashIndex + 1),
};
}
function parsePossiblyDoubleStringifiedJson(value) {
let parsed = JSON.parse(value);
if (typeof parsed === "string") parsed = JSON.parse(parsed);
return parsed;
}
function validateEnvironment() {
const missing = [];
if (!TABLE_NAME) missing.push("DOCUFLOW_DEV_TABLE_NAME");
if (!PROCESSED_BUCKET) missing.push("DOCUFLOW_DEV_PROCESSED_BUCKET");
if (!missing.length) return null;
return formatResponse(
500,
false,
null,
`Missing environment variables: ${missing.join(", ")}.`,
"CONFIGURATION",
"MISSING_ENVIRONMENT_VARIABLE"
);
}
function formatResponse(
statusCode,
success,
data,
errorMessage = null,
errorStage = null,
errorCode = "UNKNOWN_ERROR"
) {
const body = { success, data, error: null };
if (errorMessage) body.error = { errorCode, errorMessage, errorStage };
return {
statusCode,
headers: {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "Authorization,Content-Type",
"Access-Control-Allow-Methods": "OPTIONS,PATCH",
"Content-Type": "application/json",
},
body: JSON.stringify(body),
};
}
function isRecord(value) {
return Boolean(value) && typeof value === "object" && !Array.isArray(value);
}
function log(level, data) {
const writer = level === "ERROR" ? console.error : console.log;
writer(
JSON.stringify({
level,
service: "docuflow-dev-data-review-update-lambda",
...data,
})
);
}
Extended Lambdas (Optional project additions)
The following three functions are not part of the approved MVP Lambda inventory. They are additional project features for document deletion, Dashboard data, and explicit process/retry controls. Deploy them only when the corresponding extended frontend and API routes are used.
EXTENDED LAMBDA E1: Create Delete Document Lambda
This function is used to delete the metadata of one or more user documents in DynamoDB and their associated files on S3 Raw & Processed.
Click Create function ➔ Author from scratch.
Configure basic information:
- Function name: Enter exactly
docuflow-dev-data-delete-lambda. - Runtime: Choose the standard programming language for the project (
Node.js 24.x).
- Architecture: Select
arm64.
- Function name: Enter exactly
Configure Permissions:
- Expand the Additional settings section.
- Under Custom execution role, select Use an existing role and choose the
docuflow-dev-data-lambda-role.
Add tags:
- Click on tags, select the necessary tags according to the setup.
- Click Save.
- Click on tags, select the necessary tags according to the setup.
Click the Create function button.
This function is used to delete the data of 1 or more user documents that the user wants to delete.
Source Code (index.mjs):
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import {
DynamoDBDocumentClient,
GetCommand,
DeleteCommand,
} from "@aws-sdk/lib-dynamodb";
import {
S3Client,
ListObjectsV2Command,
DeleteObjectsCommand,
} from "@aws-sdk/client-s3";
const docClient = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const s3Client = new S3Client({});
const TABLE_NAME = process.env.DOCUFLOW_DEV_TABLE_NAME;
const PROCESSED_BUCKET = process.env.DOCUFLOW_DEV_PROCESSED_BUCKET;
const RAW_BUCKET =
process.env.DOCUFLOW_DEV_RAW_BUCKET ||
process.env.DOCUFLOW_DEV_RAW_BUCKET_NAME;
const DELETE_CONCURRENCY = 10;
export const handler = async (event) => {
try {
const method = event?.requestContext?.http?.method || event?.httpMethod || "";
if (method === "OPTIONS") return formatResponse(200, true, null);
if (method !== "DELETE") {
return formatResponse(
405,
false,
null,
"Only DELETE is supported.",
"API",
"METHOD_NOT_ALLOWED"
);
}
const missingEnvironment = [
["DOCUFLOW_DEV_TABLE_NAME", TABLE_NAME],
["DOCUFLOW_DEV_RAW_BUCKET", RAW_BUCKET],
["DOCUFLOW_DEV_PROCESSED_BUCKET", PROCESSED_BUCKET],
]
.filter(([, value]) => !value)
.map(([name]) => name);
if (missingEnvironment.length) {
return formatResponse(
500,
false,
null,
`Missing environment variables: ${missingEnvironment.join(", ")}.`,
"CONFIGURATION",
"MISSING_ENVIRONMENT_VARIABLE"
);
}
const userId = getUserId(event);
if (!userId) {
return formatResponse(
401,
false,
null,
"Missing authenticated Cognito user.",
"AUTH",
"UNAUTHORIZED"
);
}
const body = parseBody(event);
if (body === null) {
return formatResponse(
400,
false,
null,
"Request body must be valid JSON.",
"VALIDATION",
"INVALID_INPUT"
);
}
const pathDocumentId = event?.pathParameters?.documentId;
const isSingleDelete = Boolean(pathDocumentId);
const idsToDelete = uniqueDocumentIds(
pathDocumentId ? [pathDocumentId] : body?.documentIds
);
if (!idsToDelete.length) {
return formatResponse(
400,
false,
null,
"Missing documentId or documentIds.",
"VALIDATION",
"INVALID_INPUT"
);
}
if (idsToDelete.length > 100) {
return formatResponse(
400,
false,
null,
"A maximum of 100 documents can be deleted per request.",
"VALIDATION",
"INVALID_INPUT"
);
}
const settledResults = await mapWithConcurrency(
idsToDelete,
DELETE_CONCURRENCY,
(documentId) => deleteDocument(userId, documentId)
);
const results = { successful: [], failed: [] };
for (const result of settledResults) {
if (result.success) results.successful.push(result.documentId);
else {
results.failed.push({
documentId: result.documentId,
reason: result.reason,
});
}
}
if (isSingleDelete) {
if (results.successful.length) {
const deletedAt = new Date().toISOString();
return formatResponse(200, true, {
documentId: idsToDelete[0],
deleted: true,
deletedAt,
});
}
const failure = results.failed[0];
return formatResponse(
failure?.reason === "NOT_FOUND" ? 404 : 500,
false,
null,
failure?.reason === "NOT_FOUND"
? "Document not found."
: "Document could not be deleted.",
failure?.reason === "NOT_FOUND" ? "DYNAMODB" : "DELETE",
failure?.reason === "NOT_FOUND" ? "NOT_FOUND" : "DELETE_FAILED"
);
}
const deletedAt = new Date().toISOString();
return formatResponse(results.failed.length ? 207 : 200, true, {
documentIds: results.successful,
deletedCount: results.successful.length,
deletedAt,
failed: results.failed,
});
} catch (error) {
log("ERROR", {
message: "Unhandled delete error",
errorName: error?.name,
errorMessage: error?.message,
});
return formatResponse(
500,
false,
null,
error?.message || "Unknown error.",
"DELETE",
"UNKNOWN_ERROR"
);
}
};
async function deleteDocument(userId, documentId) {
try {
const key = { PK: `USER#${userId}`, SK: `DOC#${documentId}` };
const { Item: item } = await docClient.send(
new GetCommand({ TableName: TABLE_NAME, Key: key })
);
if (!item) return { documentId, success: false, reason: "NOT_FOUND" };
await Promise.all([
emptyS3Directory(RAW_BUCKET, `raw/${userId}/${documentId}/`),
emptyS3Directory(
PROCESSED_BUCKET,
`processed/${userId}/${documentId}/`
),
]);
await docClient.send(
new DeleteCommand({
TableName: TABLE_NAME,
Key: key,
ConditionExpression: "attribute_exists(PK) AND attribute_exists(SK)",
})
);
log("INFO", { documentId, userId, message: "Document deleted" });
return { documentId, success: true };
} catch (error) {
log("ERROR", {
documentId,
userId,
errorName: error?.name,
errorMessage: error?.message,
message: "Document deletion failed",
});
return {
documentId,
success: false,
reason: error?.name || error?.message || "DELETE_FAILED",
};
}
}
async function emptyS3Directory(bucket, prefix) {
let continuationToken;
do {
const listResult = await s3Client.send(
new ListObjectsV2Command({
Bucket: bucket,
Prefix: prefix,
ContinuationToken: continuationToken,
})
);
const objects = (listResult.Contents || [])
.filter((item) => item.Key)
.map((item) => ({ Key: item.Key }));
if (objects.length) {
const deleteResult = await s3Client.send(
new DeleteObjectsCommand({
Bucket: bucket,
Delete: { Objects: objects, Quiet: true },
})
);
if (deleteResult.Errors?.length) {
const failedKeys = deleteResult.Errors.map((item) => item.Key)
.filter(Boolean)
.slice(0, 10);
throw new Error(
`S3_DELETE_PARTIAL_FAILURE:${failedKeys.join(",")}`
);
}
}
continuationToken = listResult.IsTruncated
? listResult.NextContinuationToken
: undefined;
} while (continuationToken);
}
async function mapWithConcurrency(values, concurrency, worker) {
const results = new Array(values.length);
let nextIndex = 0;
async function runWorker() {
while (nextIndex < values.length) {
const index = nextIndex;
nextIndex += 1;
results[index] = await worker(values[index]);
}
}
await Promise.all(
Array.from(
{ length: Math.min(concurrency, values.length) },
() => runWorker()
)
);
return results;
}
function getUserId(event) {
return (
event?.requestContext?.authorizer?.jwt?.claims?.sub ||
event?.requestContext?.authorizer?.claims?.sub ||
null
);
}
function parseBody(event) {
if (!event?.body) return {};
if (typeof event.body === "object" && event.body !== null) return event.body;
try {
const bodyText = event.isBase64Encoded
? Buffer.from(event.body, "base64").toString("utf8")
: event.body;
const parsed = JSON.parse(bodyText);
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
? parsed
: null;
} catch {
return null;
}
}
function uniqueDocumentIds(values) {
if (!Array.isArray(values)) return [];
return [
...new Set(
values
.filter((value) => typeof value === "string")
.map((value) => value.trim())
.filter(Boolean)
),
];
}
function formatResponse(
statusCode,
success,
data,
errorMessage = null,
errorStage = null,
errorCode = "UNKNOWN_ERROR"
) {
const body = { success, data, error: null };
if (errorMessage) body.error = { errorCode, errorMessage, errorStage };
return {
statusCode,
headers: {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "Authorization,Content-Type",
"Access-Control-Allow-Methods": "OPTIONS,DELETE",
"Content-Type": "application/json",
},
body: JSON.stringify(body),
};
}
function log(level, data) {
const writer = level === "ERROR" ? console.error : console.log;
writer(
JSON.stringify({
level,
service: "docuflow-dev-data-delete-lambda",
...data,
})
);
}
EXTENDED LAMBDA E2: CREATE DASHBOARD LAMBDA
The dashboard handler aggregates KPIs, recent activity, notifications, and status distribution for the frontend Dashboard. Create docuflow-dev-data-dashboard-lambda using the same runtime, architecture, and execution role pattern as the other Data Lambdas.
Source Code (index.mjs):
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import {
DynamoDBDocumentClient,
QueryCommand,
UpdateCommand,
} from "@aws-sdk/lib-dynamodb";
const docClient = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const TABLE_NAME = process.env.DOCUFLOW_DEV_TABLE_NAME;
const DEFAULT_PAGE_SIZE = 50;
const MAX_PAGE_SIZE = 100;
const DEFAULT_VND_EXCHANGE_RATES = {
VND: 1,
USD: 25000,
EUR: 27000,
GBP: 31500,
JPY: 170,
CNY: 3450,
KRW: 18,
SGD: 18500,
THB: 690,
AUD: 16500,
CAD: 18200,
CHF: 28000,
HKD: 3200,
INR: 300,
IDR: 1.55,
MYR: 5300,
PHP: 430,
TWD: 780,
};
const VND_EXCHANGE_RATES = loadExchangeRates();
export const handler = async (event) => {
try {
const method = getMethod(event);
if (method === "OPTIONS") {
return formatResponse(200, true, null);
}
if (method !== "GET" && method !== "PATCH") {
return formatResponse(
405,
false,
null,
"Only GET and PATCH are supported.",
"API",
"METHOD_NOT_ALLOWED"
);
}
if (!TABLE_NAME) {
return formatResponse(
500,
false,
null,
"DOCUFLOW_DEV_TABLE_NAME environment variable is missing.",
"CONFIGURATION",
"MISSING_ENVIRONMENT_VARIABLE"
);
}
const userId = getUserId(event);
if (!userId) {
return formatResponse(
401,
false,
null,
"Missing authenticated Cognito user.",
"AUTH",
"UNAUTHORIZED"
);
}
const route = getRoute(event);
if (method === "PATCH" && route === "NOTIFICATION_DETAIL") {
return acknowledgeNotification(event, userId);
}
if (method === "GET" && route === "NOTIFICATIONS") {
const documents = await queryUserItems(userId, "DOC#");
const acknowledgements = await queryUserItems(userId, "NOTIF#");
const notifications = buildNotifications(documents, acknowledgements);
return formatResponse(
200,
true,
paginate(notifications, event?.queryStringParameters)
);
}
if (method === "GET" && route === "ACTIVITY") {
const documents = await queryUserItems(userId, "DOC#");
return formatResponse(
200,
true,
paginate(buildActivity(documents), event?.queryStringParameters)
);
}
if (method === "GET" && route === "REPORTS_SUMMARY") {
const documents = await queryUserItems(userId, "DOC#");
return formatResponse(200, true, buildReportSummary(documents));
}
return formatResponse(
404,
false,
null,
"Route not found.",
"API",
"NOT_FOUND"
);
} catch (error) {
logError("Unhandled dashboard error.", error);
return formatResponse(
500,
false,
null,
error?.message || "Unknown error.",
"DYNAMODB",
"UNKNOWN_ERROR"
);
}
};
async function acknowledgeNotification(event, userId) {
const notificationId = event?.pathParameters?.notificationId;
if (
!notificationId ||
notificationId.length > 240 ||
!/^notif-[A-Za-z0-9._:-]+$/.test(notificationId)
) {
return formatResponse(
400,
false,
null,
"Invalid notificationId.",
"VALIDATION",
"INVALID_INPUT"
);
}
const now = new Date().toISOString();
await docClient.send(
new UpdateCommand({
TableName: TABLE_NAME,
Key: {
PK: `USER#${userId}`,
SK: `NOTIF#${notificationId}`,
},
UpdateExpression:
"SET notificationId = :notificationId, unread = :unread, acknowledgedAt = :now, updatedAt = :now, createdAt = if_not_exists(createdAt, :now)",
ExpressionAttributeValues: {
":notificationId": notificationId,
":unread": false,
":now": now,
},
})
);
return formatResponse(200, true, {
notificationId,
unread: false,
acknowledgedAt: now,
});
}
async function queryUserItems(userId, sortKeyPrefix) {
const items = [];
let exclusiveStartKey;
do {
const result = await docClient.send(
new QueryCommand({
TableName: TABLE_NAME,
KeyConditionExpression: "PK = :pk AND begins_with(SK, :prefix)",
ExpressionAttributeValues: {
":pk": `USER#${userId}`,
":prefix": sortKeyPrefix,
},
ExclusiveStartKey: exclusiveStartKey,
})
);
items.push(...(result.Items || []));
exclusiveStartKey = result.LastEvaluatedKey;
} while (exclusiveStartKey);
return items;
}
function buildNotifications(documents, acknowledgementItems) {
const acknowledgedIds = new Set(
acknowledgementItems
.filter((item) => item.unread === false)
.map((item) => item.notificationId)
.filter(Boolean)
);
const notifications = documents
.map(toDocumentSummary)
.map((document) => notificationForDocument(document, acknowledgedIds))
.filter(Boolean);
return notifications.sort(compareNewestFirst);
}
function notificationForDocument(document, acknowledgedIds) {
const timestamp = document.updatedAt || document.createdAt || epochIso();
if (document.status === "REVIEW_REQUIRED") {
const id = `notif-${document.documentId}-review`;
return {
id,
documentId: document.documentId,
kind: "ACTION",
title: "Yêu cầu kiểm duyệt",
body: document.reviewReasonCodes.length
? document.reviewReasonCodes.join(", ")
: "Tài liệu cần được kiểm tra lại.",
timestamp,
unread: !acknowledgedIds.has(id),
requiresAction: true,
severity: "warning",
document,
};
}
if (document.status === "FAILED") {
const id = `notif-${document.documentId}-failed`;
return {
id,
documentId: document.documentId,
kind: "FAILED",
title: "Xử lý tài liệu thất bại",
body:
document.errorMessage ||
"Pipeline xử lý tài liệu bị lỗi. Có thể chạy lại tài liệu này.",
timestamp,
unread: !acknowledgedIds.has(id),
requiresAction: true,
severity: "critical",
document,
};
}
if (["APPROVED", "CORRECTED", "EXTRACTED"].includes(document.status)) {
return {
id: `notif-${document.documentId}-complete`,
documentId: document.documentId,
kind: "COMPLETE",
title:
document.status === "APPROVED"
? "Tài liệu đã được phê duyệt"
: "Tài liệu đã xử lý xong",
body: `${document.originalFileName} đã hoàn tất bước ${document.status}.`,
timestamp,
unread: false,
requiresAction: false,
severity: "success",
document,
};
}
if (["QUEUED", "PROCESSING"].includes(document.status)) {
return {
id: `notif-${document.documentId}-processing`,
documentId: document.documentId,
kind: "PROCESSING",
title:
document.status === "QUEUED"
? "Tài liệu đang chờ xử lý"
: "Tài liệu đang được xử lý",
body: `${document.originalFileName} hiện có trạng thái ${document.status}.`,
timestamp,
unread: false,
requiresAction: false,
severity: "info",
document,
};
}
return null;
}
function buildActivity(documents) {
return documents
.map(toDocumentSummary)
.map((document) => ({
id: `activity-${document.documentId}-${document.status.toLowerCase()}`,
documentId: document.documentId,
kind: activityKind(document.status),
title: activityTitle(document.status),
detail: `${document.originalFileName} hiện có trạng thái ${document.status}.`,
timestamp: document.updatedAt || document.createdAt || epochIso(),
actor:
["APPROVED", "CORRECTED"].includes(document.status) &&
document.reviewedBy
? document.reviewedBy
: "system",
source: activitySource(document.status),
severity: activitySeverity(document.status),
document,
}))
.sort(compareNewestFirst);
}
function activityKind(status) {
if (["APPROVED", "CORRECTED"].includes(status)) return "APPROVAL";
if (status === "REVIEW_REQUIRED") return "REVIEW";
if (["PROCESSING", "EXTRACTED", "FAILED"].includes(status)) {
return "PROCESSING";
}
return "UPLOAD";
}
function activityTitle(status) {
switch (status) {
case "UPLOADED":
return "Tài liệu đã được tải lên";
case "QUEUED":
return "Tài liệu được đưa vào hàng đợi";
case "PROCESSING":
return "Tài liệu đang được xử lý";
case "EXTRACTED":
return "Tài liệu đã trích xuất xong";
case "REVIEW_REQUIRED":
return "Tài liệu cần kiểm duyệt";
case "FAILED":
return "Tài liệu xử lý thất bại";
case "CORRECTED":
return "Tài liệu đã được hiệu chỉnh";
case "APPROVED":
return "Tài liệu đã được phê duyệt";
default:
return "Cập nhật tài liệu";
}
}
function activitySeverity(status) {
if (status === "FAILED") return "error";
if (status === "REVIEW_REQUIRED") return "warning";
if (["EXTRACTED", "CORRECTED", "APPROVED"].includes(status)) {
return "success";
}
return "info";
}
function activitySource(status) {
if (["APPROVED", "CORRECTED", "REVIEW_REQUIRED"].includes(status)) {
return "DocuFlow Review";
}
return "DocuFlow Workflow";
}
function buildReportSummary(documents) {
const summaries = documents.map(toDocumentSummary);
const approved = summaries.filter((document) => document.status === "APPROVED");
const pending = summaries.filter((document) => document.status !== "APPROVED");
const totalAmounts = summarizeAmounts(summaries);
const approvedAmounts = summarizeAmounts(approved);
const pendingAmounts = summarizeAmounts(pending);
const confidenceValues = summaries
.map((document) => normalizeConfidence(document.confidenceScore))
.filter((value) => value !== null);
return {
totalDocuments: summaries.length,
approvedDocuments: approved.length,
reviewRequiredDocuments: summaries.filter(
(document) => document.status === "REVIEW_REQUIRED"
).length,
failedDocuments: summaries.filter(
(document) => document.status === "FAILED"
).length,
totalAmountVnd: totalAmounts.convertedVnd,
approvedAmountVnd: approvedAmounts.convertedVnd,
pendingAmountVnd: pendingAmounts.convertedVnd,
averageConfidence: confidenceValues.length
? confidenceValues.reduce((sum, value) => sum + value, 0) /
confidenceValues.length
: 0,
amountsByCurrency: totalAmounts.byCurrency,
approvedAmountsByCurrency: approvedAmounts.byCurrency,
pendingAmountsByCurrency: pendingAmounts.byCurrency,
unconvertedCurrencies: Array.from(
new Set([
...totalAmounts.unconvertedCurrencies,
...approvedAmounts.unconvertedCurrencies,
...pendingAmounts.unconvertedCurrencies,
])
).sort(),
exchangeRateSource: process.env.DOCUFLOW_DEV_VND_EXCHANGE_RATES
? "DOCUFLOW_DEV_VND_EXCHANGE_RATES"
: "DOCUFLOW_DEMO_STATIC_RATES",
generatedAt: new Date().toISOString(),
};
}
function summarizeAmounts(documents) {
const byCurrency = {};
const unconvertedCurrencies = new Set();
let convertedVnd = 0;
for (const document of documents) {
const amount = finiteNumber(document.totalAmount);
if (amount === null) continue;
const currency = normalizeCurrency(document.currency);
byCurrency[currency] = (byCurrency[currency] || 0) + amount;
const rate = VND_EXCHANGE_RATES[currency];
if (Number.isFinite(rate)) {
convertedVnd += amount * rate;
} else {
unconvertedCurrencies.add(currency);
}
}
return {
byCurrency,
convertedVnd,
unconvertedCurrencies: Array.from(unconvertedCurrencies),
};
}
function toDocumentSummary(item) {
const safe = removeKeys(item);
return {
documentId: String(safe.documentId || ""),
originalFileName: String(safe.originalFileName || "original.pdf"),
documentType: String(safe.documentType || "UNKNOWN"),
status: String(safe.status || "UNKNOWN").toUpperCase(),
vendorName: String(safe.vendorName || ""),
invoiceDate: String(safe.invoiceDate || ""),
currency: normalizeCurrency(safe.currency),
totalAmount: finiteNumber(safe.totalAmount) || 0,
confidenceScore: finiteNumber(safe.confidenceScore) || 0,
reviewStatus: safe.reviewStatus || null,
reviewReasonCodes: normalizeStringArray(safe.reviewReasonCodes),
reviewedBy: safe.reviewedBy || null,
errorMessage: safe.errorMessage || null,
createdAt: safe.createdAt || null,
updatedAt: safe.updatedAt || null,
};
}
function paginate(items, query = {}) {
const requestedLimit = Number.parseInt(query?.limit || "", 10);
const limit = Number.isFinite(requestedLimit)
? Math.min(Math.max(requestedLimit, 1), MAX_PAGE_SIZE)
: DEFAULT_PAGE_SIZE;
const offset = decodeNextToken(query?.nextToken);
const pageItems = items.slice(offset, offset + limit);
const nextOffset = offset + pageItems.length;
return {
items: pageItems,
nextToken:
nextOffset < items.length ? encodeNextToken(nextOffset) : null,
};
}
function encodeNextToken(offset) {
return Buffer.from(JSON.stringify({ version: 1, offset }), "utf8").toString(
"base64url"
);
}
function decodeNextToken(token) {
if (!token) return 0;
try {
const value = JSON.parse(
Buffer.from(String(token), "base64url").toString("utf8")
);
return Number.isSafeInteger(value?.offset) && value.offset >= 0
? value.offset
: 0;
} catch {
return 0;
}
}
function loadExchangeRates() {
const configured = process.env.DOCUFLOW_DEV_VND_EXCHANGE_RATES;
if (!configured) return DEFAULT_VND_EXCHANGE_RATES;
try {
const parsed = JSON.parse(configured);
return Object.fromEntries(
Object.entries(parsed)
.map(([currency, rate]) => [
normalizeCurrency(currency),
Number(rate),
])
.filter(([, rate]) => Number.isFinite(rate) && rate > 0)
);
} catch (error) {
logError(
"Invalid DOCUFLOW_DEV_VND_EXCHANGE_RATES; using demo rates.",
error
);
return DEFAULT_VND_EXCHANGE_RATES;
}
}
function normalizeConfidence(value) {
const numeric = finiteNumber(value);
if (numeric === null || numeric < 0) return null;
return Math.min(numeric > 1 ? numeric / 100 : numeric, 1);
}
function normalizeCurrency(value) {
const currency = String(value || "").trim().toUpperCase();
return /^[A-Z]{3}$/.test(currency) ? currency : "UNKNOWN";
}
function normalizeStringArray(value) {
if (Array.isArray(value)) return value.map(String).filter(Boolean);
if (!value) return [];
try {
const parsed = JSON.parse(value);
return Array.isArray(parsed) ? parsed.map(String).filter(Boolean) : [];
} catch {
return [];
}
}
function finiteNumber(value) {
if (value === null || value === undefined || value === "") return null;
const numeric = Number(value);
return Number.isFinite(numeric) ? numeric : null;
}
function compareNewestFirst(left, right) {
return new Date(right.timestamp).getTime() - new Date(left.timestamp).getTime();
}
function epochIso() {
return new Date(0).toISOString();
}
function removeKeys(item = {}) {
const { PK, SK, GSI1PK, GSI1SK, ...safe } = item;
return safe;
}
function getUserId(event) {
const claims =
event?.requestContext?.authorizer?.jwt?.claims ||
event?.requestContext?.authorizer?.claims ||
{};
return claims.sub || null;
}
function getMethod(event) {
return event?.requestContext?.http?.method || event?.httpMethod || "";
}
function getRoute(event) {
const path = String(
event?.rawPath || event?.path || event?.resource || ""
).replace(/\/+$/, "");
if (/\/notifications\/[^/]+$/.test(path)) return "NOTIFICATION_DETAIL";
if (path.endsWith("/notifications")) return "NOTIFICATIONS";
if (path.endsWith("/activity")) return "ACTIVITY";
if (path.endsWith("/reports/summary")) return "REPORTS_SUMMARY";
return "UNKNOWN";
}
function formatResponse(
statusCode,
success,
data,
errorMessage = null,
errorStage = null,
errorCode = "UNKNOWN_ERROR"
) {
const body = { success, data, error: null };
if (errorMessage) {
body.error = { errorCode, errorMessage, errorStage };
}
return {
statusCode,
headers: {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "Authorization,Content-Type",
"Access-Control-Allow-Methods": "OPTIONS,GET,PATCH",
"Content-Type": "application/json",
},
body: JSON.stringify(body),
};
}
function logError(message, error, details = {}) {
console.error(
JSON.stringify({
level: "ERROR",
service: "docuflow-dev-data-dashboard-lambda",
message,
errorName: error?.name,
errorMessage: error?.message,
...details,
})
);
}
EXTENDED LAMBDA E3: CREATE PROCESS CONTROL LAMBDA
The process-control handler verifies document ownership, checks the raw S3 object, and starts Step Functions for process or retry requests. Create docuflow-dev-data-process-control-lambda with the same runtime and architecture; its execution role requires Raw S3 read access, DynamoDB read/write access, and states:StartExecution.
Source Code (index.mjs):
import { randomUUID } from "node:crypto";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import {
DynamoDBDocumentClient,
GetCommand,
UpdateCommand,
} from "@aws-sdk/lib-dynamodb";
import { HeadObjectCommand, S3Client } from "@aws-sdk/client-s3";
import { SFNClient, StartExecutionCommand } from "@aws-sdk/client-sfn";
const docClient = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const s3Client = new S3Client({});
const sfnClient = new SFNClient({});
const TABLE_NAME = process.env.DOCUFLOW_DEV_TABLE_NAME;
const RAW_BUCKET =
process.env.DOCUFLOW_DEV_RAW_BUCKET ||
process.env.DOCUFLOW_DEV_RAW_BUCKET_NAME;
const STATE_MACHINE_ARN =
process.env.DOCUFLOW_DEV_STATE_MACHINE_ARN ||
process.env.STATE_MACHINE_ARN;
const RETRYABLE_STATUSES = new Set(["FAILED", "REVIEW_REQUIRED"]);
const ACTIVE_STATUSES = new Set(["QUEUED", "PROCESSING"]);
export const handler = async (event) => {
try {
return await handleRequest(event);
} catch (error) {
logError("Unhandled process-control error.", error);
return formatResponse(
500,
false,
null,
error?.message || "Unknown error.",
"UNKNOWN",
"UNKNOWN_ERROR"
);
}
};
async function handleRequest(event) {
const method = getHttpMethod(event);
if (method === "OPTIONS") {
return formatResponse(200, true, null);
}
if (method !== "POST") {
return formatResponse(
405,
false,
null,
"Method not allowed.",
"API",
"METHOD_NOT_ALLOWED"
);
}
const envError = validateEnvironment();
if (envError) return envError;
const userId = getUserId(event);
if (!userId) {
return formatResponse(
401,
false,
null,
"Missing authenticated Cognito user.",
"AUTH",
"UNAUTHORIZED"
);
}
const documentId = event?.pathParameters?.documentId;
if (!documentId) {
return formatResponse(
400,
false,
null,
"Missing documentId.",
"VALIDATION",
"INVALID_INPUT"
);
}
const action = getRouteAction(event);
if (!action) {
return formatResponse(
404,
false,
null,
"Route not found.",
"API",
"NOT_FOUND"
);
}
const body = parseBody(event);
if (body === null) {
return formatResponse(
400,
false,
null,
"Request body must be valid JSON.",
"SCHEMA_VALIDATION",
"SCHEMA_VALIDATION_FAILED"
);
}
const key = {
PK: `USER#${userId}`,
SK: `DOC#${documentId}`,
};
let existingItem;
try {
const result = await docClient.send(
new GetCommand({
TableName: TABLE_NAME,
Key: key,
ConsistentRead: true,
})
);
existingItem = result.Item || null;
} catch (error) {
logError("Failed to load document metadata.", error, {
documentId,
userId,
action,
});
return formatResponse(
500,
false,
null,
"Could not load document metadata.",
"DYNAMODB",
"DYNAMODB_READ_FAILED"
);
}
if (action === "RETRY" && !existingItem) {
return formatResponse(
404,
false,
null,
"Document not found.",
"DYNAMODB",
"NOT_FOUND"
);
}
if (existingItem && ACTIVE_STATUSES.has(existingItem.status)) {
return formatResponse(200, true, {
documentId,
...(action === "RETRY"
? { retryStarted: false }
: { started: false }),
status: existingItem.status,
message: "Document is already queued or processing.",
executionId: existingItem.workflowExecutionArn || null,
});
}
if (
action === "RETRY" &&
!RETRYABLE_STATUSES.has(existingItem?.status)
) {
return formatResponse(
409,
false,
null,
`Document status ${existingItem?.status || "UNKNOWN"} cannot be retried.`,
"WORKFLOW",
"INVALID_DOCUMENT_STATUS"
);
}
const rawLocation = resolveRawLocation({
requestRawS3Key: body.rawS3Key,
storedRawS3Key:
existingItem?.rawS3Key ||
existingItem?.storage?.rawS3Key ||
existingItem?.raw?.s3Key,
});
if (rawLocation.error) {
return formatResponse(
rawLocation.statusCode,
false,
null,
rawLocation.error,
"S3_RAW",
rawLocation.errorCode
);
}
const expectedPrefix = `raw/${userId}/${documentId}/`;
if (!rawLocation.key.startsWith(expectedPrefix)) {
return formatResponse(
403,
false,
null,
"The raw S3 key does not belong to the authenticated document owner.",
"S3_RAW",
"RAW_KEY_FORBIDDEN"
);
}
let rawObject;
try {
rawObject = await s3Client.send(
new HeadObjectCommand({
Bucket: RAW_BUCKET,
Key: rawLocation.key,
})
);
} catch (error) {
const notFound =
error?.name === "NotFound" ||
error?.name === "NoSuchKey" ||
error?.$metadata?.httpStatusCode === 404;
logError("Raw document object validation failed.", error, {
documentId,
userId,
rawS3Key: rawLocation.key,
});
return formatResponse(
notFound ? 404 : 500,
false,
null,
notFound
? "Uploaded raw document was not found in S3."
: "Could not validate the uploaded raw document.",
"S3_RAW",
notFound ? "RAW_OBJECT_NOT_FOUND" : "S3_HEAD_FAILED"
);
}
const now = new Date().toISOString();
const requestId = randomUUID();
const objectMetadata = rawObject?.Metadata || {};
const originalFileName =
decodeMetadataValue(objectMetadata["original-file-name"]) ||
body.originalFileName ||
existingItem?.originalFileName ||
inferOriginalFileName(rawLocation.key);
const mimeType =
rawObject?.ContentType ||
body.mimeType ||
existingItem?.mimeType ||
existingItem?.contentType ||
inferMimeType(rawLocation.key);
const metadataDocumentType = normalizeDocumentType(
objectMetadata["document-type"]
);
const requestedDocumentType = normalizeDocumentType(
body.documentType || existingItem?.documentType
);
const documentType =
metadataDocumentType !== "UNKNOWN"
? metadataDocumentType
: requestedDocumentType;
const pageCount = normalizePageCount(
objectMetadata["page-count"] || body.pageCount || existingItem?.pageCount
);
const fileExtension = inferFileExtension(originalFileName, rawLocation.key);
if (pageCount > 1) {
return formatResponse(
422,
false,
null,
"Multi-page documents require the asynchronous Textract workflow.",
"TEXTRACT_VALIDATION",
"MULTI_PAGE_REQUIRES_ASYNC_TEXTRACT"
);
}
try {
await queueDocument({
action,
documentId,
documentType,
fileExtension,
existingItem,
key,
mimeType,
now,
originalFileName,
pageCount,
rawS3Key: rawLocation.key,
requestId,
userId,
});
} catch (error) {
if (error?.name === "ConditionalCheckFailedException") {
return formatResponse(
409,
false,
null,
"Document status changed before processing could start. Refresh and try again.",
"DYNAMODB",
"DOCUMENT_STATE_CONFLICT"
);
}
logError("Failed to queue document metadata.", error, {
documentId,
userId,
action,
});
return formatResponse(
500,
false,
null,
"Could not queue document for processing.",
"DYNAMODB",
"DYNAMODB_UPDATE_FAILED"
);
}
const executionName = sanitizeExecutionName(
`docuflow-dev-${action.toLowerCase()}-${documentId}-${requestId}`
);
const workflowInput = {
documentId,
userId,
documentType,
bucket: RAW_BUCKET,
key: rawLocation.key,
rawS3Bucket: RAW_BUCKET,
rawS3Key: rawLocation.key,
s3RawPath: `s3://${RAW_BUCKET}/${rawLocation.key}`,
fileName: originalFileName,
originalFileName,
fileExtension,
pageCount,
fileSizeBytes: rawObject?.ContentLength ?? null,
contentType: mimeType,
mimeType,
source: action === "RETRY" ? "API_RETRY" : "API_PROCESS",
requestId,
requestedAt: now,
};
let executionArn;
try {
const result = await sfnClient.send(
new StartExecutionCommand({
stateMachineArn: STATE_MACHINE_ARN,
name: executionName,
input: JSON.stringify(workflowInput),
})
);
executionArn = result.executionArn;
} catch (error) {
const failure = classifyWorkflowStartFailure(error);
await markWorkflowStartFailed({
documentId,
error,
key,
requestId,
userId,
});
logError("Failed to start workflow execution.", error, {
documentId,
userId,
action,
requestId,
});
return formatResponse(
failure.statusCode,
false,
null,
failure.message,
failure.errorStage,
failure.errorCode
);
}
try {
await docClient.send(
new UpdateCommand({
TableName: TABLE_NAME,
Key: key,
UpdateExpression:
"SET workflowExecutionArn = :executionArn, workflowExecutionName = :executionName, updatedAt = :updatedAt",
ConditionExpression: "processingRequestId = :requestId",
ExpressionAttributeValues: {
":executionArn": executionArn,
":executionName": executionName,
":requestId": requestId,
":updatedAt": new Date().toISOString(),
},
})
);
} catch (error) {
// The workflow is already running; do not return a false failure to the client.
logError("Workflow started but execution metadata could not be persisted.", error, {
documentId,
userId,
executionArn,
requestId,
});
}
logInfo("Workflow execution started.", {
documentId,
userId,
rawS3Key: rawLocation.key,
executionArn,
action,
requestId,
});
if (action === "RETRY") {
return formatResponse(200, true, {
documentId,
retryStarted: true,
status: "QUEUED",
updatedAt: now,
executionId: executionArn,
});
}
return formatResponse(200, true, {
documentId,
started: true,
status: "QUEUED",
message: "Processing started.",
executionId: executionArn,
});
}
async function queueDocument({
action,
documentId,
documentType,
fileExtension,
existingItem,
key,
mimeType,
now,
originalFileName,
pageCount,
rawS3Key,
requestId,
userId,
}) {
const expressionAttributeValues = {
":schemaVersion": "1.0.0",
":documentId": documentId,
":userId": userId,
":documentType": documentType,
":fileExtension": fileExtension,
":originalFileName": originalFileName,
":pageCount": pageCount,
":mimeType": mimeType,
":status": "QUEUED",
":rawS3Key": rawS3Key,
":createdAt": now,
":updatedAt": now,
":gsi1pk": "STATUS#QUEUED",
":gsi1sk": now,
":requestId": requestId,
":source": action === "RETRY" ? "API_RETRY" : "API_PROCESS",
};
let conditionExpression;
if (!existingItem) {
conditionExpression = "attribute_not_exists(PK) AND attribute_not_exists(SK)";
} else if (action === "RETRY") {
conditionExpression = "#status IN (:failed, :reviewRequired)";
expressionAttributeValues[":failed"] = "FAILED";
expressionAttributeValues[":reviewRequired"] = "REVIEW_REQUIRED";
} else {
conditionExpression =
"attribute_not_exists(#status) OR #status = :uploaded";
expressionAttributeValues[":uploaded"] = "UPLOADED";
}
await docClient.send(
new UpdateCommand({
TableName: TABLE_NAME,
Key: key,
UpdateExpression:
"SET schemaVersion = :schemaVersion, documentId = :documentId, userId = :userId, documentType = :documentType, originalFileName = :originalFileName, fileExtension = :fileExtension, pageCount = :pageCount, mimeType = :mimeType, contentType = :mimeType, #status = :status, rawS3Key = :rawS3Key, updatedAt = :updatedAt, createdAt = if_not_exists(createdAt, :createdAt), GSI1PK = :gsi1pk, GSI1SK = :gsi1sk, processingRequestId = :requestId, processingSource = :source REMOVE errorMessage",
ConditionExpression: conditionExpression,
ExpressionAttributeNames: {
"#status": "status",
},
ExpressionAttributeValues: expressionAttributeValues,
})
);
}
async function markWorkflowStartFailed({
documentId,
error,
key,
requestId,
userId,
}) {
const now = new Date().toISOString();
try {
await docClient.send(
new UpdateCommand({
TableName: TABLE_NAME,
Key: key,
UpdateExpression:
"SET #status = :failed, errorMessage = :errorMessage, updatedAt = :updatedAt, GSI1PK = :gsi1pk, GSI1SK = :gsi1sk",
ConditionExpression: "processingRequestId = :requestId",
ExpressionAttributeNames: {
"#status": "status",
},
ExpressionAttributeValues: {
":failed": "FAILED",
":errorMessage":
error?.message || "Step Functions execution could not be started.",
":updatedAt": now,
":gsi1pk": "STATUS#FAILED",
":gsi1sk": now,
":requestId": requestId,
},
})
);
} catch (rollbackError) {
logError("Failed to mark workflow start failure in DynamoDB.", rollbackError, {
documentId,
userId,
requestId,
});
}
}
function resolveRawLocation({ requestRawS3Key, storedRawS3Key }) {
const requestLocation = parseS3Location(requestRawS3Key);
const storedLocation = parseS3Location(storedRawS3Key);
if (requestLocation.bucket && requestLocation.bucket !== RAW_BUCKET) {
return {
statusCode: 403,
error: "The requested raw object belongs to a different S3 bucket.",
errorCode: "RAW_BUCKET_FORBIDDEN",
};
}
if (storedLocation.bucket && storedLocation.bucket !== RAW_BUCKET) {
return {
statusCode: 500,
error: "Stored raw object points to an unexpected S3 bucket.",
errorCode: "INVALID_STORED_RAW_LOCATION",
};
}
if (
requestLocation.key &&
storedLocation.key &&
requestLocation.key !== storedLocation.key
) {
return {
statusCode: 409,
error: "rawS3Key does not match the document metadata.",
errorCode: "RAW_KEY_MISMATCH",
};
}
const key = storedLocation.key || requestLocation.key;
if (!key) {
return {
statusCode: 400,
error: "rawS3Key is required.",
errorCode: "INVALID_INPUT",
};
}
return { bucket: RAW_BUCKET, key };
}
function parseS3Location(value) {
if (!value || typeof value !== "string") {
return { bucket: null, key: null };
}
const trimmed = value.trim();
if (!trimmed) return { bucket: null, key: null };
if (!trimmed.startsWith("s3://")) {
return { bucket: null, key: normalizeS3Key(trimmed) };
}
const location = trimmed.slice(5);
const separatorIndex = location.indexOf("/");
if (separatorIndex < 1) {
return { bucket: location || null, key: null };
}
return {
bucket: location.slice(0, separatorIndex),
key: normalizeS3Key(location.slice(separatorIndex + 1)),
};
}
function normalizeS3Key(value) {
return String(value || "").replace(/^\/+/, "");
}
function parseBody(event) {
if (!event?.body) return {};
let value = event.body;
if (event.isBase64Encoded) {
value = Buffer.from(value, "base64").toString("utf8");
}
if (typeof value === "object" && value !== null) return value;
try {
const parsed = JSON.parse(value);
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
? parsed
: null;
} catch {
return null;
}
}
function getUserId(event) {
const claims =
event?.requestContext?.authorizer?.jwt?.claims ||
event?.requestContext?.authorizer?.claims ||
{};
return claims.sub || null;
}
function getHttpMethod(event) {
return event?.requestContext?.http?.method || event?.httpMethod || "";
}
function getRouteAction(event) {
const route = String(
event?.routeKey ||
event?.resource ||
event?.rawPath ||
event?.path ||
""
).toLowerCase();
if (route.includes("/retry")) return "RETRY";
if (route.includes("/process")) return "PROCESS";
return null;
}
function inferMimeType(rawS3Key) {
const lower = rawS3Key.toLowerCase();
if (lower.endsWith(".pdf")) return "application/pdf";
if (lower.endsWith(".jpg") || lower.endsWith(".jpeg")) {
return "image/jpeg";
}
if (lower.endsWith(".png")) return "image/png";
return "application/octet-stream";
}
function inferOriginalFileName(rawS3Key) {
return rawS3Key.split("/").pop() || "original.pdf";
}
function normalizeDocumentType(value) {
const type = String(value || "").toUpperCase();
return type === "INVOICE" || type === "RECEIPT" ? type : "UNKNOWN";
}
function normalizePageCount(value) {
const numeric = Number(value);
return Number.isInteger(numeric) && numeric > 0 ? numeric : 1;
}
function decodeMetadataValue(value) {
if (!value) return null;
try {
return decodeURIComponent(value);
} catch {
return value;
}
}
function inferFileExtension(originalFileName, rawS3Key) {
const source = String(originalFileName || rawS3Key || "").toLowerCase();
const match = source.match(/\.([a-z0-9]+)$/);
return match?.[1] || "";
}
function sanitizeExecutionName(value) {
return value.replace(/[^a-zA-Z0-9-_]/g, "-").slice(0, 80);
}
function validateEnvironment() {
const missing = [];
if (!TABLE_NAME) missing.push("DOCUFLOW_DEV_TABLE_NAME");
if (!RAW_BUCKET) missing.push("DOCUFLOW_DEV_RAW_BUCKET");
if (!STATE_MACHINE_ARN) missing.push("DOCUFLOW_DEV_STATE_MACHINE_ARN or STATE_MACHINE_ARN");
if (!missing.length) return null;
return formatResponse(
500,
false,
null,
`Missing environment variables: ${missing.join(", ")}.`,
"CONFIGURATION",
"MISSING_ENVIRONMENT_VARIABLE"
);
}
function classifyWorkflowStartFailure(error) {
const errorName = error?.name || "Unknown";
const errorMessage = error?.message || "Could not start document workflow.";
if (errorName === "AccessDeniedException") {
return {
statusCode: 403,
errorCode: "WORKFLOW_ACCESS_DENIED",
errorStage: "WORKFLOW",
message:
"Process-control Lambda is not allowed to start the Step Functions state machine.",
};
}
if (
errorName === "StateMachineDoesNotExist" ||
errorName === "ValidationException" ||
errorMessage.includes("State Machine Does Not Exist")
) {
return {
statusCode: 500,
errorCode: "INVALID_STATE_MACHINE_ARN",
errorStage: "CONFIGURATION",
message:
"Configured Step Functions state machine ARN is invalid or does not exist.",
};
}
if (errorName === "ExecutionLimitExceeded") {
return {
statusCode: 429,
errorCode: "WORKFLOW_EXECUTION_LIMIT_EXCEEDED",
errorStage: "WORKFLOW",
message: "Step Functions execution limit exceeded. Try again later.",
};
}
return {
statusCode: 502,
errorCode: "WORKFLOW_START_FAILED",
errorStage: "WORKFLOW",
message: errorMessage,
};
}
function formatResponse(
statusCode,
success,
data,
errorMessage = null,
errorStage = null,
errorCode = "UNKNOWN_ERROR"
) {
const body = { success, data, error: null };
if (errorMessage) {
body.error = { errorCode, errorMessage, errorStage };
}
return {
statusCode,
headers: {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "Authorization,Content-Type",
"Access-Control-Allow-Methods": "OPTIONS,POST",
"Content-Type": "application/json",
},
body: JSON.stringify(body),
};
}
function logInfo(message, details = {}) {
console.log(
JSON.stringify({
level: "INFO",
service: "docuflow-dev-data-process-control-lambda",
message,
...details,
})
);
}
function logError(message, error, details = {}) {
console.error(
JSON.stringify({
level: "ERROR",
service: "docuflow-dev-data-process-control-lambda",
message,
errorName: error?.name,
errorMessage: error?.message,
...details,
})
);
}
COMMON CONFIGURATION: ENVIRONMENT VARIABLES & TIMEOUT
Our code needs to know what the Table name and Bucket name are to connect. In addition, reading files sometimes takes time, so we need to increase the default wait time so the function doesn’t get interrupted.
Perform the following steps for the three core Data Lambdas and any extended Lambdas that you choose to deploy:
- At the details interface of each created Lambda function, switch to the Configuration tab.
- Select the General configuration section in the left vertical menu, click the Edit button.
- Change the Timeout from 3 sec (Default) to 10 seconds (or 15 seconds) to ensure the function does not get interrupted midway when querying large data or reading files from S3.
Click Save.
- Select the Environment variables section in the left vertical menu:
- Click the Edit button, then select Add environment variable.
- First variable pair:
- Key:
DOCUFLOW_DEV_TABLE_NAME - Value:
docuflow-dev-documents-table
- Key:
- Second variable pair:
- Key:
DOCUFLOW_DEV_PROCESSED_BUCKET - Value: Enter exactly the Processed S3 bucket name (e.g.,
docuflow-dev-processed-<AWS_ACCOUNT_ID>-ap-southeast-1)
- Key:
- Third variable pair:
- Key:
DOCUFLOW_DEV_RAW_BUCKET - Value: Enter exactly the Raw S3 bucket name (e.g.,
docuflow-dev-raw-<AWS_ACCOUNT_ID>-ap-southeast-1)
- Key:
- For Process Control, also add
DOCUFLOW_DEV_STATE_MACHINE_ARNwith the deployed processing State Machine ARN. - For Dashboard, optionally add
DOCUFLOW_DEV_VND_EXCHANGE_RATESas a JSON object when report totals must be normalized to VND.
- Click Save to apply.
After completing this configuration for the selected functions, the Lambdas are ready to integrate with API Gateway and Step Functions.