Neon Realtime

import { and, eq, inArray, ne, sql } from "drizzle-orm"
import { Hono } from "hono"
import { HTTPException } from "hono/http-exception"
import { validator } from "hono/validator"

import { requireUser } from "./auth"
import { db } from "./db"
import { channelMembers, messages } from "./schema"

export type MutationResponse =
  | { readonly kind: "committed"; readonly txid: string }
  | { readonly kind: "unchanged" }

// The id of the transaction that made the change, read in the write's own RETURNING
const txid = sql<string>`pg_current_xact_id()::text`

const mutationResponse = (row?: { txid: string }): MutationResponse =>
  row ? { kind: "committed", txid: row.txid } : { kind: "unchanged" }

const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i

function isUuid(value: unknown): value is string {
  return typeof value === "string" && UUID.test(value)
}

function isBody(value: unknown): value is string {
  return typeof value === "string" && value.trim() !== ""
}

const app = new Hono()
  // ...the sealing routes: /api/v1/messages/live and /api/v1/inbox/live
  .post(
    "/api/v1/messages",
    validator("json", (value, c) => {
      const batch: unknown = value?.batch

      if (
        !Array.isArray(batch) ||
        batch.length === 0 ||
        !batch.every(
          (item): item is { id: string; channelId: number; body: string } =>
            isUuid(item?.id) && Number.isInteger(item?.channelId) && isBody(item?.body)
        )
      ) {
        return c.json({ error: "batch must be a non-empty array of { id, channelId, body }" }, 400)
      }

      return { batch }
    }),
    async (c) => {
      const user = await requireUser(c)
      const { batch } = c.req.valid("json")
      const channelIds = [...new Set(batch.map((message) => message.channelId))]
      const rows = batch.map(({ id, channelId, body }) => ({ id, channel_id: channelId, body }))

      // One statement: the batch is written only if the user is a member of every channel in it
      const [sent] = await db
        .insert(messages)
        .select(
          db
            .select({
              id: sql`batch.id`.as("id"),
              channel_id: sql`batch.channel_id`.as("channel_id"),
              author_id: sql`${user.id}`.as("author_id"),
              body: sql`batch.body`.as("body"),
              created_at: sql`now()`.as("created_at"),
            })
            .from(sql`jsonb_to_recordset(${JSON.stringify(rows)}::jsonb) as batch(id uuid, channel_id integer, body text)`)
            .where(
              eq(
                db.$count(
                  channelMembers,
                  and(eq(channelMembers.user_id, user.id), inArray(channelMembers.channel_id, channelIds))
                ),
                channelIds.length
              )
            )
        )
        .returning({ txid })

      if (!sent) {
        throw new HTTPException(403, { message: "Not a member of this channel" })
      }

      return c.json(mutationResponse(sent))
    }
  )
  .patch(
    "/api/v1/messages",
    validator("json", (value, c) => {
      const batch: unknown = value?.batch

      if (
        !Array.isArray(batch) ||
        batch.length === 0 ||
        !batch.every((item): item is { id: string; body: string } => isUuid(item?.id) && isBody(item?.body))
      ) {
        return c.json({ error: "batch must be a non-empty array of { id, body }" }, 400)
      }

      return { batch }
    }),
    async (c) => {
      const user = await requireUser(c)
      const { batch } = c.req.valid("json")

      const [edited] = await db
        .update(messages)
        .set({ body: sql`edit.body` })
        .from(sql`jsonb_to_recordset(${JSON.stringify(batch)}::jsonb) as edit(id uuid, body text)`)
        .where(and(eq(messages.id, sql`edit.id`), eq(messages.author_id, user.id), ne(messages.body, sql`edit.body`)))
        .returning({ txid })

      return c.json(mutationResponse(edited))
    }
  )
  .delete(
    "/api/v1/messages",
    validator("json", (value, c) => {
      const batch: unknown = value?.batch

      if (!Array.isArray(batch) || batch.length === 0 || !batch.every(isUuid)) {
        return c.json({ error: "batch must be a non-empty array of uuids" }, 400)
      }

      return { batch }
    }),
    async (c) => {
      const user = await requireUser(c)
      const { batch } = c.req.valid("json")

      const [deleted] = await db
        .delete(messages)
        .where(and(inArray(messages.id, batch), eq(messages.author_id, user.id)))
        .returning({ txid })

      return c.json(mutationResponse(deleted))
    }
  )

