How to Sync pgvector Embeddings in Real-Time using PostgreSQL Triggers and Node.js Background Workers

Introduction

If you've shipped a Retrieval-Augmented Generation (RAG) system to production, you've probably already run into a problem that doesn't show up in your demo: your vector index goes stale.

Someone edits a support article. A product description changes. A user updates their profile bio. Your application dutifully writes the new text to Postgres — and your pgvector embedding column keeps happily returning search results based on the old text, sometimes for minutes, sometimes for hours, depending on how often your cron job runs.

For a blog CMS, that's an inconvenience. For an AI agent making decisions, answering customer questions, or recommending actions based on retrieved context, that's a silent correctness bug. The agent isn't hallucinating — it's confidently retrieving accurate embeddings of outdated information.

The industry default fix is a scheduled job: every 5, 15, or 60 minutes, scan for rows with a stale updated_at, regenerate their embeddings, and write them back. It works, until it doesn't. Cron-based re-embedding pipelines have three structural weaknesses:

• Staleness windows — even a 5-minute cron job means your search index is wrong for up to 5 minutes after every edit, and in practice jobs often run longer than that under load.
• Wasted compute — most cron scans re-check every "possibly stale" row even when nothing changed, or require expensive updated_at diffing logic that's easy to get wrong.
• Missed deletes — a DELETE on your source table rarely has a corresponding cleanup step in a polling job, leaving orphaned embeddings in your vector index that your retriever can still surface.

This guide replaces polling with an event-driven synchronization pipeline: PostgreSQL triggers capture row-level changes the instant they happen, an outbox table durably queues the work, and an isolated Node.js background worker processes that queue asynchronously to regenerate embeddings — all without holding a single extra lock on your hot path.

By the end of this deep-dive, you'll have a production-ready blueprint you can drop into any Node.js + PostgreSQL + pgvector stack.

Why Polling Fails for Vector Search Specifically

It's worth being precise about why this problem is worse for vector search than for typical cache invalidation.

In a traditional cache-invalidation setup, staleness usually means "the user sees an old value for a few seconds." In a RAG pipeline, staleness means the retrieval step itself is corrupted. The LLM downstream has no way of knowing the context it was handed is outdated — it will reason over it as ground truth. This is especially dangerous for:

• Compliance and legal content — an outdated policy embedding retrieved after a policy change.
• Pricing and inventory — a customer-facing agent quoting a stale price.
• Support knowledge bases — an agent citing a deprecated troubleshooting step.

The fix isn't "poll faster." Polling more frequently just trades staleness for database load, and you eventually hit a wall where your scan query itself becomes a performance problem on large tables. The fix is to stop polling entirely and switch to push-based change capture.

Architecture Overview

Here's the full picture before we write any code:

1. Source table — your normal table (e.g. articles) with a pgvector column (embedding vector(1536)).
2. PostgreSQL trigger — fires on INSERT, UPDATE, and DELETE, and inserts a lightweight row into an embedding_jobs outbox table describing what changed.
3. pg_notify — the same trigger fires a NOTIFY on a channel, acting as a real-time doorbell.
4. Node.js worker — a separate, isolated process that:
   • LISTENs on that channel for instant wake-ups.
   • Falls back to short-interval polling of the outbox table as a safety net (in case a notification is missed while the worker is restarting).
   • Claims pending jobs using FOR UPDATE SKIP LOCKED so multiple worker instances can run concurrently without double-processing.
   • Calls your embedding provider (OpenAI, Cohere, a self-hosted model, etc.).
   • Writes the new vector back to the source table.
   • Marks the job complete or retries/dead-letters it on failure.

flowchart LR
    A[App writes to articles table] --> B[PostgreSQL Trigger]
    B --> C[(embedding_jobs outbox table)]
    B --> D[pg_notify channel]
    D --> E[Node.js Worker: LISTEN]
    C --> E
    E --> F[Embedding Provider API]
    F --> E
    E --> G[(UPDATE articles.embedding)]

