Realtime

WebSocket push model — automatic data synchronization

Gencow includes built-in realtime via WebSocket. When data changes on the server, all connected clients receive updates automatically — no manual refetching, no polling.

TanStack Query apps: Realtime also works through `GencowTanstackProvider`. See the TanStack Query guide for adapter-specific cache key and invalidation patterns.

How It Works

Gencow uses two realtime paths depending on whether the query requires authentication and whether the query is generated by createCrud().

Public queries (server-side allowAnonymous, mirrored in generated api.ts) can run without a JWT. Public createCrud() list queries subscribe with a public query grant and receive invalidation, so the client refetches over HTTP without auth.

Auth-required queries (the default for createCrud()) also receive invalidation only over WebSocket. When data changes, the server signals subscribed clients to refetch; the client automatically calls POST /api/query with its JWT, and PostgreSQL RLS runs on that read. Query result data is never transmitted over WebSocket for private queries.

Custom public queries can still use ctx.realtime.refresh() or ctx.realtime.emit() when the pushed payload is safe for every subscriber.

Public CRUD query                    Auth-required query
─────────────────                    ───────────────────

useQuery(api.posts.list, {})           useQuery(api.tasks.list, {})
  │                                    │
  ├─ HTTP: initial data                ├─ HTTP: initial data
  ├─ WS: subscribe (public grant)      ├─ WS: subscribe (signed grant)
  │                                    │
  [mutation]                           [mutation]
  │                                    │
  ├─ WS: invalidate (no data)          ├─ WS: invalidate (no data)
  ├─ HTTP: auto-refetch ──► re-render  ├─ HTTP: auto-refetch ──► re-render

1. useQuery fetches initial data over HTTP
2. It subscribes over WebSocket for updates (auth queries use a short-lived subscription grant)
3. After a mutation, the server either sends invalidate or, for custom public pushes, sends safe data
4. Your component re-renders automatically — no manual refetching

useQuery — Reactive Data

import { useQuery } from "@gencow/react";
import { api } from "../gencow/api";

function TaskList() {
    // useQuery returns { data, isLoading, isFetching, error, refetch }
    // createCrud().list returns { data: Task[], total: number }
    const { data: result, isLoading, isFetching, refetch } = useQuery(api.tasks.list, {});

    if (isLoading && !result) return <div>Loading...</div>;

    return (
        <div>
            <ul>
                {result?.data.map((t) => (
                    <li key={t.id}>{t.title}</li>
                ))}
            </ul>
            {/* Manual refetch (rarely needed — invalidation triggers automatic refetch) */}
            <button onClick={refetch} disabled={isFetching}>Refresh</button>
        </div>
    );
}

With Arguments

const task = useQuery(api.tasks.getById, { id: 42 });

Conditional Queries (Skip)

// Method A: "skip" token — recommended
const { data: messages } = useQuery(
    api.chat.getMessages,
    conversationId ? { conversationId } : "skip"
);

// Method B: enabled option (TanStack Query-style)
const { data: messages } = useQuery(
    api.chat.getMessages,
    { conversationId },
    { enabled: !!conversationId }
);
// → skip state returns undefined (no API call, no WS subscription)

Public Queries (No Auth)

// Generated api.ts carries allowAnonymous from the server definition — no hook options needed.
const publicPosts = useQuery(api.posts.listPublic);

Note: When your procedure uses .allowAnonymous() or createCrud(table, { allowAnonymous: true }), codegen emits { allowAnonymous: true } on the matching entry in api.ts. useQuery reads that default and runs without a JWT — you do not need hook options for public endpoints. The legacy { public: true } option still works but is deprecated. Server-side access control (Postgres RLS, ownerRls, etc.) is still enforced on the HTTP read — allowAnonymous only skips the auth gate, it does not bypass row policies.

useMutation — Write + Auto-Refresh

import { useMutation } from "@gencow/react";
import { api } from "../gencow/api";

function CreateTask() {
    const { mutate: create, isPending, error } = useMutation(api.tasks.create);

    const handleSubmit = async () => {
        await create({ title: "New task" });
        // No manual refetch — WebSocket invalidation triggers automatic HTTP refetch
    };

    return (
        <button onClick={handleSubmit} disabled={isPending}>
            {isPending ? "Creating..." : "Create"}
        </button>
    );
}

