Automate RAG Index Refresh with EventBridge in Node.js 22

Stale embeddings are like a hidden leak in retrieval-augmented generation (RAG) systems. They can silently corrode the reliability of your AI-driven applications, delivering outdated data until someone finally calls out the discrepancy. With Node.js 22 and AWS EventBridge Scheduler, though, you can tackle this by automating the refreshing of your vector store. This keeps your embeddings current without any manual effort.

Why RAG Staleness Matters

RAG uses a language model to pull relevant text from a vector store for generating responses. These vectors represent numerical impressions of text. If the documents evolve and the vector store doesn’t follow suit, you’re likely to dish out obsolete answers—risking both user trust and compliance.

Consider stale embeddings like old books in a library that are never swapped out. Users repeatedly cite outdated facts because the collection isn’t refreshed.

A Simple Example

type Vector = number[];
type Doc = { id: string; text: string; embedding: Vector };
let store: Doc[] = [
  {
    id: "1",
    text: "The API returns 200 OK on success.",
    embedding: [0.12, 0.34, 0.56] // generated months ago
  }
];

// A user asks about the new status code (it changed to 201)
const queryEmbedding = [0.12, 0.34, 0.56]; // same as old doc
const nearest = store.reduce((best, doc) => {
  // naive cosine similarity placeholder
  const sim = doc.embedding.reduce((s, v, i) => s + v * queryEmbedding[i], 0);
  return sim > best.sim ? { doc, sim } : best;
}, { doc: null as Doc | null, sim: -Infinity });

console.log("Returned doc:", nearest.doc?.text);
// → "The API returns 200 OK on success."

Even after the API update to 201, the system sticks to the old 200 response because the embedding wasn’t refreshed.

Automating Refresh with EventBridge Scheduler

EventBridge Scheduler can make your life easier by automating the refresh. It’s a managed service that triggers events at designated times. You just set up a schedule, and it pings an HTTPS endpoint to start the refresh.

Setting Up Your Schedule

import {
  SchedulerClient,
  CreateScheduleCommand,
} from "@aws-sdk/client-scheduler";

const client = new SchedulerClient({ region: "us-east-1" });

async function createDailyRefreshSchedule() {
  const cmd = new CreateScheduleCommand({
    Name: "RagEmbeddingRefresh",
    ScheduleExpression: "rate(24 hours)",
    ScheduleExpressionTimezone: "America/New_York",
    Target: {
      Arn: "arn:aws:scheduler:::aws-sdk:apprunner:CreateService",
      HttpParameters: {
        HeaderParameters: {
          "Content-Type": "application/json",
        },
      },
      Uri: "https://my-rag-refresh.service.aws-region.amazonaws.com/refresh",
    },
    Description: "Refreshes embeddings for the RAG vector store each night",
    FlexibleTimeWindow: { Mode: "OFF" },
  });
  const response = await client.send(cmd);
  console.log("Schedule created:", response);
}

createDailyRefreshSchedule().catch(console.error);

This script sets up a daily job using a rate expression. Make sure your timezone is set correctly to prevent any surprises due to daylight saving changes.

Gotchas and Tips

• Time-zone/DST Edge Cases: Time zone shifts can mess with your schedule timings, so keep an eye on them.
• Rate vs. Cron: Stick with rate expressions for simplicity unless you need pinpoint precision.
• Minimum Resolution: Events can't fire more than once a second with EventBridge Scheduler.

Deploying a Secure Node.js 22 Service

Your EventBridge Scheduler will call an HTTPS endpoint, and AWS App Runner is where you can deploy this service. It’ll handle all the embedding refresh work.

Example Express Application

import express, { Request, Response } from "express";

const app = express();
app.use(express.json());

app.get("/health", (_req: Request, res: Response) => {
  res.json({ status: "ok", timestamp: new Date().toISOString() });
});

app.post("/refresh", async (_req: Request, res: Response) => {
  try {
    const { refreshEmbeddings } = await import("./refresh");
    await refreshEmbeddings();
    res.json({ result: "success", refreshedAt: new Date().toISOString() });
  } catch (err) {
    console.error("Refresh failed:", err);
    res.status(500).json({ result: "error", message: (err as Error).message });
  }
});

export default app;