export type AppType = typeof app

export default app

import type { AppType, MutationResponse } from "../backend/server"

export async function sendMessages(
  batch: Pick<Message, "id" | "channel_id" | "body">[]
): Promise<MutationResponse> {
  const res = await server.api.v1.messages.$post({
    json: { batch: batch.map(({ id, channel_id, body }) => ({ id, channelId: channel_id, body })) }
  })

  if (!res.ok) {
    throw new Error(`Could not send the messages (${res.status})`)
  }

  return res.json()
}

export async function editMessages(batch: { id: string; body: string }[]): Promise<MutationResponse> {
  const res = await server.api.v1.messages.$patch({ json: { batch } })

  if (!res.ok) {
    throw new Error(`Could not edit the messages (${res.status})`)
  }

  return res.json()
}

export async function deleteMessages(ids: string[]): Promise<MutationResponse> {
  const res = await server.api.v1.messages.$delete({ json: { batch: ids } })

  if (!res.ok) {
    throw new Error(`Could not delete the messages (${res.status})`)
  }

  return res.json()
}

import type { Message } from "../backend/schema"
import type { MutationResponse } from "../backend/server"
import { deleteMessages, editMessages, sendMessages } from "./api"

export type MessageMutation =
  | { readonly type: "send"; readonly message: Message }
  | { readonly type: "edit"; readonly id: string; readonly body: string }
  | { readonly type: "delete"; readonly id: string }

export type OptimisticMessage = Message & {
  readonly pending?: boolean;
}

// The server sets its own author and timestamp, but keeps the client's id
export function newMessage(channelId: number, authorId: string, body: string): Message {
  return { id: crypto.randomUUID(), channel_id: channelId, author_id: authorId, body, created_at: new Date() }
}

export function applyMutation(
  messages: readonly OptimisticMessage[],
  mutation: MessageMutation
): readonly OptimisticMessage[] {
  switch (mutation.type) {
    case "send":
      // Once the sent message has synced, show the synced row only
      if (messages.some((message) => message.id === mutation.message.id)) {
        return messages
      }

      // Newest first, to match the query's order
      return [{ ...mutation.message, pending: true }, ...messages]
    case "edit":
      return messages.map((message) =>
        message.id === mutation.id ? { ...message, body: mutation.body, pending: true } : message
      )
    case "delete":
      return messages.filter((message) => message.id !== mutation.id)
  }
}

export function writeMutation(mutation: MessageMutation): Promise<MutationResponse> {
  switch (mutation.type) {
    case "send":
      return sendMessages([mutation.message])
    case "edit":
      return editMessages([{ id: mutation.id, body: mutation.body }])
    case "delete":
      return deleteMessages([mutation.id])
  }
}

import type { LiveQueryState, MaterializedLiveQuerySubscription } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { applyMutation, type MessageMutation, type OptimisticMessage, writeMutation } from "./mutations"

export function optimisticMessages(
  subscription: MaterializedLiveQuerySubscription<Message>,
  render: (messages: readonly OptimisticMessage[], state: LiveQueryState, error: string | undefined) => void
) {
  // A Map keeps insertion order, so mutations apply in the order they were made
  const pending = new Map<number, MessageMutation>()
  let nextKey = 0
  let error: string | undefined
  let stopped = false

  function update() {
    if (stopped) {
      return
    }

    const snapshot = subscription.getSnapshot()
    const synced: readonly OptimisticMessage[] = snapshot.data ?? []

    render([...pending.values()].reduce(applyMutation, synced), snapshot, error)
  }

  async function mutate(mutation: MessageMutation) {
    const key = nextKey++

    error = undefined
    pending.set(key, mutation)
    update()

    try {
      const result = await writeMutation(mutation)

      if (result.kind === "committed") {
        await subscription.awaitTxId(result.txid, 10_000)
      }
    } catch {
      error = "Couldn't confirm the change"
    } finally {
      pending.delete(key)
      update()
    }
  }

  const unsubscribe = subscription.onChange(update)

  update()

  return {
    mutate,
    stop: () => {
      stopped = true
      unsubscribe()
    }
  }
}