Key concept: After create() completes, the server invalidates related subscriptions over WebSocket. Each connected useQuery(api.tasks.list, {}) hook refetches over HTTP with its own JWT and re-renders with fresh, RLS-scoped data.

createCrud() List Scopes

Generated CRUD write procedures invalidate list subscribers automatically:

With the default createCrud(table) config, all list subscribers for the table share one bucket. Use explicit scoped realtime when a table has tenant, workspace, or other partitioned list views:

export const tasksCrud = createCrud(tasks, {
    prefix: "tasks",
    realtime: {
        policy: "authorized",
        resourceIdentity: "tasks",
        scope: (args) => ({ workspaceId: args?.filters?.workspaceId }),
        scopeFromRow: ({ row }) => ({ workspaceId: row.workspaceId }),
        authorize: async ({ ctx, args }) => {
            const user = ctx.auth.requireAuth();
            await assertWorkspaceMember(ctx.db, user.id, args?.filters?.workspaceId);
        },
    },
});

Set realtime: false to disable both list and exact get invalidation. Set realtime: { listRealtime: false } to disable list fanout while preserving exact get({ id }) invalidation for update and remove. Exact generated get watchers still subscribe through the normal generated get query contract, so public CRUD uses a public query grant and protected CRUD uses an authorized query grant.

Server-Side Realtime: invalidate(), refresh() & emit()

When writing custom write procedures (instead of createCrud()), use ctx.realtime to push updates to connected clients:

ctx.realtime.invalidate(queryKey)

Signals subscribed clients to refetch with their own auth token and query args. Recommended for private/RLS queries and most list updates.

import { procedure } from "./runtime";

export const markAllDone = procedure.mutation
    .name("tasks.markAllDone")
    .handler(async ({ context: ctx }) => {
        const session = ctx.auth.requireAuth();
        await ctx.db.update(tasks)
            .set({ done: true })
            .where(eq(tasks.userId, session.user.id));

        ctx.realtime.invalidate("tasks.list");
    });

ctx.realtime.refresh(queryKey)

Re-runs a query handler on the server and pushes the result to all WebSocket subscribers. Use this for public or server-safe aggregate queries where the server can safely recompute the result without a user-specific auth context.

ctx.realtime.emit(queryKey, data)

Push specific data directly without re-running the query. Use this only for public or opaque data that is safe for every subscriber of that query key.

Do not use emit() to push private/RLS row payloads. For private lists, call invalidate() so each client refetches with its own auth and query args. For private payload push, use an authorized realtime channel.

import { v } from "@gencow/core";
import { procedure } from "./runtime";

export const updatePublicTicker = procedure.mutation
    .name("metrics.updatePublicTicker")
    .input(v.object({ activeUsers: v.number() }))
    .handler(async ({ context: ctx, input }) => {
        await ctx.db.insert(publicMetrics).values(input);
        ctx.realtime.emit("metrics.publicTicker", input);
    });

Declarative authorized realtime channels

Use an authorized channel when you must push a private payload without making clients refetch a query. New channels are pure declarations in the app-owned API catalog: they are not process-global registrations.

For example, a conversation channel can deliver a new message directly to conversation members. A private task list usually works better with query invalidation: each client refetches the list under its own database policies. Channels are event subscriptions; they do not automatically update query caches.

Set up the builder

Bind the builder once in gencow/runtime.ts, alongside procedure. Import it from ./runtime everywhere you declare a channel.

// gencow/runtime.ts
import type { GencowSchema } from "./generated/db-schema.gen";
import {
    createGencowProcedureBuilders,
    createGencowRealtimeChannelBuilder,
} from "@gencow/core";

export const procedure = createGencowProcedureBuilders<GencowSchema>();
export const realtimeChannel = createGencowRealtimeChannelBuilder<GencowSchema>();

If the app has a custom user identity type, pass it as the second generic to both builders.

Declare and register a channel

.name(...) and the terminal .authorize(...) are required. .input(...) validates the client subscription scope, .event(...) validates each outbound payload, and .channelKey(...) derives one deterministic fanout key from the parsed input.

// gencow/conversation-events.ts
import { v } from "@gencow/core";
import { realtimeChannel } from "./runtime";

