atproto/packages/lex/lex-server/examples/subscription.mjs

#! /usr/bin/env node

/* eslint-env node */
/* eslint-disable @typescript-eslint/no-namespace */
/* eslint-disable n/no-extraneous-import */

import { scheduler } from 'node:timers/promises'
import { l } from '@atproto/lex'
import { LexError, LexRouter } from '@atproto/lex-server'
import { serve, upgradeWebSocket } from '@atproto/lex-server/nodejs'

// This code would typically be generated by @atproto/lex
const nsid = 'com.example.echo'
const message = l.typedObject(
  nsid,
  'message',
  l.object({
    message: l.string(),
    cursor: l.integer({ minimum: 0 }),
  }),
)
const main = l.subscription(
  nsid,
  l.params({
    message: l.string({ minLength: 1 }),
    cursor: l.optional(l.withDefault(l.integer({ minimum: 0 }), 0)),
    limit: l.optional(
      l.withDefault(l.integer({ minimum: 1, maximum: 100 }), 10),
    ),
  }),
  l.typedUnion([l.typedRef(() => message)], false),
  ['LimitReached'],
)
const com = { example: { echo: { main, message } } }

const router = new LexRouter({
  upgradeWebSocket,
  fallback: indexHtml,
  onHandlerError: ({ error, method }) => {
    console.error(`Handler error in method ${method.nsid}:`, error)
  },
})
  //
  .add(com.example.echo, async function* ({ signal, params }) {
    const { message, cursor, limit } = params

    for (let i = 0; i < limit; i++) {
      yield com.example.echo.message.$build({
        message: message,
        cursor: cursor + i,
      })

      // Wait 1 second between messages (stop waiting if the request is aborted)
      await scheduler.wait(1_000, { signal })
    }

    throw new LexError('LimitReached', `Limit of ${limit} messages reached`)
  })

serve(router.fetch, { port: 8080 })

async function indexHtml() {
  return new Response(
    html`
      <h1>Open dev tools and look at the console</h1>
      <script type="module">
        import { decodeMultiple } from 'https://cdn.jsdelivr.net/npm/cbor-x@1.6.0/+esm'

        const host = window.location.host
        const nsid = 'com.example.echo'
        const params = new URLSearchParams(window.location.search)
        if (!params.has('message')) {
          params.set('message', 'Hello, world!')
        }

        const url = 'ws://' + host + '/xrpc/' + nsid + '?' + params.toString()

        const ws = new WebSocket(url)
        ws.binaryType = 'arraybuffer'

        ws.addEventListener('message', async (event) => {
          const bytes = new Uint8Array(event.data)
          let { length, 0: header, 1: data } = await decodeMultiple(bytes)
          if (length !== 2) {
            console.warn('Invalid message format', bytes)
          } else if (header.op === 1) {
            if (
              data &&
              typeof data === 'object' &&
              typeof header.t === 'string' &&
              !('$type' in data)
            ) {
              data.$type = header.t.startsWith('#') ? nsid + header.t : header.t
            }

            console.log('Message frame', data)
          } else if (header.op === -1) {
            console.warn('Error frame', data)
          } else {
            console.warn('Unknown message', header, data)
          }
        })

        ws.addEventListener('close', (event) => {
          console.info('Closed', {
            code: event.code,
            reason: event.reason,
            wasClean: event.wasClean,
          })
        })

        setTimeout(() => {
          ws.close()
        }, 20_000)

        // Expose for debugging
        window.ws = ws
      </script>
    `,
    {
      status: 200,
      headers: { 'content-type': 'text/html' },
    },
  )
}

/**
 * Simple HTML template tag function to enable syntax highlighting.
 * @param {TemplateStringsArray} parts
 * @param {...never} args
 * @returns {string}
 */
function html(parts, ...args) {
  if (args.length) throw new Error('No substitutions allowed in HTML template')
  return parts[0]
}