Queries · Neon Realtime · neondatabase/agent-skills · Skills Docs

Queries · Neon Realtime · neondatabase/agent-skills · Skills Docs

import { integer, pgTable, primaryKey, text, timestamp, uuid } from "drizzle-orm/pg-core"

export const messages = pgTable(
  "messages", {
    id: uuid("id").primaryKey(),
    channel_id: integer("channel_id").notNull(),
    author_id: text("author_id").notNull(),
    body: text("body").notNull(),
    created_at: timestamp("created_at", { withTimezone: true }).notNull().defaultNow()
  }
)

export const channelMembers = pgTable(
  "channel_members", {
    channel_id: integer("channel_id").notNull(),
    user_id: text("user_id").notNull()
  },
  (table) => [primaryKey({ columns: [table.channel_id, table.user_id] })]
)

export type Message = typeof messages.$inferSelect

import { drizzle } from "drizzle-orm/node-postgres"
import { Pool } from "pg"
import * as schema from "./schema"

const pool = new Pool({
  // Neon URLs say sslmode=require. Asking pg for verify-full keeps today's checks and skips its warning
  connectionString: process.env.DATABASE_URL!.replace("sslmode=require", "sslmode=verify-full")
})
export const db = drizzle(pool, { schema })

// So an idle connection error, for example when the compute scales to zero, doesn't crash the API
pool.on("error", (error: Error) => console.error(error))

import { rawSql } from "@neon/realtime/server"

import type { Message } from "./schema"

export const allMessages = rawSql<Message>(
  `SELECT author_id, body, channel_id, created_at, id FROM messages`
)

import { rawSql } from "@neon/realtime/server"

import type { Message } from "./schema"

export function messagesQuery(userId: string, channelId: number) {
  return rawSql<Message>(
    `SELECT messages.author_id, messages.body, messages.channel_id,
       messages.created_at, messages.id FROM messages
     INNER JOIN channel_members ON channel_members.channel_id = messages.channel_id
     WHERE channel_members.user_id = $1 AND messages.channel_id = $2
     ORDER BY messages.created_at DESC
     LIMIT 50`,
    [userId, channelId]
  )
}

import { pgParam, rawSql } from "@neon/realtime/server"

import type { Message } from "./schema"

export function messagesSinceQuery(userId: string, channelId: number, since: Date) {
  return rawSql<Message>(
    `SELECT messages.author_id, messages.body, messages.channel_id,
       messages.created_at, messages.id FROM messages
     INNER JOIN channel_members ON channel_members.channel_id = messages.channel_id
     WHERE channel_members.user_id = $1 AND messages.channel_id = $2 AND messages.created_at > $3
     ORDER BY messages.created_at DESC
     LIMIT 50`,
    [userId, channelId, pgParam.timestamptz(since)]
  )
}

import { createRealtime } from "@neon/realtime/server"
import { drizzleAdapter } from "@neon/realtime-drizzle"

export const realtime = createRealtime({
  secret: process.env.NEON_REALTIME_SECRET!,
  db: "neondb",
  adapter: drizzleAdapter()
})

import { and, desc, eq, getTableColumns } from "drizzle-orm"

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

export function messagesQuery(userId: string, channelId: number) {
  return db
    .select(getTableColumns(messages))
    .from(messages)
    .innerJoin(channelMembers, eq(channelMembers.channel_id, messages.channel_id))
    .where(and(eq(channelMembers.user_id, userId), eq(messages.channel_id, channelId)))
    .orderBy(desc(messages.created_at))
    .limit(50)
}

import type { Context } from "hono"
import { HTTPException } from "hono/http-exception"
import { createRemoteJWKSet, jwtVerify } from "jose"

const authUrl = process.env.NEON_AUTH_BASE_URL!
const origin = new URL(authUrl).origin
const jwks = createRemoteJWKSet(new URL(`${authUrl}/.well-known/jwks.json`))

async function verifyUserId(token: string): Promise<string | undefined> {
  try {
    const { payload } = await jwtVerify(token, jwks, { issuer: origin, audience: origin })

    return payload.sub
  } catch {
    return undefined
  }
}

export async function getUser(headers: Headers): Promise<{ id: string } | undefined> {
  const header = headers.get("Authorization")
  const token = header?.startsWith("Bearer ") ? header.slice("Bearer ".length) : undefined
  const userId = token ? await verifyUserId(token) : undefined

  return userId ? { id: userId } : undefined
}

export async function requireUser(c: Context): Promise<{ id: string }> {
  const user = await getUser(c.req.raw.headers)

  if (!user) {
    throw new HTTPException(401, { message: "Unauthorized" })
  }

  return user
}

import { Hono } from "hono"
import { validator } from "hono/validator"

import { requireUser } from "./auth"
import { messagesQuery } from "./queries"
import { realtime } from "./realtime"

const app = new Hono().post(
  "/api/v1/messages/live",
  validator("json", (value, c) => {
    const channelId: unknown = value?.channelId

    if (typeof channelId !== "number" || !Number.isInteger(channelId)) {
      return c.json({ error: "channelId must be an integer" }, 400)
    }

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

    const query = messagesQuery(user.id, channelId)
    const sealedQuery = await realtime.seal({ query })

    return c.json(sealedQuery)
  }
)

export type AppType = typeof app

export default app

import { createServerFn } from "@tanstack/react-start"

import { messagesQuery } from "./backend/queries"
import { realtime } from "./backend/realtime"
import { requireUser } from "./session-middleware"

export const sealMessages = createServerFn({ method: "POST" })
  .middleware([requireUser])
  .validator((data: { channelId: number }) => {
    if (!Number.isInteger(data.channelId)) {
      throw new Error("channelId must be an integer")
    }

    return data
  })
  .handler(({ data, context }) =>
    realtime.seal({ query: messagesQuery(context.user.id, data.channelId) })
  )