Realtime Database Polling

In the previous articles, we’ve covered how to make UI automatically, when the front-end receives updated database entities while the user interacts with the app but also when the app is interacted by text or voice AI interfaces. In this article, we explain how to keep the UI in sync with database changes triggered by other users or third-party services (such as MCP or Telegram bot described at this article), using only HTTP protocol (no WebSockets) to ensure compatibility with any hosting provider.

The way we achieve this is by implementing database polling powered by JSONLines. The server sends updates to clients whenever the database changes, and the client reconnects automatically when the connection is closed.

The component below demonstrates a simple polling example that receives incremental updates from the server every second. After 10 updates, the server closes the connection, and the client reconnects automatically. We can use the same approach to receive database updates in real time, by having the server send updates whenever the database changes.

Poll Ticker

0

View on examples.vovk.dev »

A small delay (up to a half of a second) is expected due to the CORS preflight. See the Network tab in dev tools for details.

Redis DB as event bus

While we could poll the main Postgres database for changes, that approach is inefficient. Instead, we use Redis as an event bus: whenever the main database changes, we write a small event to Redis Database. Our polling service reads these events every second and sends them to clients with the app open.

Because we use Prisma as our ORM, we can use Prisma Extensions to hook into database operations and write events to Redis. This is where the DatabaseService mentioned in the previous article comes into play.

src/modules/database/DatabaseService.ts


import { PrismaClient } from "../../../prisma/generated/client";
import { PrismaNeon } from "@prisma/adapter-neon";
import DatabaseEventsService, { type DBChange } from "./DatabaseEventsService";
import type { BaseEntity } from "@/types";
import "./neon-local"; // Setup Neon for local development
 
export default class DatabaseService {
  static get prisma() {
    return (this.#prisma ??= this.getClient());
  }
  static #prisma: ReturnType<typeof DatabaseService.getClient> | null = null;
 
  private static getClient() {
    const prisma = new PrismaClient({
      adapter: new PrismaNeon({
        connectionString: `${process.env.DATABASE_URL}`,
      }),
    });
 
    DatabaseEventsService.beginEmitting();
 
    return prisma.$extends({
      name: "events",
      query: {
        $allModels: {
          async $allOperations({ model, operation, args, query }) {
            const allowedOperations = [
              "create",
              "update",
              "delete",
              "findMany",
              "findUnique",
              "findFirst",
              "count",
            ] as const;
            type AllowedOperation = (typeof allowedOperations)[number];
            if (!allowedOperations.includes(operation as AllowedOperation)) {
              throw new Error(
                `Unsupported database operation "${operation}" on model "${model}"`,
              );
            }
            const result = (await query(args)) as BaseEntity | BaseEntity[];
 
            const now = new Date().toISOString();
            let change: DBChange | null = null;
 
            const makeChange = (
              entity: BaseEntity,
              type: DBChange["type"],
            ) => ({
              id: entity.id,
              entityType: entity.entityType,
              date:
                type === "delete"
                  ? now
                  : entity.updatedAt
                    ? new Date(entity.updatedAt).toISOString()
                    : now,
              type,
            });
 
            switch (operation as AllowedOperation) {
              case "create":
                if ("entityType" in result)
                  change = makeChange(result, "create");
                break;
 
              case "update":
                if ("entityType" in result)
                  change = makeChange(result, "update");
                break;
 
              case "delete":
                if ("entityType" in result) {
                  change = makeChange(result, "delete");
                  // Automatically add __isDeleted flag to deletion results
                  Object.assign(result, { __isDeleted: true });
                }
                break;
 
              case "findMany":
              case "findUnique":
              case "findFirst":
              case "count":
                // no events
                break;
 
              default:
                console.warn(
                  `Unhandled Prisma operation: ${operation} for model: ${model}`,
                );
                break;
            }
 
            if (change) {
              await DatabaseEventsService.createChanges([change]);
            }
 
            return result;
          },
        },
      },
    });
  }
}

The code above is fetched from GitHub repository.

The getClient method calls DatabaseEventsService.beginEmitting() to start emitting events. The beginEmitting function runs a setInterval that connects to Redis and periodically checks for new events. When a new event is found, it emits it via mitt .
prisma.$extends hooks into some of the Prisma model operations, determines whether an operation modifies data, and if so calls await DatabaseEventsService.createChanges([change]) to persist a change entry in Redis. The change captures creates, updates, and deletions:


export type DBChange = {
  id: string;
  entityType: EntityType;
  date: string;
  type: "create" | "update" | "delete";
};

The date field indicates when the change occurred.

For create and update operations, it uses the updatedAt DB field (important: all write operations must select this field).
For delete operations, it uses the current time.

The delete operation also adds an __isDeleted property. The front end checks this property to hide the deleted entity by setting enumerable: false on the entity registry item (see the Setup article).

Operations like find... and count do not trigger changes and are passed through as-is.

In addition to beginEmitting and createChanges, DatabaseEventsService provides a connect method and an emitter (a mitt instance). These are used by the polling service (DatabasePollService, discussed next) to be notified about new events.

src/modules/database/DatabaseEventsService.ts


import { EntityType } from "../../../prisma/generated/client";
import mitt from "mitt";
import { createClient } from "redis";
 
export type DBChange = {
  id: string;
  entityType: EntityType;
  date: string;
  type: "create" | "update" | "delete";
};
 
export default class DatabaseEventsService {
  public static readonly DB_KEY = "db_updates";
 