if (require.main === module) {
  const PORT = process.env.PORT ? parseInt(process.env.PORT) : 8080;
  app.listen(PORT, () => {
    console.log(`RAG refresh service listening on http://0.0.0.0:${PORT}`);
  });
}

This straightforward Express app relies on the --experimental-strip-types flag, which is great since it lets you run TypeScript directly without bothering with a build step.

Embedding Refresh Logic

Your refresh logic needs to grab the latest documents, create new embeddings with an Amazon Bedrock model, and then dump them into DynamoDB in batches.

import {
  DynamoDBDocumentClient,
  BatchWriteCommand,
} from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { BedrockRuntimeClient, InvokeModelCommand } from "@aws-sdk/client-bedrock-runtime";
import fetch from "node-fetch";

const ddbClient = new DynamoDBClient({});
const ddbDocClient = DynamoDBDocumentClient.from(ddbClient);
const bedrockClient = new BedrockRuntimeClient({});

async function getEmbedding(text: string): Promise {
  const payload = JSON.stringify({ inputText: text });
  const command = new InvokeModelCommand({
    ModelId: "amazon.titan-embed-text-v1",
    ContentType: "application/json",
    Accept: "application/json",
    Body: Buffer.from(payload),
  });
  const response = await bedrockClient.send(command);
  const bodyString = Buffer.from(response.Body as Uint8Array).toString("utf-8");
  const parsed = JSON.parse(bodyString);
  return parsed.embedding as number[];
}

async function loadSourceDocuments(): Promise> {
  const resp = await fetch(SOURCE_DOCS_URL);
  if (!resp.ok) {
    throw new Error(`Failed to fetch source docs: ${resp.statusText}`);
  }
  return (await resp.json()) as Array<{ id: string; text: string };
}

export async function refreshEmbeddings(): Promise {
  console.log("Starting embedding refresh…");
  const docs = await loadSourceDocuments();
  const BATCH_SIZE = 25;
  for (let i = 0; i < docs.length; i += BATCH_SIZE) {
    const batch = docs.slice(i, i + BATCH_SIZE);
    const writeRequests = await Promise.all(
      batch.map(async (doc) => {
        const embedding = await getEmbedding(doc.text);
        return {
          PutRequest: {
            Item: {
              pk: `DOC#${doc.id}`,
              sk: "EMBEDDING",
              text: doc.text,
              embedding,
              refreshedAt: new Date().toISOString(),
              ttl: Math.floor(Date.now() / 1000) + 60 * 60 * 24 * 30,
            },
          },
        };
      })
    );
    const batchInput: BatchWriteCommandInput = {
      RequestItems: {
        [DDB_TABLE]: writeRequests,
      },
    };
    let attempts = 0;
    let unprocessed = writeRequests;
    while (unprocessed.length && attempts < 3) {
      const cmd = new BatchWriteCommand({
        RequestItems: { [DDB_TABLE]: unprocessed },
      });
      const result = await ddbDocClient.send(cmd);
      unprocessed = result.UnprocessedItems?.[DDB_TABLE] ?? [];
      if (unprocessed.length) {
        attempts++;
        console.warn(`Retry ${attempts}: ${unprocessed.length} items unprocessed`);
        await new Promise((r) => setTimeout(r, 1000 * attempts));
      }
    }
    if (unprocessed.length) {
      console.error("Failed to write some items after retries:", unprocessed);
    } else {
      console.log(`Batch ${i / BATCH_SIZE + 1} written successfully`);
    }
  }
  console.log("Embedding refresh completed");
}

This script pulls in documents, creates embeddings, and dumps them into DynamoDB using batch operations to cut down on round-trip overhead. The retry logic ensures resilience, and a batch size of 25 aligns nicely with DynamoDB's write capacity units to handle throughput efficiently.

Potential Limitations

While automating RAG index refreshes with EventBridge Scheduler and Node.js 22 is efficient, it's not without its limits. If your document collection balloons, you might need to shard partition keys for better distribution across DynamoDB partitions. Also, the --experimental-strip-types flag, though useful, is still experimental and might not be ideal for all production setups.

For more on similar topics, check out articles like Building a Node.js Chatbot that Uses Real-Time Weather Data or Automated Testing for Node.js APIs with Vitest and Supertest.