export const conversationEvents = realtimeChannel
    .name("conversation.events")
    .input(v.object({
        workspaceId: v.string(),
        conversationId: v.string(),
    }))
    .event(v.object({
        type: v.literal("message.created"),
        workspaceId: v.string(),
        conversationId: v.string(),
        message: v.object({ id: v.string(), body: v.string() }),
    }))
    .channelKey(
        ({ workspaceId, conversationId }) =>
            `workspace:${workspaceId}:conversation:${conversationId}`,
    )
    .authorize(async ({ context, input }) => {
        const user = context.auth.requireAuth();
        const membership = await requireConversationMember(context.db, user.id, input);
        return { ttlSeconds: 300, epoch: membership.epoch };
    });

Register it in the default gencow/index.ts export. The bag key becomes the client-generated property name; .name("conversation.events") remains the logical channel name and signed-grant selector.

// gencow/index.ts
import { defineApi } from "@gencow/core";
import { conversationEvents } from "./conversation-events";

export default defineApi({
    realtimeChannels: { conversationEvents },
});

The selected defineApi({ realtimeChannels }) catalog is the authority for which channels may be authorized, published, invalidated, or revoked. Do not register the same logical name twice.

Run gencow codegen after registering it to generate api.realtimeChannels.conversationEvents for the frontend.

Note — two names with different jobs: In this example, conversationEvents is the key your frontend imports from the generated API, while "conversation.events" is the channel's logical name. Unlike workflow starts, realtime client properties follow the registration key. Renaming that key requires updating frontend references and rerunning codegen.

Publish from trusted backend code

Pass the declaration object—not its string name—to get typed input and event payloads. publish() is asynchronous because a Standard Schema validator may be asynchronous.

import { v } from "@gencow/core";
import { procedure } from "./runtime";
import { conversationEvents } from "./conversation-events";

export const sendMessage = procedure.mutation
    .name("messages.send")
    .input(v.object({
        workspaceId: v.string(),
        conversationId: v.string(),
        body: v.string(),
    }))
    .handler(async ({ context: ctx, input }) => {
        const user = ctx.auth.requireAuth();
        await requireConversationMember(ctx.db, user.id, input);
        const message = await insertMessage(ctx.db, user.id, input);

        await ctx.realtime.channel(conversationEvents, {
            workspaceId: input.workspaceId,
            conversationId: input.conversationId,
        }).publish({
            type: "message.created",
            workspaceId: input.workspaceId,
            conversationId: input.conversationId,
            message: { id: message.id, body: message.body },
        });

        return message;
    });

requireConversationMember() and insertMessage() are app-owned helpers: the first checks membership in the requested workspace and conversation, and the second persists the message. Add sendMessage to the existing defineApi({ procedures: { sendMessage } }) registration.

Note — subscriber checks and write checks: .authorize() runs when a subscriber obtains or refreshes a grant. Publishing does not call it again for the writer or every recipient. The mutation must check the writer's permission separately before saving or publishing, as above. Requiring a login alone does not prove membership in a conversation.

channelKey() must be deterministic, side-effect-free, and inexpensive: do not read the database, network, clock, random values, or authorization state in it. Put access checks in .authorize(). It may return false to deny a subscription or an object with optional ttlSeconds and epoch to allow it. When access changes, persist the new membership state and epoch first, then revoke grants that do not match the new epoch:

await ctx.realtime.channel(conversationEvents, { workspaceId, conversationId }).revoke({
    epoch: newMembershipEpoch,
    reason: "membership_changed",
});

Advanced — scope and revocation: Every subscriber mapped to one channelKey receives the same event payload. Include both workspace and conversation identity when IDs are not globally unique. Return the current access epoch from .authorize(); revoke({ epoch: newMembershipEpoch }) removes subscriptions with a different epoch, while revoke() removes all subscriptions for that key. Clients request a fresh grant, so updated membership checks must deny users whose access was removed.

Event payloads are JSON wire DTOs. Do not publish Date, Map, bigint, undefined, functions, class instances, circular values, or non-finite numbers. For a date, declare createdAt: v.string() in the event schema and publish createdAt: new Date().toISOString().

