add jetstream

lexicons/xyz/statusphere/getStatuses.json

···
           "type": "object",
           "required": ["statuses"],
           "properties": {
-            "cursor": { "type": "string" },
             "statuses": {
               "type": "array",
               "items": {

packages/appview/src/context.ts

···
 import { Firehose } from '@atproto/sync'
 import pino from 'pino'
 
-import { Database } from './db'
-import { BidirectionalResolver } from './id-resolver'
+import { Database } from '#/db'
+import { BidirectionalResolver } from '#/id-resolver'
+import { Jetstream } from '#/ingestors'
 
 // Application state passed to the router and elsewhere
 export type AppContext = {
   db: Database
-  ingester: Firehose
+  ingester: Firehose | Jetstream<any>
   logger: pino.Logger
   oauthClient: OAuthClient
   resolver: BidirectionalResolver

packages/appview/src/db.ts

···
   },
 }
 
+migrations['003'] = {
+  async up(db: Kysely<unknown>) {},
+  async down(_db: Kysely<unknown>) {},
+}
+
 migrations['002'] = {
   async up(db: Kysely<unknown>) {
     await db.schema
       .createTable('cursor')
       .addColumn('id', 'integer', (col) => col.primaryKey())
       .addColumn('seq', 'integer', (col) => col.notNull())
+      .execute()
+
+    // Insert initial cursor values:
+    // id=1 is for firehose, id=2 is for jetstream
+    await db
+      .insertInto('cursor' as never)
+      .values([
+        { id: 1, seq: 0 },
+        { id: 2, seq: 0 },
+      ])
       .execute()
   },
   async down(db: Kysely<unknown>) {

packages/appview/src/index.ts

···
 import { createDb, migrateToLatest } from '#/db'
 import * as error from '#/error'
 import { createBidirectionalResolver, createIdResolver } from '#/id-resolver'
-import { createIngester } from '#/ingester'
+import { createFirehoseIngester, createJetstreamIngester } from '#/ingestors'
 import { createServer } from '#/lexicons'
 import { env } from '#/lib/env'
 
···
     // Create the atproto utilities
     const oauthClient = await createClient(db)
     const baseIdResolver = createIdResolver()
-    const ingester = await createIngester(db, baseIdResolver)
+    const ingester = await createJetstreamIngester(db)
+    // Alternative: const ingester = await createFirehoseIngester(db, baseIdResolver)
     const resolver = createBidirectionalResolver(baseIdResolver)
     const ctx = {
       db,
···
         })
       }
     } else {
-      server.xrpc.router.set('trust proxy', true)
+      app.set('trust proxy', true)
     }
 
     // Use the port from env (should be 3001 for the API server)

packages/appview/src/ingester.ts packages/appview/src/ingestors/firehose.ts

···
 
 import type { Database } from '#/db'
 
-export async function createIngester(db: Database, idResolver: IdResolver) {
+export async function createFirehoseIngester(
+  db: Database,
+  idResolver: IdResolver,
+) {
   const logger = pino({ name: 'firehose ingestion' })
 
   const cursor = await db

packages/appview/src/ingestors/index.ts

···
+export * from './jetstream'
+export * from './firehose'

packages/appview/src/ingestors/jetstream.ts

···
+import { XyzStatusphereStatus } from '@statusphere/lexicon'
+import pino from 'pino'
+import WebSocket from 'ws'
+
+import type { Database } from '#/db'
+
+export async function createJetstreamIngester(db: Database) {
+  const logger = pino({ name: 'jetstream ingestion' })
+
+  const cursor = await db
+    .selectFrom('cursor')
+    .where('id', '=', 2)
+    .select('seq')
+    .executeTakeFirst()
+
+  logger.info(`start cursor: ${cursor?.seq}`)
+
+  // For throttling cursor writes
+  let lastCursorWrite = 0
+
+  return new Jetstream<XyzStatusphereStatus.Record>({
+    logger,
+    cursor: cursor?.seq || undefined,
+    setCursor: async (seq) => {
+      const now = Date.now()
+
+      if (now - lastCursorWrite >= 30000) {
+        lastCursorWrite = now
+        logger.info(`writing cursor: ${seq}`)
+        await db
+          .updateTable('cursor')
+          .set({ seq })
+          .where('id', '=', 2)
+          .execute()
+      }
+    },
+    handleEvent: async (evt) => {
+      // ignore account and identity events
+      if (
+        evt.kind !== 'commit' ||
+        evt.commit.collection !== 'xyz.statusphere.status'
+      )
+        return
+
+      const now = new Date()
+      const uri = `at://${evt.did}/${evt.commit.collection}/${evt.commit.rkey}`
+
+      if (
+        (evt.commit.operation === 'create' ||
+          evt.commit.operation === 'update') &&
+        XyzStatusphereStatus.isRecord(evt.commit.record)
+      ) {
+        const validatedRecord = XyzStatusphereStatus.validateRecord(
+          evt.commit.record,
+        )
+        if (!validatedRecord.success) return
+
+        // Store the status in our SQLite
+        await db
+          .insertInto('status')
+          .values({
+            uri,
+            authorDid: evt.did,
+            status: validatedRecord.value.status,
+            createdAt: validatedRecord.value.createdAt,
+            indexedAt: now.toISOString(),
+          })
+          .onConflict((oc) =>
+            oc.column('uri').doUpdateSet({
+              status: validatedRecord.value.status,
+              indexedAt: now.toISOString(),
+            }),
+          )
+          .execute()
+      } else if (evt.commit.operation === 'delete') {
+        // Remove the status from our SQLite
+        await db.deleteFrom('status').where('uri', '=', uri).execute()
+      }
+    },
+    onError: (err) => {
+      logger.error({ err }, 'error during jetstream ingestion')
+    },
+    wantedCollections: ['xyz.statusphere.status'],
+  })
+}
+
+export class Jetstream<T> {
+  private logger: pino.Logger
+  private handleEvent: (evt: JetstreamEvent<T>) => Promise<void>
+  private onError: (err: unknown) => void
+  private setCursor?: (seq: number) => Promise<void>
+  private cursor?: number
+  private ws?: WebSocket
+  private isStarted = false
+  private wantedCollections: string[]
+
+  constructor({
+    logger,
+    cursor,
+    setCursor,
+    handleEvent,
+    onError,
+    wantedCollections,
+  }: {
+    logger: pino.Logger
+    cursor?: number
+    setCursor?: (seq: number) => Promise<void>
+    handleEvent: (evt: any) => Promise<void>
+    onError: (err: any) => void
+    wantedCollections: string[]
+  }) {
+    this.logger = logger
+    this.cursor = cursor
+    this.setCursor = setCursor
+    this.handleEvent = handleEvent
+    this.onError = onError
+    this.wantedCollections = wantedCollections
+  }
+
+  constructUrlWithQuery = (): string => {
+    const params = new URLSearchParams()
+    params.append('wantedCollections', this.wantedCollections.join(','))
+    if (this.cursor !== undefined) {
+      params.append('cursor', this.cursor.toString())
+    }
+    return `wss://jetstream.mozzius.dev/subscribe?${params.toString()}`
+  }
+
+  start() {
+    if (this.isStarted) return
+    this.isStarted = true
+    this.ws = new WebSocket(this.constructUrlWithQuery())
+
+    this.ws.on('open', () => {
+      this.logger.info('Jetstream connection opened.')
+    })
+
+    this.ws.on('message', async (data) => {
+      try {
+        const event: JetstreamEvent<T> = JSON.parse(data.toString())
+
+        // Update cursor if provided
+        if (event.time_us !== undefined && this.setCursor) {
+          await this.setCursor(event.time_us)
+        }
+
+        await this.handleEvent(event)
+      } catch (err) {
+        this.onError(err)
+      }
+    })
+
+    this.ws.on('error', (err) => {
+      this.onError(err)
+    })
+
+    this.ws.on('close', (code, reason) => {
+      this.logger.error(`Jetstream closed. Code: ${code}, Reason: ${reason}`)
+      this.isStarted = false
+    })
+  }
+
+  destroy() {
+    if (this.ws) {
+      this.ws.close()
+      this.isStarted = false
+    }
+  }
+}
+
+type JetstreamEvent<T> = {
+  did: string
+  time_us: number
+} & (CommitEvent<T> | AccountEvent | IdentityEvent)
+
+type CommitEvent<T> = {
+  kind: 'commit'
+  commit:
+    | {
+        operation: 'create' | 'update'
+        record: T
+        rev: string
+        collection: string
+        rkey: string
+        cid: string
+      }
+    | {
+        operation: 'delete'
+        rev: string
+        collection: string
+        rkey: string
+      }
+}
+
+type IdentityEvent = {
+  kind: 'identity'
+  identity: {
+    did: string
+    handle: string
+    seq: number
+    time: string
+  }
+}
+
+type AccountEvent = {
+  kind: 'account'
+  account: {
+    active: boolean
+    did: string
+    seq: number
+    time: string
+  }
+}

packages/appview/src/lexicons/lexicons.ts

···
             type: 'object',
             required: ['statuses'],
             properties: {
-              cursor: {
-                type: 'string',
-              },
               statuses: {
                 type: 'array',
                 items: {

packages/appview/src/lexicons/types/xyz/statusphere/getStatuses.ts

···
 export type InputSchema = undefined
 
 export interface OutputSchema {
-  cursor?: string
   statuses: XyzStatusphereDefs.StatusView[]
 }
 

packages/client/src/components/StatusForm.tsx

···
 import useAuth from '#/hooks/useAuth'
 import api from '#/services/api'
 
-const STATUS_OPTIONS = [
+export const STATUS_OPTIONS = [
   '👍',
   '👎',
   '💙',

packages/client/src/components/StatusList.tsx

···
 import { useQuery } from '@tanstack/react-query'
 
 import api from '#/services/api'
+import { STATUS_OPTIONS } from './StatusForm'
 
 const StatusList = () => {
   // Use React Query to fetch and cache statuses
···
 
   // Destructure data
   const statuses = data?.statuses || []
+  
+  // Get a random emoji from the STATUS_OPTIONS array
+  const randomEmoji = STATUS_OPTIONS[Math.floor(Math.random() * STATUS_OPTIONS.length)]
 
   if (isPending && !data) {
     return (
-      <div className="py-4 text-center text-gray-500 dark:text-gray-400">
-        Loading statuses...
+      <div className="py-8 text-center">
+        <div className="text-5xl mb-2 animate-pulse inline-block">
+          {randomEmoji}
+        </div>
+        <div className="text-gray-500 dark:text-gray-400">
+          Loading statuses...
+        </div>
       </div>
     )
   }

packages/client/src/pages/HomePage.tsx

···
 import Header from '#/components/Header'
-import StatusForm from '#/components/StatusForm'
+import StatusForm, { STATUS_OPTIONS } from '#/components/StatusForm'
 import StatusList from '#/components/StatusList'
 import { useAuth } from '#/hooks/useAuth'
 
 const HomePage = () => {
   const { user, loading, error } = useAuth()
 
+  // Get a random emoji from the STATUS_OPTIONS array
+  const randomEmoji =
+    STATUS_OPTIONS[Math.floor(Math.random() * STATUS_OPTIONS.length)]
+
   if (loading) {
     return (
-      <div className="flex justify-center items-center py-16">
-        <div className="text-center p-6">
-          <h2 className="text-2xl font-semibold mb-2 text-gray-800 dark:text-gray-200">
-            Loading Statusphere...
-          </h2>
-          <p className="text-gray-600 dark:text-gray-400">
-            Setting up your experience
-          </p>
-        </div>
+      <div className="flex justify-center items-center h-[80vh]">
+        <div className="text-9xl animate-pulse">{randomEmoji}</div>
       </div>
     )
   }

packages/client/vite.config.ts

···
       '^/(xrpc|oauth|client-metadata\.json)/.*': {
         target: 'http://localhost:3001',
         changeOrigin: true,
+        configure: (proxy, _options) => {
+          proxy.on('error', (err, _req, _res) => {
+            console.log('PROXY ERROR', err);
+          });
+          proxy.on('proxyReq', (proxyReq, req, _res) => {
+            console.log('PROXY REQUEST', req.method, req.url);
+          });
+          proxy.on('proxyRes', (proxyRes, req, _res) => {
+            console.log('PROXY RESPONSE', req.method, req.url, proxyRes.statusCode);
+          });
+        },
       },
     },
   },

packages/lexicon/src/lexicons.ts

···
             type: 'object',
             required: ['statuses'],
             properties: {
-              cursor: {
-                type: 'string',
-              },
               statuses: {
                 type: 'array',
                 items: {

packages/lexicon/src/types/xyz/statusphere/getStatuses.ts

···
 export type InputSchema = undefined
 
 export interface OutputSchema {
-  cursor?: string
   statuses: XyzStatusphereDefs.StatusView[]
 }