feat: Message type guard helpers (#1572)

Addresses #1453

Adds the type guards to check whether a message is a `ControlMessage` or
a `ChangeMessage`, documented, tested, and exported.

I've also honed in on the message type definitions a bit more as I had
experienced an uncaught bug due to the generality of the `Headers` type.

.changeset/pretty-donuts-rule.md

@@ -0,0 +1,5 @@
+---
+"@electric-sql/client": patch
+---
+
+Add `Message` type guard helpers `isChangeMessage` and `isControlMessage`.

examples/nextjs-example/app/match-stream.ts

@@ -1,4 +1,8 @@
-import { ShapeStream, ChangeMessage } from "@electric-sql/client"
+import {
+  ShapeStream,
+  ChangeMessage,
+  isChangeMessage,
+} from "@electric-sql/client"
 
 export async function matchStream<T>({
   stream,
@@ -21,7 +25,7 @@ export async function matchStream<T>({
     const unsubscribe = stream.subscribe((messages) => {
       for (const message of messages) {
         if (
-          `key` in message &&
+          isChangeMessage(message) &&
           operations.includes(message.headers.operation)
         ) {
           if (

examples/redis-client/src/index.ts

@@ -1,5 +1,5 @@
 import { createClient } from 'redis'
-import { ShapeStream, Message } from '@electric-sql/client'
+import { ShapeStream, Message, isChangeMessage } from '@electric-sql/client'
 
 // Create a Redis client
 const REDIS_HOST = `localhost`
@@ -21,7 +21,7 @@ client.connect().then(() => {
 
     // Loop through each message and make writes to the Redis hash for action messages
     messages.forEach((message) => {
-      if (!(`key` in message)) return
+      if (!isChangeMessage(message)) return
       // Upsert/delete
       switch (message.headers.action) {
         case `delete`:

examples/remix-basic/app/match-stream.ts

@@ -1,4 +1,8 @@
-import { ShapeStream, ChangeMessage } from "@electric-sql/client"
+import {
+  ShapeStream,
+  ChangeMessage,
+  isChangeMessage,
+} from "@electric-sql/client"
 
 export async function matchStream({
   stream,
@@ -21,7 +25,7 @@ export async function matchStream({
     const unsubscribe = stream.subscribe((messages) => {
       for (const message of messages) {
         if (
-          `key` in message &&
+          isChangeMessage(message) &&
           operations.includes(message.headers.operation)
         ) {
           if (matchFn({ operationType: message.headers.operation, message })) {

packages/typescript-client/src/client.ts

@@ -1,6 +1,6 @@
-import { ArgumentsType } from 'vitest'
 import { Message, Value, Offset, Schema } from './types'
 import { MessageParser, Parser } from './parser'
+import { isChangeMessage, isControlMessage } from './helpers'
 
 export type ShapeData = Map<string, { [key: string]: Value }>
 export type ShapeChangedCallback = (value: ShapeData) => void
@@ -196,7 +196,7 @@ export class ShapeStream {
     this.backoffOptions = options.backoffOptions ?? BackoffDefaults
     this.fetchClient =
       options.fetchClient ??
-      ((...args: ArgumentsType<typeof fetch>) => fetch(...args))
+      ((...args: Parameters<typeof fetch>) => fetch(...args))
 
     this.start()
   }
@@ -270,7 +270,8 @@ export class ShapeStream {
       if (batch.length > 0) {
         const lastMessage = batch[batch.length - 1]
         if (
-          lastMessage.headers?.[`control`] === `up-to-date` &&
+          isControlMessage(lastMessage) &&
+          lastMessage.headers.control === `up-to-date` &&
           !this.isUpToDate
         ) {
           this.isUpToDate = true
@@ -514,7 +515,7 @@ export class Shape {
     let newlyUpToDate = false
 
     messages.forEach((message) => {
-      if (`key` in message) {
+      if (isChangeMessage(message)) {
         dataMayHaveChanged = [`insert`, `update`, `delete`].includes(
           message.headers.operation
         )
@@ -535,19 +536,22 @@ export class Shape {
         }
       }
 
-      if (message.headers?.[`control`] === `up-to-date`) {
-        isUpToDate = true
-        if (!this.hasNotifiedSubscribersUpToDate) {
-          newlyUpToDate = true
+      if (isControlMessage(message)) {
+        switch (message.headers.control) {
+          case `up-to-date`:
+            isUpToDate = true
+            if (!this.hasNotifiedSubscribersUpToDate) {
+              newlyUpToDate = true
+            }
+            break
+          case `must-refetch`:
+            this.data.clear()
+            this.error = false
+            isUpToDate = false
+            newlyUpToDate = false
+            break
         }
       }
-
-      if (message.headers?.[`control`] === `must-refetch`) {
-        this.data.clear()
-        this.error = false
-        isUpToDate = false
-        newlyUpToDate = false
-      }
     })
 
     // Always notify subscribers when the Shape first is up to date.

packages/typescript-client/src/helpers.ts

@@ -0,0 +1,47 @@
+import { ChangeMessage, ControlMessage, Message, Value } from './types'
+
+/**
+ * Type guard for checking {@link Message} is {@link ChangeMessage}.
+ *
+ * See [TS docs](https://www.typescriptlang.org/docs/handbook/advanced-types.html#user-defined-type-guards)
+ * for information on how to use type guards.
+ *
+ * @param message - the message to check
+ * @returns true if the message is a {@link ChangeMessage}
+ *
+ * @example
+ * ```ts
+ * if (isChangeMessage(message)) {
+ *   const msgChng: ChangeMessage = message // Ok
+ *   const msgCtrl: ControlMessage = message // Err, type mismatch
+ * }
+ * ```
+ */
+export function isChangeMessage<T extends Value = { [key: string]: Value }>(
+  message: Message<T>
+): message is ChangeMessage<T> {
+  return `key` in message
+}
+
+/**
+ * Type guard for checking {@link Message} is {@link ControlMessage}.
+ *
+ * See [TS docs](https://www.typescriptlang.org/docs/handbook/advanced-types.html#user-defined-type-guards)
+ * for information on how to use type guards.
+ *
+ * @param message - the message to check
+ * @returns true if the message is a {@link ControlMessage}
+ *
+ *  * @example
+ * ```ts
+ * if (isControlMessage(message)) {
+ *   const msgChng: ChangeMessage = message // Err, type mismatch
+ *   const msgCtrl: ControlMessage = message // Ok
+ * }
+ * ```
+ */
+export function isControlMessage<T extends Value = { [key: string]: Value }>(
+  message: Message<T>
+): message is ControlMessage {
+  return !isChangeMessage(message)
+}

packages/typescript-client/src/index.ts

@@ -1,2 +1,3 @@
 export * from './client'
 export * from './types'
+export * from './helpers'

packages/typescript-client/src/types.ts

@@ -10,11 +10,11 @@ export type Value =
 export type Offset = `-1` | `${number}_${number}`
 
 interface Header {
-  [key: string]: Value
+  [key: Exclude<string, `operation` | `control`>]: Value
 }
 
 export type ControlMessage = {
-  headers: Header
+  headers: Header & { control: `up-to-date` | `must-refetch` }
 }
 
 export type ChangeMessage<T> = {

packages/typescript-client/test/client.test.ts

@@ -1,4 +1,4 @@
-import { ArgumentsType, describe, expect, inject, vi } from 'vitest'
+import { describe, expect, inject, vi } from 'vitest'
 import { v4 as uuidv4 } from 'uuid'
 import { setTimeout as sleep } from 'node:timers/promises'
 import { testWithIssuesTable as it } from './support/test-context'
@@ -119,7 +119,7 @@ describe(`Shape`, () => {
     })
 
     let requestsMade = 0
-    const fetchWrapper = async (...args: ArgumentsType<typeof fetch>) => {
+    const fetchWrapper = async (...args: Parameters<typeof fetch>) => {
       // clear the shape and modify the data after the initial request
       if (requestsMade === 1) {
         await clearIssuesShape()

packages/typescript-client/test/helpers.test-d.ts

@@ -0,0 +1,45 @@
+import { describe, expectTypeOf, it } from 'vitest'
+import {
+  ChangeMessage,
+  ControlMessage,
+  isChangeMessage,
+  isControlMessage,
+  Message,
+} from '../src'
+
+describe(`helpers`, () => {
+  it(`should respect ChangeMessages type guard`, () => {
+    const message = {
+      headers: {
+        operation: `insert`,
+      },
+      offset: `-1`,
+      key: `foo`,
+      value: { foo: `bar` },
+    } as Message<{ foo: string }>
+
+    if (isChangeMessage(message)) {
+      const msgChng: ChangeMessage<{ foo: string }> = message
+      expectTypeOf(msgChng).toEqualTypeOf<ChangeMessage<{ foo: string }>>()
+
+      // @ts-expect-error - should have type mismatch
+      message as ControlMessage
+    }
+  })
+
+  it(`should respect ControlMessages type guard`, () => {
+    const message = {
+      headers: {
+        control: `up-to-date`,
+      },
+    } as Message<{ [key: string]: string }>
+
+    if (isControlMessage(message)) {
+      const msgCtrl: ControlMessage = message
+      expectTypeOf(msgCtrl).toEqualTypeOf<ControlMessage>()
+
+      // @ts-expect-error - should have type mismatch
+      message as ChangeMessage<{ foo: string }>
+    }
+  })
+})

packages/typescript-client/test/helpers.test.ts

@@ -0,0 +1,28 @@
+import { describe, expect, it } from 'vitest'
+import { isChangeMessage, isControlMessage, Message } from '../src'
+
+describe(`helpers`, () => {
+  it(`should correctly detect ChangeMessages`, () => {
+    const message = {
+      headers: {
+        operation: `insert`,
+      },
+      offset: `-1`,
+      key: `key`,
+      value: { key: `value` },
+    } as Message
+
+    expect(isChangeMessage(message)).toBe(true)
+    expect(isControlMessage(message)).toBe(false)
+  })
+
+  it(`should correctly detect ControlMessages`, () => {
+    const message = {
+      headers: {
+        control: `up-to-date`,
+      },
+    } as Message
+    expect(isControlMessage(message)).toBe(true)
+    expect(isChangeMessage(message)).toBe(false)
+  })
+})