From ccb68ec37eba0edac3ada11a6ad4eb82dc9e1629 Mon Sep 17 00:00:00 2001 From: Gerald Leban Date: Sat, 10 Oct 2026 09:35:40 +0200 Subject: [PATCH] feat(jetstream): persist stream cursor in sqlite database --- .env.example | 2 +- app/models/jetstream_cursor.ts | 5 ++ ...16905415_create_jetstream_cursors_table.ts | 17 +++++ database/schema.ts | 41 ++++-------- services/jetstream_service.ts | 64 ++++++++++--------- start/env.ts | 2 +- 6 files changed, 71 insertions(+), 60 deletions(-) create mode 100644 app/models/jetstream_cursor.ts create mode 100644 database/migrations/1791616905415_create_jetstream_cursors_table.ts diff --git a/.env.example b/.env.example index 17448d6..f45ddd7 100644 --- a/.env.example +++ b/.env.example @@ -17,7 +17,7 @@ PUBLIC_URL=http://$HOST:$PORT/ # ATPROTO_OAUTH_CLIENT_ID= # Jetstream & Internal -# JETSTREAM_URL= +JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe # INTERNAL_APP_URL=http://app:3333 # Cloudflare Tunnel diff --git a/app/models/jetstream_cursor.ts b/app/models/jetstream_cursor.ts new file mode 100644 index 0000000..eabbcaf --- /dev/null +++ b/app/models/jetstream_cursor.ts @@ -0,0 +1,5 @@ +import { JetstreamCursorSchema } from '#database/schema' + +export default class JetstreamCursor extends JetstreamCursorSchema { + static readonly selfAssignPrimaryKey = true +} diff --git a/database/migrations/1791616905415_create_jetstream_cursors_table.ts b/database/migrations/1791616905415_create_jetstream_cursors_table.ts new file mode 100644 index 0000000..7d12e24 --- /dev/null +++ b/database/migrations/1791616905415_create_jetstream_cursors_table.ts @@ -0,0 +1,17 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +export default class extends BaseSchema { + protected tableName = 'jetstream_cursors' + + async up() { + this.schema.createTable(this.tableName, (table) => { + table.string('id').primary().defaultTo('default') + table.string('cursor').notNullable() + table.timestamp('updated_at').notNullable() + }) + } + + async down() { + this.schema.dropTable(this.tableName) + } +} diff --git a/database/schema.ts b/database/schema.ts index 1383439..3e691c8 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -7,6 +7,17 @@ import { BaseModel, column } from '@adonisjs/lucid/orm' import { DateTime } from 'luxon' +export class JetstreamCursorSchema extends BaseModel { + static $columns = ['cursor', 'id', 'updatedAt'] as const + $columns = JetstreamCursorSchema.$columns + @column() + declare cursor: string + @column({ isPrimary: true }) + declare id: string + @column.dateTime({ autoCreate: true, autoUpdate: true }) + declare updatedAt: DateTime +} + export class KitLikeSchema extends BaseModel { static $columns = ['createdAt', 'did', 'id', 'kitId', 'uri'] as const $columns = KitLikeSchema.$columns @@ -23,24 +34,7 @@ export class KitLikeSchema extends BaseModel { } export class KitSchema extends BaseModel { - static $columns = [ - 'acquiredAt', - 'cid', - 'createdAt', - 'did', - 'grade', - 'id', - 'indexedAt', - 'manufacturer', - 'name', - 'photoCid', - 'photos', - 'recordCreatedAt', - 'scale', - 'status', - 'updatedAt', - 'uri', - ] as const + static $columns = ['acquiredAt', 'cid', 'createdAt', 'did', 'grade', 'id', 'indexedAt', 'manufacturer', 'name', 'photoCid', 'photos', 'recordCreatedAt', 'scale', 'status', 'updatedAt', 'uri'] as const $columns = KitSchema.$columns @column.dateTime() declare acquiredAt: DateTime | null @@ -99,16 +93,7 @@ export class OauthStateSchema extends BaseModel { } export class ProfileSchema extends BaseModel { - static $columns = [ - 'avatarUrl', - 'createdAt', - 'description', - 'did', - 'displayName', - 'handle', - 'indexedAt', - 'updatedAt', - ] as const + static $columns = ['avatarUrl', 'createdAt', 'description', 'did', 'displayName', 'handle', 'indexedAt', 'updatedAt'] as const $columns = ProfileSchema.$columns @column() declare avatarUrl: string | null diff --git a/services/jetstream_service.ts b/services/jetstream_service.ts index 3966a89..43d072a 100644 --- a/services/jetstream_service.ts +++ b/services/jetstream_service.ts @@ -1,16 +1,15 @@ -import app from '@adonisjs/core/services/app' import logger from '@adonisjs/core/services/logger' import { inject } from '@adonisjs/core' import env from '#start/env' import Kit from '#kits/models/kit' +import KitLike from '#kits/models/kit_like' +import JetstreamCursor from '#models/jetstream_cursor' import Profile from '#profile/models/profile' import { DateTime } from 'luxon' import { AtUri } from '@atproto/syntax' import { lexParse, getBlobCidString, isBlobRef } from '@atproto/lex' -import KitLike from '#kits/models/kit_like' import { main as gunplaKitLexicon } from '#lexicons/build/gunpla/kit' import { main as gunplaLikeLexicon } from '#lexicons/build/gunpla/like' -import fs from 'node:fs' import WebSocket from 'ws' type JetstreamCommit = Record> = @@ -35,14 +34,12 @@ type JetstreamEvent = export interface JetstreamServiceOptions { endpoint?: string wantedCollections?: string[] - cursorFilePath?: string } @inject() export class JetstreamService { private endpoint: string private wantedCollections: string[] - private cursorFilePath: string private ws: WebSocket | null = null private isRunning = false private reconnectTimeout: NodeJS.Timeout | null = null @@ -51,35 +48,30 @@ export class JetstreamService { private lastSavedCursorTime = 0 constructor(options: JetstreamServiceOptions = {}) { - this.endpoint = - options.endpoint ?? - env.get('JETSTREAM_URL') ?? - 'wss://jetstream2.us-east.bsky.network/subscribe' + this.endpoint = options.endpoint ?? env.get('JETSTREAM_URL') this.wantedCollections = options.wantedCollections ?? ['build.gunpla.kit', 'build.gunpla.like'] - this.cursorFilePath = options.cursorFilePath ?? app.tmpPath('jetstream_cursor.txt') } - loadCursor(): string | null { + async loadCursor(): Promise { try { - if (fs.existsSync(this.cursorFilePath)) { - const cursor = fs.readFileSync(this.cursorFilePath, 'utf-8').trim() - if (cursor) return cursor - } + const record = await JetstreamCursor.find('default') + if (record?.cursor) return record.cursor } catch (err) { - logger.error({ err }, 'Failed to read Jetstream cursor file') + logger.error({ err }, 'Failed to read Jetstream cursor from database') } return null } - saveCursor(cursor: number | string, force = false): void { + async saveCursor(cursor: number | string, force = false): Promise { this.lastCursor = cursor const now = Date.now() if (!force && now - this.lastSavedCursorTime < 2000) return try { - fs.writeFileSync(this.cursorFilePath, cursor.toString(), 'utf-8') + const cursorStr = cursor.toString() + await JetstreamCursor.updateOrCreate({ id: 'default' }, { id: 'default', cursor: cursorStr }) this.lastSavedCursorTime = now } catch (err) { - logger.error({ err }, 'Failed to write Jetstream cursor file') + logger.error({ err }, 'Failed to write Jetstream cursor to database') } } @@ -143,12 +135,16 @@ export class JetstreamService { } async processEvent(event: JetstreamEvent): Promise { - this.saveCursor(event.time_us) - - if (event.kind !== 'commit') return + if (event.kind !== 'commit') { + await this.saveCursor(event.time_us) + return + } const { collection, operation, rkey } = event.commit - if (!this.wantedCollections.includes(collection)) return + if (!this.wantedCollections.includes(collection)) { + await this.saveCursor(event.time_us) + return + } const did = event.did const uri = `at://${did}/${collection}/${rkey}` @@ -158,6 +154,7 @@ export class JetstreamService { logger.info(`[Jetstream] deleting kit ${uri}`) await Kit.query().where('uri', uri).delete() await this.broadcast('kits', { type: 'delete', uri }) + await this.saveCursor(event.time_us) return } logger.info(`[Jetstream] ${operation} kit ${uri}`) @@ -168,6 +165,7 @@ export class JetstreamService { { collection: 'build.gunpla.kit', rkey, error: validation.error }, '[Jetstream] invalid kit record — skipping' ) + await this.saveCursor(event.time_us) return } await this.resolveHandle(did) @@ -200,15 +198,17 @@ export class JetstreamService { } ) await this.broadcast('kits', { type: operation, uri }) + await this.saveCursor(event.time_us) return } if (collection === 'build.gunpla.like') { if (operation === 'delete') { - logger.info('[Jetstream] deleting like ' + uri) + logger.info(`[Jetstream] deleting like ${uri}`) const existingLike = await KitLike.findBy('uri', uri) if (!existingLike) { - logger.debug('[Jetstream] like not found locally ' + uri + ' — no-op') + logger.debug(`[Jetstream] like not found locally ${uri} — no-op`) + await this.saveCursor(event.time_us) return } const kitId = existingLike.kitId @@ -222,10 +222,11 @@ export class JetstreamService { likeCount: Number(row.$extras.count), }) } + await this.saveCursor(event.time_us) return } - logger.info('[Jetstream] ' + operation + ' like ' + uri) + logger.info(`[Jetstream] ${operation} like ${uri}`) const { record } = event.commit const validation = gunplaLikeLexicon.safeValidate(record) if (!validation.success) { @@ -233,6 +234,7 @@ export class JetstreamService { { collection: 'build.gunpla.like', rkey, error: validation.error }, '[Jetstream] invalid like record — skipping' ) + await this.saveCursor(event.time_us) return } @@ -240,6 +242,7 @@ export class JetstreamService { const targetKit = await Kit.findBy('uri', subjectUri) if (!targetKit) { logger.debug({ subjectUri, did }, '[Jetstream] like targets unindexed kit — skipping') + await this.saveCursor(event.time_us) return } @@ -250,6 +253,7 @@ export class JetstreamService { uri: targetKit.uri, likeCount: Number(row.$extras.count), }) + await this.saveCursor(event.time_us) } } @@ -259,11 +263,11 @@ export class JetstreamService { this.connect() } - private connect(): void { + private async connect(): Promise { if (!this.isRunning) return this.cleanup() - const cursor = this.loadCursor() + const cursor = await this.loadCursor() const params = this.wantedCollections .map((c) => `wantedCollections=${encodeURIComponent(c)}`) .join('&') @@ -323,14 +327,14 @@ export class JetstreamService { } } - stop(): void { + async stop(): Promise { this.isRunning = false if (this.reconnectTimeout) { clearTimeout(this.reconnectTimeout) this.reconnectTimeout = null } this.cleanup() - if (this.lastCursor) this.saveCursor(this.lastCursor, true) + if (this.lastCursor) await this.saveCursor(this.lastCursor, true) logger.info('[Jetstream] stopped') } } diff --git a/start/env.ts b/start/env.ts index 399f5ff..67eb246 100644 --- a/start/env.ts +++ b/start/env.ts @@ -39,7 +39,7 @@ export default await Env.create(new URL('../', import.meta.url), { | Variables for configuring the Jetstream firehose |---------------------------------------------------------- */ - JETSTREAM_URL: Env.schema.string.optional({ format: 'url', tld: false }), + JETSTREAM_URL: Env.schema.string(), /* |---------------------------------------------------------- -- 2.54.0