  private static readonly INTERVAL = 1_000;
  private static lastTimestamp = Date.now();
 
  private static redisClient = createClient({
    url: process.env.REDIS_URL,
  });
 
  public static emitter = mitt<{
    [DatabaseEventsService.DB_KEY]: DBChange[];
  }>();
 
  // ensure Redis is connected
  private static async connect() {
    if (!this.redisClient.isOpen) {
      await this.redisClient.connect();
      this.redisClient.on("error", (err) => {
        console.error("Redis Client Error", err);
      });
    }
  }
 
  // push one update into our ZSET, with score = timestamp
  public static async createChanges(changes: DBChange[]) {
    if (changes.length === 0) return;
 
    await this.connect();
 
    // build array of { score, value } objects
    const entries = changes.map(({ id, entityType, type, date }) => ({
      score: Date.now(),
      value: JSON.stringify({ id, entityType, date, type }),
    }));
 
    // one multi(): batch ZADD + EXPIRE
    await this.redisClient
      .multi()
      .zAdd(this.DB_KEY, entries)
      .expire(this.DB_KEY, (this.INTERVAL * 60) / 1000)
      .exec();
  }
 
  public static beginEmitting() {
    setInterval(async () => {
      await this.connect();
 
      const now = Date.now();
 
      // get everything with score ∈ (lastTimestamp, now]
      const raw = await this.redisClient.zRangeByScore(
        this.DB_KEY,
        this.lastTimestamp + 1,
        now,
      );
 
      this.lastTimestamp = now;
 
      if (raw.length > 0) {
        const updates = raw.map((s) => JSON.parse(s) as DBChange);
        this.emitter.emit(this.DB_KEY, updates);
      }
    }, this.INTERVAL);
  }
}

The code above is fetched from GitHub repository.

Polling controller and service

With Redis change entries and the change emitter in place, we can implement a polling endpoint that streams updates to clients in real time. The DatabasePollController exposes a single JSONLines endpoint, and DatabasePollService uses a JSONLinesResponse instance (received from the controller) to send data to clients. The service closes the connection safely after 30 seconds, so clients should reconnect.

DatabasePollService.ts

src/modules/database/DatabasePollService.ts


import { JSONLinesResponse, type VovkIteration } from "vovk";
import { forEach, groupBy } from "lodash";
import type DatabasePollController from "./DatabasePollController";
import DatabaseEventsService, { type DBChange } from "./DatabaseEventsService";
import DatabaseService from "./DatabaseService";
 
export default class PollService {
  static poll(
    resp: JSONLinesResponse<VovkIteration<typeof DatabasePollController.poll>>,
  ) {
    setTimeout(resp.close, 30_000);
 
    let asOldAs = new Date();
    // 10 minutes ago; TODO: use latest update date from registry
    asOldAs.setMinutes(asOldAs.getMinutes() - 10);
 
    DatabaseEventsService.emitter.on(
      DatabaseEventsService.DB_KEY,
      (changes) => {
        const deleted = changes.filter((change) => change.type === "delete");
        const createdOrUpdated = changes.filter(
          (change) => change.type === "create" || change.type === "update",
        );
 
        for (const deletedEntity of deleted) {
          resp.send({
            id: deletedEntity.id,
            entityType: deletedEntity.entityType,
            __isDeleted: true,
          });
        }
        // group by entityType and date, so the date is maximum date for the given entity: { entityType: string, date: string }[]
        forEach(groupBy(createdOrUpdated, "entityType"), (changes) => {
          const maxDateItem = changes.reduce(
            (max, change) => {
              const changeDate = new Date(change.date);
              return changeDate.getTime() > new Date(max.date).getTime()
                ? change
                : max;
            },
            { date: new Date(0) } as unknown as DBChange,
          );
 
          if (new Date(maxDateItem.date).getTime() > asOldAs.getTime()) {
            void DatabaseService.prisma[maxDateItem.entityType as "user"]
              .findMany({
                where: {
                  updatedAt: {
                    gt: asOldAs,
                  },
                },
              })
              .then((entities) => {
                for (const entity of entities) {
                  resp.send(entity);
                }
              });
            asOldAs = new Date(maxDateItem.date);
          }
        });
      },
    );
  }
}

The code above is fetched from GitHub repository.

When a delete DB change is emitted via DatabaseEventsService.emitter, the service sends an event with id, entityType, and __isDeleted: true. The front end uses __isDeleted to hide the entity by making it non-enumerable in the registry. That’s why deletions need to be explicit, even if cascade deletions are handled automatically by the database. See UserController.deleteUser method.
When an update or create change is emitted, the service fetches the full entity from Postgres (since Redis stores only metadata) and sends it to clients.

Client-side logic

On the client side (for example, in a React component), call DatabasePollRPC.poll() to receive a stream of database events. As with any JSONLines RPC method, it returns an async iterable that you can consume in a for await loop. Because the server may close the connection or a network error may occur, wrap the logic in a retry loop and avoid reconnecting on AbortError thrown when the user disables polling or navigates away. Since the fetcher is already configured, the loop body can be empty—you do not need to handle data manually.

Front-end code, besides the polling logic, also includes on/off toggle state saved to localStorage, so the user can enable or disable polling as needed.

Here’s the useDatabasePolling hook that implements the described logic, returning the [isPollingEnabled, setIsPollingEnabled] state tuple:

src/hooks/useDatabasePolling.ts


import { useEffect, useRef, useState } from "react";
import { DatabasePollRPC } from "vovk-client";
 
/**
 * Hook to manage database polling state.
 * @example const [isPollingEnabled, setIsPollingEnabled] = useDatabasePolling(false);
 */
export default function useDatabasePolling(initialValue = false) {
  const MAX_RETRIES = 5;
  const [isPollingEnabled, setIsPollingEnabled] = useState(initialValue);
  const pollingAbortControllerRef = useRef<AbortController | null>(null);
 
  useEffect(() => {
    const isEnabled = localStorage.getItem("isPollingEnabled");
    setIsPollingEnabled(isEnabled === "true");
  }, []);
 
  useEffect(() => {
    localStorage.setItem("isPollingEnabled", isPollingEnabled.toString());
    async function poll(retries = 0) {
      if (!isPollingEnabled) {
        pollingAbortControllerRef.current?.abort();
        return;
      }
      try {
        while (true) {
          console.log("START POLLING");
          const iterable = await DatabasePollRPC.poll();
          pollingAbortControllerRef.current = iterable.abortController;
 
          for await (const iteration of iterable) {
            console.log("New DB update:", iteration);
          }
        }
      } catch (error) {
        if (
          retries < MAX_RETRIES &&
          (error as Error & { cause?: Error }).cause?.name !== "AbortError"
        ) {
          console.error("Polling failed, retrying...", error);
          await new Promise((resolve) => setTimeout(resolve, 2000));
          return poll(retries + 1);
        }
      }
    }
 
    void poll();
 
    return () => {
      pollingAbortControllerRef.current?.abort();
    };
  }, [isPollingEnabled]);
 
  return [isPollingEnabled, setIsPollingEnabled] as const;
}

The code above is fetched from GitHub repository.

Usage:


const [isPollingEnabled, setIsPollingEnabled] = useDatabasePolling(false);

Setting up MCP server

Coming soon.

Bonus: Telegram bot (unstable)

As an example of a third-party source of database changes, see the TelegramService and the accompanying TelegramController . It accepts text or voice messages, performs voice-to-text transcription if needed using the OpenAI Whisper API, and uses the same createLLMTools function on the server.

The Telegram API library is implemented with OpenAPI Mixins and used as a TelegramAPI module to call Telegram API methods.

vovk.config.mjs


// @ts-check
/** @type {import('vovk').VovkConfig} */
const config = {
  // ...
  outputConfig: {
    // ...
    segments: {
      telegram: {
        openAPIMixin: {
          source: {
            url: "https://raw.githubusercontent.com/sys-001/telegram-bot-api-versions/refs/heads/main/files/openapi/yaml/v183.yaml",
            fallback: ".openapi-cache/telegram.yaml",
          },
          getModuleName: "TelegramAPI",
          getMethodName: ({ path }) => path.replace(/^\//, ""),
          errorMessageKey: "description",
        },
      },
    },
  },
};
 
export default config;

The TelegramService class handles interaction with the Telegram API and generates AI responses using the Vercel AI SDK.

src/modules/telegram/TelegramService.ts


import OpenAI from "openai";
import { TelegramAPI } from "vovk-client";
 
const openai = new OpenAI();
 
export default class TelegramService {
  static get apiRoot() {
    const TELEGRAM_BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN;
    if (!TELEGRAM_BOT_TOKEN) {
      throw new Error("Missing TELEGRAM_BOT_TOKEN environment variable");
    }
    return `https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}`;
  }
 
  // ...
 
  private static async generateAIResponse(
    chatId: number,
    userMessage: string,
    systemPrompt: string,
  ): Promise<{ botResponse: string; messages: ModelMessage[] }> {
    // Get chat history
    const history = await this.getChatHistory(chatId);
    const messages = [
      ...this.formatHistoryForVercelAI(history),
      { role: "user", content: userMessage } as const,
    ];
    const { tools } = createLLMTools({
      modules: {
        UserController,
        TaskController,
      },
    });
  
    // Generate a response using Vercel AI SDK
    const { text } = await generateText({
      model: vercelOpenAI("gpt-5"),
      system: systemPrompt,
      messages,
      stopWhen: stepCountIs(16),
      tools: {
        ...Object.fromEntries(
          tools.map(({ name, execute, description, parameters }) => [
            name,
            tool({
              execute,
              description,
              inputSchema: jsonSchema(parameters as JSONSchema7),
            }),
          ]),
        ),
      },
    });
 
    const botResponse = text || "I couldn't generate a response.";
 
    // Add user message to history
    await this.addToHistory(chatId, "user", userMessage);
    // Add assistant response to history
    await this.addToHistory(chatId, "assistant", botResponse);
 
    messages.push({
      role: "assistant",
      content: botResponse,
    });
 
    return { botResponse, messages };
  }
 
  private static async sendTextMessage(
    chatId: number,
    text: string,
  ): Promise<void> {
    await TelegramAPI.sendMessage({
      body: {
        chat_id: chatId,
        text: text,
        parse_mode: "html",
      },
      apiRoot: this.apiRoot,
    });
  }
 
  private static async sendVoiceMessage(
    chatId: number,
    text: string,
  ): Promise<void> {
    try {
      // Generate speech from text using OpenAI TTS
      const speechResponse = await openai.audio.speech.create({
        model: "tts-1",
        voice: "alloy",
        input: text,
        response_format: "opus",
      });
 
      // Convert the response to a Buffer
      const voiceBuffer = Buffer.from(await speechResponse.arrayBuffer());
 
      const formData = new FormData();
      formData.append("chat_id", String(chatId));
      formData.append(
        "voice",
        new Blob([voiceBuffer], { type: "audio/ogg" }),
        "voice.ogg",
      );
 
      // Send the voice message
      await TelegramAPI.sendVoice({
        body: formData,
        apiRoot: this.apiRoot,
      });
    } catch (error) {
      console.error("Error generating voice message:", error);
      // Fallback to text message if voice generation fails
      await this.sendTextMessage(chatId, text);
    }
  }
 
  // ...
}

Realtime Database Polling

0

Redis DB as event bus

Polling controller and service

DatabasePollService.ts

DatabasePollController.ts

Client-side logic

Setting up MCP server

Bonus: Telegram bot (unstable)