TODO: deletions need to be explicit

Realtime polling (experimental)

The goal

Make the UI update automatically when the data is changed by other users or third-party services
Use only HTTP protocol (no WebSockets, no SSE) to ensure compatibility with any hosting provider

Introduction

In the previous article, we covered how to use the OpenAPI Realtime API to deliver instant UI updates while the user interacts with the app. The LLM functions created with createLLMTools ran in the browser, where the AI performed HTTP requests to the server using RPC modules.

On this page, we explain how to use database polling to receive updates triggered by other users or third-party services (such as MCP), keeping the UI in sync with the database. To ensure the app can be deployed to any hosting provider, we avoid non-HTTP protocols such as WebSockets and instead use HTTP polling powered by JSONLines.

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. Our polling service reads these events every second and relays them to clients with the app open.

Because we use Prisma as our ORM, we rely on 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. beginEmitting 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 all 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 set 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 previous 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.
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. 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.


const [isPollingEnabled, setIsPollingEnabled] = useState(false);
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 < 5 &&
        (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]);

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:


const { tools } = createLLMTools({
  modules: {
    UserController,
    TaskController,
  },
});

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 polling (experimental)

The goal

Introduction

0

Redis DB as event bus

Polling controller and service

DatabasePollService.ts

DatabasePollController.ts

Client-side logic

Bonus: Telegram bot (unstable)