Advanced — transformed event schemas: publish() accepts a validator's input, while listeners receive its normalized output. Codegen can describe that output precisely only if the validator supplies representable Standard JSON Schema v1 output metadata (~standard.jsonSchema.output(...)). Otherwise the generated listener type is unknown. A plain v.object(...) event schema is a straightforward choice when you want the same JSON shape at both ends.

Subscribe with the generated ref

After gencow codegen, subscribe with the generated ref. It carries the input and event types but never ships the authorizer or key builder to the browser. Passing null or enabled: false skips the subscription.

import { useRealtimeChannel } from "@gencow/react";
import { api } from "../gencow/api";

type ConversationProps = { workspaceId: string; conversationId?: string };

function Conversation({ workspaceId, conversationId }: ConversationProps) {
    useRealtimeChannel(
        conversationId ? {
            channel: api.realtimeChannels.conversationEvents,
            args: { workspaceId, conversationId },
        } : null,
        (event) => {
            console.log("conversation event", event.message);
        },
    );

    return null;
}

The hook requests a short-lived authorized grant and automatically resubscribes when the authenticated session or channel args change.

Note — receiving an event is not loading history: The listener handles events delivered while subscribed. Load existing messages with a query, and reconcile against that query after a reconnect when missed messages matter. Use functional state updates and upsert messages by ID if both the mutation response and a channel event can add the same message to your UI.

Deprecated: defineRealtimeChannels() and string channels

defineRealtimeChannels() and ctx.realtime.channel("name", args) remain only as untyped migration compatibility APIs. defineRealtimeChannels() writes to the old global registry, while new declarations are app-scoped. Do not use either form in new code.

// Deprecated legacy API.
import { defineRealtimeChannels, v } from "@gencow/core";

export const legacyChannels = defineRealtimeChannels({
    "conversation.events": {
        args: v.object({ conversationId: v.string() }),
        authorize: (ctx) => {
            ctx.auth.requireAuth();
            return { ttlSeconds: 300 };
        },
    },
});

ctx.realtime.channel("conversation.events", { conversationId }).publish({ type: "message.created" });

Migrate to a realtimeChannel.name(...).input(...).event(...).authorize(...) definition, register it in defineApi({ realtimeChannels }), publish with the declaration object, and subscribe with api.realtimeChannels.<bagKey>. During the migration, keep the logical .name("conversation.events") and the same scope inputs and key mapping when you intend to preserve the existing channel. Remove its old registry entry and run gencow codegen.

Note — incremental migration: Other legacy channels can continue to work. realtimeChannels: {} does not disable legacy registrations, and the same logical channel name cannot be registered through both forms. Existing string subscribers can resolve a migrated declaration during compatibility; switch them to generated refs to gain input and event typing.

When to Use What

Warning: If a custom mutation doesn't call emit(), invalidate(), or refresh(), the server logs a warning — connected clients won't see the update until they refresh manually.

useStatus — Connection Health

Monitor the WebSocket connection status:

import { useStatus } from "@gencow/react";

function ConnectionIndicator() {
    const { isConnected } = useStatus();

    return (
        <span style={{ color: isConnected ? "green" : "red" }}>
            {isConnected ? "● Connected" : "○ Disconnected"}
        </span>
    );
}

The WebSocket automatically reconnects with exponential backoff (1s → 1.5s → 2.25s → ... up to 15s).

Cloud Plan Limits

Realtime limits count concurrent WebSocket connections, not HTTP visitors or API requests. One browser tab with active realtime queries usually consumes one connection.

The account limit is shared by every app owned by the account. The per-app safety limit prevents one app from consuming the entire allowance. Gencow Cloud checks both limits when each WebSocket connection opens. Enterprise limits are set by agreement.

❌ Don't Use fetch() Directly

// ❌ WRONG — bypasses realtime, no auto-refresh
fetch("/api/query", {
    method: "POST",
    body: JSON.stringify({ name: "tasks.list", args: {} }),
});

// ❌ WRONG — creating manual wrappers
const apiPost = (name, args) => fetch("/api/mutation", { ... });

// ✅ CORRECT — use hooks for automatic realtime sync
const { data: tasks } = useQuery(api.tasks.list, {});
const { mutate: create } = useMutation(api.tasks.create);

Next Steps

• Storage — File upload & download
• Cron Jobs — Scheduled tasks
• Deployment — Cloud deployment