Notice what's not in this diagram: no cron scheduler, no synchronous HTTP call inside a database transaction, and no shared connection pool between your API and the worker. That separation is the entire point.

Step 1: The Source Table

Assume a typical articles table used to power a documentation search feature:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE articles (
    id BIGSERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    body TEXT NOT NULL,
    embedding vector(1536),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX ON articles USING hnsw (embedding vector_cosine_ops);

Nothing exotic here — this is a standard pgvector setup with an HNSW index for approximate nearest-neighbor search.

Step 2: The Outbox Table

The outbox pattern is the backbone of this architecture. Instead of trying to make the trigger "do the work," the trigger's only job is to cheaply record that work needs to happen.

CREATE TABLE embedding_jobs (
    id BIGSERIAL PRIMARY KEY,
    source_table TEXT NOT NULL,
    source_id BIGINT NOT NULL,
    operation TEXT NOT NULL CHECK (operation IN ('UPSERT', 'DELETE')),
    payload JSONB,
    status TEXT NOT NULL DEFAULT 'pending'
        CHECK (status IN ('pending', 'processing', 'done', 'failed')),
    attempts INT NOT NULL DEFAULT 0,
    last_error TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    processed_at TIMESTAMPTZ
);

-- Speeds up the worker's claim query dramatically on a busy table
CREATE INDEX idx_embedding_jobs_pending
    ON embedding_jobs (created_at)
    WHERE status = 'pending';

Keeping this as its own table (rather than a status column bolted onto articles) matters for two reasons: it keeps write amplification on your hot table to a minimum, and it gives you a clean audit trail of every embedding regeneration attempt, which is invaluable when debugging "why does this search result look wrong."

Step 3: The Trigger Function

This is where the real-time capture happens. The trigger fires on every row-level change and writes a job — nothing more.

CREATE OR REPLACE FUNCTION fn_enqueue_embedding_job()
RETURNS TRIGGER AS $$
DECLARE
    v_operation TEXT;
    v_source_id BIGINT;
    v_payload JSONB;
BEGIN
    IF TG_OP = 'DELETE' THEN
        v_operation := 'DELETE';
        v_source_id := OLD.id;
        v_payload := jsonb_build_object('id', OLD.id);
    ELSE
        -- Skip re-embedding if only unrelated columns changed
        IF TG_OP = 'UPDATE' AND OLD.title = NEW.title AND OLD.body = NEW.body THEN
            RETURN NEW;
        END IF;

        v_operation := 'UPSERT';
        v_source_id := NEW.id;
        v_payload := jsonb_build_object(
            'id', NEW.id,
            'title', NEW.title,
            'body', NEW.body
        );
    END IF;

    INSERT INTO embedding_jobs (source_table, source_id, operation, payload)
    VALUES ('articles', v_source_id, v_operation, v_payload);

    PERFORM pg_notify('embedding_jobs_channel', v_source_id::TEXT);

    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_articles_embedding_sync
AFTER INSERT OR UPDATE OR DELETE ON articles
FOR EACH ROW
EXECUTE FUNCTION fn_enqueue_embedding_job();

A few deliberate design choices worth calling out:

• AFTER trigger, not BEFORE — we want the write to articles to succeed first; the embedding job is a side effect, not a precondition.
• Column diffing on UPDATE — if someone updates a view_count or last_accessed_at column, we don't want to burn embedding API credits re-processing unchanged text. Only fire when semantically relevant columns actually changed.
• pg_notify payload is just the ID — NOTIFY payloads are capped at 8000 bytes and are not durable, so never rely on the payload for the actual data. Use it purely as a signal to "go check the outbox table."

Step 4: The Node.js Background Worker

This worker runs as a completely separate process from your API server — a different container, a different PM2 process, a different Kubernetes deployment, whatever fits your infrastructure. The isolation is the whole point: if embedding generation is slow or your provider is rate-limiting you, your CRUD API never feels it.

Dependency Setup

npm install pg p-limit dotenv

Dedicated Connection Pool

// db/workerPool.js
import pg from "pg";

// A separate, size-capped pool exclusively for the worker.
// Never share this with your API's request-handling pool.
export const workerPool = new pg.Pool({
  connectionString: process.env.DATABASE_URL,
  max: 4, // deliberately small — this workload is not connection-hungry
  idleTimeoutMillis: 30_000,
});

The LISTEN Client

LISTEN/NOTIFY requires a single, dedicated, long-lived client connection — it cannot be pulled from a pool that recycles connections.

// worker/listener.js
import pg from "pg";

export function startListener(onNotify) {
  const client = new pg.Client({ connectionString: process.env.DATABASE_URL });

  client.connect();
  client.query("LISTEN embedding_jobs_channel");

  client.on("notification", (msg) => {
    onNotify(msg.payload);
  });

  client.on("error", (err) => {
    console.error("[listener] connection lost, reconnecting in 3s", err);
    setTimeout(() => startListener(onNotify), 3000);
  });

  return client;
}

Claiming Jobs Safely with SKIP LOCKED

This is the piece that lets you run multiple worker replicas concurrently without two workers ever grabbing the same job.

// worker/claimJobs.js
import { workerPool } from "../db/workerPool.js";

export async function claimPendingJobs(batchSize = 20) {
  const { rows } = await workerPool.query(
    `
    UPDATE embedding_jobs
    SET status = 'processing', attempts = attempts + 1
    WHERE id IN (
      SELECT id FROM embedding_jobs
      WHERE status = 'pending'
      ORDER BY created_at ASC
      LIMIT $1
      FOR UPDATE SKIP LOCKED
    )
    RETURNING *;
    `,
    [batchSize]
  );

  return rows;
}

FOR UPDATE SKIP LOCKED is the key primitive here: if another worker process has already locked a row while claiming it, this query simply skips it instead of blocking. That's what makes horizontal scaling of your worker fleet safe by default.

Processing a Batch

// worker/processBatch.js
import pLimit from "p-limit";
import { workerPool } from "../db/workerPool.js";
import { generateEmbeddings } from "../embeddings/provider.js";

const limit = pLimit(5); // bounded concurrency toward the embedding API

export async function processBatch(jobs) {
  const upserts = jobs.filter((j) => j.operation === "UPSERT");
  const deletes = jobs.filter((j) => j.operation === "DELETE");

  await Promise.all(deletes.map((job) => limit(() => handleDelete(job))));

  if (upserts.length > 0) {
    // Batch call: most providers support arrays of inputs per request
    const texts = upserts.map((j) => `${j.payload.title}\n\n${j.payload.body}`);
    const vectors = await generateEmbeddings(texts);

    await Promise.all(
      upserts.map((job, i) => limit(() => handleUpsert(job, vectors[i])))
    );
  }
}

async function handleUpsert(job, vector) {
  const client = await workerPool.connect();
  try {
    await client.query("BEGIN");
    await client.query(
      `UPDATE articles SET embedding = $1 WHERE id = $2`,
      [pgvectorLiteral(vector), job.payload.id]
    );
    await client.query(
      `UPDATE embedding_jobs SET status = 'done', processed_at = now() WHERE id = $1`,
      [job.id]
    );
    await client.query("COMMIT");
  } catch (err) {
    await client.query("ROLLBACK");
    await markFailure(job, err);
  } finally {
    client.release();
  }
}

async function handleDelete(job) {
  const client = await workerPool.connect();
  try {
    await client.query("BEGIN");
    await client.query(`UPDATE articles SET embedding = NULL WHERE id = $1`, [
      job.payload.id,
    ]);
    await client.query(
      `UPDATE embedding_jobs SET status = 'done', processed_at = now() WHERE id = $1`,
      [job.id]
    );
    await client.query("COMMIT");
  } catch (err) {
    await client.query("ROLLBACK");
    await markFailure(job, err);
  } finally {
    client.release();
  }
}

async function markFailure(job, err) {
  const MAX_ATTEMPTS = 5;
  const nextStatus = job.attempts >= MAX_ATTEMPTS ? "failed" : "pending";

  await workerPool.query(
    `UPDATE embedding_jobs SET status = $1, last_error = $2 WHERE id = $3`,
    [nextStatus, String(err.message ?? err), job.id]
  );
}

function pgvectorLiteral(vector) {
  return `[${vector.join(",")}]`;
}

Notice the pattern: every mutation to articles and embedding_jobs happens inside its own transaction, scoped tightly, and released back to the pool immediately after. No connection is held open while waiting on the embedding API — that call happens before we touch the database at all.

Tying It Together: The Worker Loop

// worker/index.js
import "dotenv/config";
import { startListener } from "./listener.js";
import { claimPendingJobs } from "./claimJobs.js";
import { processBatch } from "./processBatch.js";

let isProcessing = false;

async function drainQueue() {
  if (isProcessing) return; // avoid overlapping runs
  isProcessing = true;

  try {
    let jobs = await claimPendingJobs(20);
    while (jobs.length > 0) {
      await processBatch(jobs);
      jobs = await claimPendingJobs(20);
    }
  } catch (err) {
    console.error("[worker] drain error", err);
  } finally {
    isProcessing = false;
  }
}

// Real-time trigger via NOTIFY
startListener(() => drainQueue());

// Safety-net poll in case a NOTIFY is ever missed (e.g. during a restart)
setInterval(drainQueue, 5000);

// Catch anything left over from before this process started
drainQueue();

console.log("[worker] listening for embedding_jobs_channel notifications...");

This gives you the best of both worlds: millisecond-scale reactivity via LISTEN/NOTIFY under normal operation, and a 5-second polling safety net that guarantees eventual consistency even if a notification is dropped during a worker restart or network blip.

Real-World Example: A Documentation Search Feature

Picture a SaaS product with an internal knowledge base powering an AI support agent. Support engineers edit articles constantly throughout the day. Before this pipeline, the team ran a 10-minute cron job to refresh embeddings — and their support agent was regularly citing outdated troubleshooting steps that had been corrected minutes earlier.

After switching to this architecture:

• An article edit triggers fn_enqueue_embedding_job in under a millisecond.
• The worker receives the NOTIFY almost instantly and claims the job.
• The new embedding is written back to articles.embedding typically within 1–3 seconds (bounded mostly by embedding API latency, not database overhead).
• The AI agent's next retrieval, seconds later, reflects the corrected content.

That's a reduction from a 10-minute worst-case staleness window to a low-single-digit-second one, with less total database load than the old polling scan.

🚀 Pro Tips

• Batch your embedding API calls. Most providers (OpenAI, Cohere, Voyage) accept arrays of input strings in a single request. Claiming jobs in batches of 20–50 and sending them as one API call dramatically cuts latency and cost compared to one call per row.
• Debounce rapid successive edits. If a user is actively editing a document and saving every few seconds, consider a short debounce window in the trigger (e.g. only enqueue if no job was created for this source_id in the last N seconds) to avoid wasted embedding calls on transient intermediate states.
• Use a generated column hash to detect real content changes. Instead of comparing every relevant column manually in the trigger, maintain a content_hash column and compare hashes — it's cheaper to diff and easier to extend as your schema grows.
• Run your worker with graceful shutdown handling. Listen for SIGTERM, stop claiming new jobs, and let in-flight batches finish before exiting, so a deployment rollout never orphans a processing row.
• Expose queue depth as a metric. SELECT count(*) FROM embedding_jobs WHERE status = 'pending' is a trivial but extremely valuable health metric — alert if it grows unbounded, which usually means your worker is down or your embedding provider is failing.
• Version your embeddings. Store an embedding_model_version column alongside the vector so that when you upgrade embedding models, you can identify and re-backfill old vectors without guesswork.

Best Practices Checklist

• ✅ Keep the trigger function fast and side-effect-free beyond the outbox insert and NOTIFY.
• ✅ Never make network calls (HTTP, embedding APIs) from inside a PostgreSQL trigger or transaction.
• ✅ Use a separate, size-capped connection pool for the worker — never share it with your API.
• ✅ Use FOR UPDATE SKIP LOCKED so you can safely scale to multiple worker replicas.
• ✅ Treat pg_notify as a latency optimization, not a delivery guarantee — always back it with a durable table and a polling fallback.
• ✅ Make job processing idempotent — reprocessing the same job twice should never produce a corrupted state.
• ✅ Implement retry limits and a dead-letter (failed) status so bad rows don't loop forever.
• ✅ Monitor queue depth, processing latency, and failure rate as first-class metrics.

Common Mistakes to Avoid

• Calling the embedding API synchronously inside the trigger. This is the single most common mistake teams make when first trying to "solve" staleness. It ties your write latency to a third-party API's uptime and turns every UPDATE into a potential lock-contention incident.
• Sharing the API's connection pool with the worker. A burst of embedding jobs can exhaust the pool and cause your unrelated CRUD endpoints to start timing out — a classic case of an infrastructure concern leaking into an unrelated system.
• Forgetting DELETE events entirely. Many teams build change capture only for INSERT/UPDATE and leave orphaned vectors behind after deletes, which continue to surface in similarity search results indefinitely.
• No idempotency guarantees. If your worker crashes mid-batch and a job gets reprocessed, make sure regenerating the same embedding twice is harmless — don't, for example, append to an array column without a check.
• Relying purely on LISTEN/NOTIFY with no fallback. If your worker is restarting exactly when a notification fires, that notification is lost forever unless you also poll the outbox table.
• No monitoring on queue depth. Teams often only discover their worker has been down for six hours when a customer complains that search results are wrong — a simple queue-depth alert would have caught it in minutes.

📌 Key Takeaways

• Cron-based re-embedding is fundamentally reactive and slow; it treats vector staleness as tolerable when it's often a correctness bug for downstream AI agents.
• PostgreSQL triggers paired with an outbox table give you durable, ordered change capture with almost no overhead on the source table.
• pg_notify provides millisecond-scale reactivity but must always be backed by a durable table and polling fallback since notifications aren't guaranteed delivery.
• Isolating the worker's connection pool and using FOR UPDATE SKIP LOCKED are what let this architecture scale horizontally without ever touching your API's performance.
• Idempotency, retries, and dead-lettering aren't optional extras — they're what separates a demo from something you can trust in production.

Conclusion

Vector index staleness is one of those problems that's easy to ignore until it quietly undermines the thing your RAG system exists to do: retrieve accurate, current information. Cron jobs feel like a reasonable starting point, but they scale poorly and leave staleness windows that are simply too large for anything user-facing or agent-facing.

The pattern in this guide — PostgreSQL triggers, a durable outbox table, pg_notify for low-latency wake-ups, and an isolated Node.js worker using FOR UPDATE SKIP LOCKED — gives you a real-time, horizontally scalable synchronization pipeline that keeps your pgvector embeddings fresh within seconds of a source change, without ever competing with your CRUD API for database resources.

It's more code than a cron job, yes. But it's the difference between a RAG system that's usually right and one you can actually depend on in production.

References

• pgvector GitHub Repository
• PostgreSQL Documentation: Trigger Functions
• PostgreSQL Documentation: NOTIFY
• PostgreSQL Documentation: The Locking Clause (FOR UPDATE / SKIP LOCKED)
• node-postgres (pg) Documentation
• Transactional Outbox Pattern — microservices.io