feat(nuxt drizzle): realtime improvements and fixes

Author: Akryum

docs/plugins/nuxt-drizzle.md

@@ -17,6 +17,10 @@
       "types": "./dist/types.d.ts",
       "import": "./dist/module.mjs",
       "require": "./dist/module.cjs"
+    },
+    "./realtime": {
+      "types": "./dist/runtime/realtime-protocol.d.ts",
+      "import": "./dist/runtime/realtime-protocol.js"
     }
   },
   "main": "./dist/module.cjs",

packages/nuxt-drizzle/package.json

@@ -38,10 +38,57 @@ export interface ModuleOptions {
   apiPath?: string
 
   /**
-   * Enable WebSocket support for real-time updates
+   * Enable WebSocket support for real-time updates.
+   *
+   * Pass `true` to enable with defaults, or an object to customize the
+   * server-side route and/or the client-side endpoint.
    */
   ws?: boolean | {
+    /**
+     * Server-side route where the generated WebSocket handler is registered
+     * (and, by default, the URL the client connects to).
+     *
+     * @default '/api/rstore-realtime/ws'
+     */
     apiPath?: string
+
+    /**
+     * Fully overrides the URL the client uses to connect to the realtime
+     * WebSocket. Useful when the WebSocket is hosted on a different
+     * server (for example a dedicated realtime service).
+     *
+     * Can be an absolute URL (`wss://realtime.example.com/ws`), a
+     * protocol-relative URL (`//realtime.example.com/ws`), or a path
+     * relative to the current origin (`/ws`).
+     *
+     * When set, the server handler is still registered at `apiPath` so
+     * you can keep using the built-in handler during development — point
+     * this option at your external server in production via a runtime
+     * config override if needed.
+     *
+     * @default apiPath
+     */
+    clientEndpoint?: string
+
+    /**
+     * Heartbeat interval in milliseconds. The client sends a `ping` frame
+     * at this rate and expects a `pong` reply to detect dead connections.
+     *
+     * @default 10000
+     */
+    heartbeatInterval?: number
+
+    /**
+     * Auto-reconnect configuration forwarded to `@vueuse/core`'s
+     * `useWebSocket`. Pass `true` for defaults, `false` to disable, or an
+     * object to tune retries and backoff.
+     *
+     * @default true
+     */
+    autoReconnect?: boolean | {
+      retries?: number
+      delay?: number
+    }
   }
 
   /**
@@ -516,6 +563,11 @@ ${collections.map((collection, index) => {
 
     const wsOptions = typeof options.ws === 'object' ? options.ws : {}
     const wsApiPath = wsOptions.apiPath ?? `/api/rstore-realtime/ws`
+    // Client-side endpoint URL — falls back to the server route so same-origin
+    // setups keep working without extra configuration.
+    const wsClientEndpoint = wsOptions.clientEndpoint ?? wsApiPath
+    const wsHeartbeatInterval = wsOptions.heartbeatInterval ?? 10000
+    const wsAutoReconnect = wsOptions.autoReconnect ?? true
 
     if (options.ws) {
       nuxt.options.nitro.experimental ??= {}
@@ -582,6 +634,9 @@ export default createOfflinePlugin({
       filename: '$rstore-drizzle-config.js',
       getContents: () => `export const apiPath = ${JSON.stringify(apiPath)}
 export const wsApiPath = ${JSON.stringify(wsApiPath)}
+export const wsClientEndpoint = ${JSON.stringify(wsClientEndpoint)}
+export const wsHeartbeatInterval = ${JSON.stringify(wsHeartbeatInterval)}
+export const wsAutoReconnect = ${JSON.stringify(wsAutoReconnect)}
 export const dialect = '${drizzleConfig.dialect}'
 export const syncSerializeDateValue = ${offlineOptions?.serializeDateValue ? offlineOptions.serializeDateValue.toString() : 'undefined'}\n`,
     })

packages/nuxt-drizzle/src/module.ts

@@ -1,9 +1,10 @@
 import type { SubscriptionMessage, SubscriptionUpdateMessage } from './utils/realtime'
 // @ts-expect-error virtual module
-import { wsApiPath } from '#build/$rstore-drizzle-config.js'
+import { wsAutoReconnect, wsClientEndpoint, wsHeartbeatInterval } from '#build/$rstore-drizzle-config.js'
 import { definePlugin, realtimeReconnectEventHook } from '@rstore/vue'
 import { useWebSocket } from '@vueuse/core'
 import { watch } from 'vue'
+import { getRstoreDrizzleClientId } from './utils/client-id'
 import { getSubscriptionId } from './utils/realtime'
 
 export default definePlugin({
@@ -16,14 +17,20 @@ export default definePlugin({
 
   setup({ hook }) {
     if (import.meta.client) {
+      const clientId = getRstoreDrizzleClientId()
+
+      // Local ref-count per subscription id — a given (collection, key, where)
+      // is only sent to the server once, regardless of how many components
+      // subscribe to it locally. The matching `unsubscribe` is only emitted
+      // when the last subscriber is torn down.
       const countPerTopic: Record<string, number> = {}
       const messages = new Map<string, SubscriptionMessage>()
 
-      const ws = useWebSocket(wsApiPath, {
+      const ws = useWebSocket(wsClientEndpoint, {
         heartbeat: {
-          interval: 10000,
+          interval: wsHeartbeatInterval,
         },
-        autoReconnect: true,
+        autoReconnect: wsAutoReconnect,
       })
 
       let connectCount = 0
@@ -56,8 +63,14 @@ export default definePlugin({
           where,
         }
         const subscriptionId = getSubscriptionId(message)
-        countPerTopic[subscriptionId] ??= 1
-        countPerTopic[subscriptionId]--
+        const current = countPerTopic[subscriptionId] ?? 0
+        // Guard against decrementing below zero — a stray unsubscribe without
+        // a matching subscribe would otherwise send a spurious unsubscribe
+        // frame and corrupt the counter.
+        if (current <= 0) {
+          return
+        }
+        countPerTopic[subscriptionId] = current - 1
         if (countPerTopic[subscriptionId] === 0) {
           ws.send(JSON.stringify({
             subscription: message,
@@ -67,8 +80,8 @@ export default definePlugin({
       })
 
       hook('init', ({ store }) => {
-        watch(ws.data, async (data: string) => {
-          if (data === 'pong') {
+        watch(ws.data, async (data) => {
+          if (typeof data !== 'string' || data === 'pong') {
             return
           }
           try {
@@ -112,23 +125,26 @@ export default definePlugin({
       })
 
       watch(ws.status, (status) => {
-        if (status === 'CLOSED') {
-          // Reset counts on reconnect
-          for (const key in countPerTopic) {
-            countPerTopic[key] = 0
-          }
-        }
-        else if (status === 'OPEN') {
+        if (status === 'OPEN') {
           connectCount++
 
-          // Resubscribe to all topics
+          // Announce our clientId first so the server can tag us for
+          // skip-self echo suppression before any mutation echoes arrive.
+          if (clientId) {
+            ws.send(JSON.stringify({ init: { clientId } }), false)
+          }
+
+          // Resubscribe to all active topics. On first open this replays
+          // subscriptions issued while CONNECTING; on reconnect it restores
+          // subscriptions across a transient disconnect.
           for (const [, message] of messages) {
             ws.send(JSON.stringify({
               subscription: message,
             }), false)
           }
 
-          // Call reconnect hook
+          // Notify live queries to refresh so updates missed while offline
+          // are recovered. Only fire on true reconnects (skip first open).
           if (connectCount > 1) {
             realtimeReconnectEventHook.trigger()
           }

packages/nuxt-drizzle/src/runtime/plugin-realtime.ts

@@ -5,9 +5,20 @@ import { apiPath, dialect } from '#build/$rstore-drizzle-config.js'
 import { useRequestFetch } from '#imports'
 import { definePlugin } from '@rstore/vue'
 import SuperJSON from 'superjson'
+import { getRstoreDrizzleClientId, RSTORE_DRIZZLE_CLIENT_ID_HEADER } from './utils/client-id'
 import { and, eq } from './utils/where'
 import { filterWhere } from './where'
 
+/**
+ * Returns a headers object including the per-tab client id header when
+ * running in the browser. The server-side realtime handler uses this id to
+ * skip echoing the resulting update frame back to the originating client.
+ */
+function clientIdHeaders(): Record<string, string> | undefined {
+  const id = getRstoreDrizzleClientId()
+  return id ? { [RSTORE_DRIZZLE_CLIENT_ID_HEADER]: id } : undefined
+}
+
 export default definePlugin({
   name: 'rstore-drizzle',
 
@@ -171,6 +182,7 @@ export default definePlugin({
       const result: any = await requestFetch(`${apiPath}/${payload.collection.name}`, {
         method: 'POST',
         body: SuperJSON.stringify(payload.item),
+        headers: clientIdHeaders(),
       })
       payload.setResult(result)
     })
@@ -186,13 +198,15 @@ export default definePlugin({
       const result: any = await requestFetch(`${apiPath}/${payload.collection.name}/${payload.key}`, {
         method: 'PATCH',
         body: SuperJSON.stringify(body),
+        headers: clientIdHeaders(),
       })
       payload.setResult(result)
     })
 
     hook('deleteItem', async (payload) => {
       await requestFetch(`${apiPath}/${payload.collection.name}/${payload.key}`, {
         method: 'DELETE',
+        headers: clientIdHeaders(),
       })
     })
 
@@ -267,6 +281,7 @@ export default definePlugin({
           method: 'POST',
           body: SuperJSON.stringify({ operations: wireOps }),
           responseType: 'text',
+          headers: clientIdHeaders(),
         })
         response = SuperJSON.parse(raw) as BatchWireResponse
       }

packages/nuxt-drizzle/src/runtime/plugin.ts

@@ -0,0 +1,58 @@
+/**
+ * Public entry point for the realtime protocol used by `@rstore/nuxt-drizzle`.
+ *
+ * Import from `@rstore/nuxt-drizzle/realtime` to build a custom realtime server
+ * (or a non-Vue client) that speaks the same WebSocket protocol as the built-in
+ * plugin. This module has **no Nuxt, Nitro, Vue, Drizzle, or H3 runtime
+ * dependencies**, so it is safe to import from a standalone Node / Bun / Deno
+ * / edge service.
+ *
+ * Protocol reference:
+ * https://rstore.akryum.dev/plugins/nuxt-drizzle.html#realtime-protocol
+ */
+
+// Pub/Sub — transport-agnostic helpers for fanning out updates.
+// The in-memory default is single-process; swap it out with `setPubSub()` for
+// multi-node deployments (Redis, NATS, Postgres LISTEN/NOTIFY, …).
+export type {
+  PubSub,
+  RstoreDrizzlePubSub,
+  RstoreDrizzlePubSubChannels,
+} from './server/utils/pubsub'
+
+export {
+  createMemoryPubSub,
+  getPubSub,
+  setPubSub,
+  usePeerPubSub,
+} from './server/utils/pubsub'
+
+// Protocol message types & helpers
+export type {
+  ClientInitMessage,
+  RstoreDrizzleRealtimePayload,
+  RstoreDrizzleRealtimeUpdateType,
+  SubscriptionMessage,
+  SubscriptionUpdateMessage,
+} from './utils/realtime'
+
+export { getSubscriptionId, normalizeSubscriptionKey } from './utils/realtime'
+
+// Subscription matcher — pure function used by the built-in WebSocket
+// handler; re-exported so custom realtime servers can share the exact
+// match semantics without re-implementing them.
+export type { RstoreDrizzleDialect } from './utils/subscription-match'
+
+export { subscriptionMatches } from './utils/subscription-match'
+
+// `where` filter — operator AST + evaluator
+export type {
+  RestoreDrizzleConditionModifier,
+  RstoreDrizzleBinaryOperator,
+  RstoreDrizzleCondition,
+  RstoreDrizzleConditionGroup,
+  RstoreDrizzleTernaryOperator,
+  RstoreDrizzleUnaryOperator,
+} from './utils/types'
+
+export { filterWhere } from './where'