import { type FormEvent, startTransition, useEffect, useOptimistic, useState } from "react"
import { useLiveQuery } from "@neon/realtime-react"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { LocalTime } from "./LocalTime"
import {
  applyMutation,
  type MessageMutation,
  newMessage,
  type OptimisticMessage,
  writeMutation
} from "./mutations"

const NO_MESSAGES: readonly OptimisticMessage[] = []

interface Props {
  sealedQuery: SealedLiveQuery<Message>;
  refreshQuery: () => Promise<SealedLiveQuery<Message>>;
  initialData?: readonly Message[];
  channelId: number;
  userId: string;
}

export function Messages({ sealedQuery, refreshQuery, initialData, channelId, userId }: Props) {
  const { data, status, error, utils } = useLiveQuery(sealedQuery, { refreshQuery, initialData })

  const [messages, addOptimistic] = useOptimistic<readonly OptimisticMessage[], MessageMutation>(
    data ?? NO_MESSAGES,
    applyMutation
  )

  const [mutationError, setMutationError] = useState<string>()

  useEffect(() => {
    return utils.onUpdate((update) => {
      if (update.type === "changes") {
        console.debug(`Applied ${update.changes.length} changes from transactions ${update.batch.txids.join(", ")}`)
      }
    })
  }, [utils])

  function mutate(mutation: MessageMutation) {
    setMutationError(undefined)

    startTransition(async () => {
      addOptimistic(mutation)

      try {
        const result = await writeMutation(mutation)

        if (result.kind === "committed") {
          await utils.awaitTxId(result.txid, 10_000)
        }
      } catch {
        setMutationError("Couldn't confirm the change")
      }
    })
  }

  function send(event: FormEvent<HTMLFormElement>) {
    event.preventDefault()

    const form = event.currentTarget
    const body = String(new FormData(form).get("body") ?? "").trim()

    if (body) {
      form.reset()
      mutate({ type: "send", message: newMessage(channelId, userId, body) })
    }
  }

  function edit(message: OptimisticMessage) {
    const body = window.prompt("Edit message", message.body)?.trim()

    if (body) {
      mutate({ type: "edit", id: message.id, body })
    }
  }

  if (status === "error") {
    return <p role="alert">Live updates stopped: {error.message}</p>
  }

  if (data === undefined) {
    return <p>Loading messages</p>
  }

  return (
    <section>
      <form onSubmit={send}>
        <input name="body" placeholder="Message" required />
        <button type="submit">Send</button>
      </form>

      {status === "stale" && <p role="status">Updating</p>}
      {mutationError && <p role="alert">{mutationError}</p>}

      <ul>
        {messages.map((message) => (
          <li key={message.id} aria-busy={message.pending}>
            <LocalTime date={message.created_at} />
            <p>{message.body}</p>
            {message.author_id === userId && (
              <>
                <button disabled={message.pending} onClick={() => edit(message)} type="button">
                  Edit
                </button>
                <button
                  disabled={message.pending}
                  onClick={() => mutate({ type: "delete", id: message.id })}
                  type="button"
                >
                  Delete
                </button>
              </>
            )}
          </li>
        ))}
      </ul>
    </section>
  )
}

// ...imports and useSealedQuery() as in the reactivity guide

export function Channel({ channelId, userId }: { channelId: number; userId: string }) {
  const sealed = useSealedQuery(channelId)
  // Renewals must be for the same query, so refresh with the same channel
  const refreshQuery = useCallback(() => fetchSealedMessages(channelId), [channelId])

  if (sealed?.sealedQuery) {
    return (
      <Messages
        sealedQuery={sealed.sealedQuery}
        refreshQuery={refreshQuery}
        channelId={channelId}
        userId={userId}
      />
    )
  }

  if (sealed?.error) {
    return <p role="alert">Couldn't open this channel</p>
  }

  return <p>Loading messages</p>
}

import { createCollection } from "@tanstack/react-db"
import { realtimeCollectionOptions } from "@neon/realtime-tanstack"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { deleteMessages, editMessages, fetchSealedInbox, sendMessages } from "./api"
import { client } from "./client"

export function createMessagesCollection(sealedQuery: SealedLiveQuery<Message>) {
  return createCollection(
    realtimeCollectionOptions({
      id: "messages",
      client,
      query: sealedQuery,
      refreshQuery: fetchSealedInbox,
      getKey: (message) => message.id,
      onInsert: async ({ transaction, collection }) => {
        const result = await sendMessages(transaction.mutations.map(({ modified }) => modified))

        if (result.kind === "committed") {
          await collection.utils.awaitTxId(result.txid, 10_000)
        }
      },
      onUpdate: async ({ transaction, collection }) => {
        const result = await editMessages(transaction.mutations.map(({ modified: { id, body } }) => ({ id, body })))

        if (result.kind === "committed") {
          await collection.utils.awaitTxId(result.txid, 10_000)
        }
      },
      onDelete: async ({ transaction, collection }) => {
        const result = await deleteMessages(transaction.mutations.map(({ original }) => original.id))

        if (result.kind === "committed") {
          await collection.utils.awaitTxId(result.txid, 10_000)
        }
      }
    })
  )
}

export type MessagesCollection = ReturnType<typeof createMessagesCollection>

import type { FormEvent } from "react"
import { eq, useLiveQuery as useCollectionQuery } from "@tanstack/react-db"

import type { Message } from "../backend/schema"
import type { MessagesCollection } from "./collections"
import { newMessage } from "./mutations"

interface Props {
  messagesCollection: MessagesCollection;
  channelId: number;
  userId: string;
}

export function Inbox({ messagesCollection, channelId, userId }: Props) {
  const { data, isReady, isError } = useCollectionQuery({
    query: (q) =>
      q
        .from({ message: messagesCollection })
        .where(({ message }) => eq(message.channel_id, channelId))
        .orderBy(({ message }) => message.created_at, "desc")
  })

  function send(event: FormEvent<HTMLFormElement>) {
    event.preventDefault()

    const form = event.currentTarget
    const body = String(new FormData(form).get("body") ?? "").trim()

    if (body) {
      form.reset()
      messagesCollection.insert(newMessage(channelId, userId, body))
    }
  }

  function edit(message: Message) {
    const body = window.prompt("Edit message", message.body)?.trim()

    if (body) {
      messagesCollection.update(message.id, (draft) => {
        draft.body = body
      })
    }
  }

  if (isError) {
    return <p role="alert">Couldn't sync messages</p>
  }

  if (!isReady) {
    return <p>Loading messages</p>
  }

  return (
    <section>
      <form onSubmit={send}>
        <input name="body" placeholder="Message" required />
        <button type="submit">Send</button>
      </form>

      <ul>
        {data.map((message) => (
          <li key={message.id} aria-busy={message.$hasPendingWrites}>
            {message.body}
            {message.author_id === userId && (
              <>
                <button disabled={message.$hasPendingWrites} onClick={() => edit(message)} type="button">
                  Edit
                </button>
                <button
                  disabled={message.$hasPendingWrites}
                  onClick={() => messagesCollection.delete(message.id)}
                  type="button"
                >
                  Delete
                </button>
              </>
            )}
          </li>
        ))}
      </ul>
    </section>
  )
}

import { StrictMode } from "react"
import { createRoot } from "react-dom/client"
import { RealtimeProvider } from "@neon/realtime-react"

import { fetchSealedInbox } from "./api"
import { authClient } from "./auth"
import { Channel } from "./Channel"
import { client } from "./client"
import { createMessagesCollection } from "./collections"
import { Inbox } from "./Inbox"

const root = createRoot(document.getElementById("root")!)
const { data: session } = await authClient.getSession()

if (!session) {
  root.render(<p>Sign in to see the messages.</p>)
} else {
  const userId = session.user.id
  const messagesCollection = createMessagesCollection(await fetchSealedInbox())

  root.render(
    <StrictMode>
      <RealtimeProvider client={client}>
        <Channel channelId={2} userId={userId} />
        <Inbox messagesCollection={messagesCollection} channelId={2} userId={userId} />
      </RealtimeProvider>
    </StrictMode>
  )
}