+139

packages/servers/labeler/README.md

reviewed

··· 1 1 + # @atcute/labeler 2 2 + 3 3 + sign and emit AT Protocol labels. 4 4 + 5 5 + ```sh 6 6 + npm install @atcute/labeler 7 7 + ``` 8 8 + 9 9 + provides the core logic for running an AT Protocol labeler like label signing, a subscription outbox 10 10 + and XRPC operation handlers. 11 11 + 12 12 + ## usage 13 13 + 14 14 + ### setting up a labeler 15 15 + 16 16 + create a `Labeler` with your signing key and storage backend, then register its routes on a router: 17 17 + 18 18 + ```ts 19 19 + import { Secp256k1PrivateKey } from '@atcute/crypto'; 20 20 + 21 21 + import { XRPCRouter } from '@atcute/xrpc-server'; 22 22 + import { createBunWebSocket } from '@atcute/xrpc-server-bun'; 23 23 + 24 24 + import { Labeler } from '@atcute/labeler'; 25 25 + 26 26 + const ws = createBunWebSocket(); 27 27 + const router = new XRPCRouter({ websocket: ws.adapter }); 28 28 + 29 29 + const labeler = new Labeler({ 30 30 + did: 'did:plc:mylabeler', 31 31 + key: await Secp256k1PrivateKey.importRaw(keyBytes), 32 32 + store: myLabelStore, // implements LabelStore 33 33 + }); 34 34 + 35 35 + labeler.register(router); 36 36 + 37 37 + export default ws.wrap(router); 38 38 + ``` 39 39 + 40 40 + `register()` adds two endpoints to the router: 41 41 + 42 42 + - **`com.atproto.label.queryLabels`** — query stored labels by URI pattern, source, and cursor 43 43 + - **`com.atproto.label.subscribeLabels`** — WebSocket subscription with backfill and live tailing 44 44 + 45 45 + ### creating labels 46 46 + 47 47 + ```ts 48 48 + // create a single label 49 49 + const saved = await labeler.createLabel({ 50 50 + uri: 'did:plc:targetuser', 51 51 + val: 'spam', 52 52 + }); 53 53 + 54 54 + // create and negate labels for a subject 55 55 + const labels = await labeler.createLabels( 56 56 + { uri: 'at://did:plc:user/app.bsky.feed.post/abc123', cid: 'bafyrei...' }, 57 57 + { create: ['nsfw'], negate: ['misleading'] }, 58 58 + ); 59 59 + ``` 60 60 + 61 61 + created labels are signed, saved to the store, and pushed to all active WebSocket subscribers. 62 62 + 63 63 + ### emitEvent endpoint 64 64 + 65 65 + to accept label creation requests over XRPC (via `tools.ozone.moderation.emitEvent`), provide an 66 66 + `auth` callback: 67 67 + 68 68 + ```ts 69 69 + import { AuthRequiredError } from '@atcute/xrpc-server'; 70 70 + import { ServiceJwtVerifier } from '@atcute/xrpc-server/auth'; 71 71 + import { 72 72 + CompositeDidDocumentResolver, 73 73 + PlcDidDocumentResolver, 74 74 + WebDidDocumentResolver, 75 75 + } from '@atcute/identity-resolver'; 76 76 + 77 77 + const jwtVerifier = new ServiceJwtVerifier({ 78 78 + serviceDid: 'did:plc:mylabeler', 79 79 + resolver: new CompositeDidDocumentResolver({ 80 80 + methods: { 81 81 + plc: new PlcDidDocumentResolver(), 82 82 + web: new WebDidDocumentResolver(), 83 83 + }, 84 84 + }), 85 85 + }); 86 86 + 87 87 + const labeler = new Labeler({ 88 88 + did: 'did:plc:mylabeler', 89 89 + key: await Secp256k1PrivateKey.importRaw(keyBytes), 90 90 + store: myLabelStore, 91 91 + auth: async (request) => { 92 92 + const authHeader = request.headers.get('authorization'); 93 93 + if (!authHeader?.startsWith('Bearer ')) { 94 94 + return false; 95 95 + } 96 96 + 97 97 + const result = await jwtVerifier.verify(authHeader.slice(7), { 98 98 + lxm: 'tools.ozone.moderation.emitEvent', 99 99 + }); 100 100 + 101 101 + return result.ok; 102 102 + }, 103 103 + }); 104 104 + ``` 105 105 + 106 106 + when `auth` is provided, the `emitEvent` endpoint is registered alongside the other routes. the 107 107 + callback receives the incoming `Request` and should return `true` to allow or `false` to reject. 108 108 + 109 109 + ### storage backend 110 110 + 111 111 + implement the `LabelStore` interface to plug in any database: 112 112 + 113 113 + ```ts 114 114 + import type { LabelStore } from '@atcute/labeler'; 115 115 + 116 116 + const store: LabelStore = { 117 117 + async save(label) { 118 118 + // insert label, assign and return a monotonic sequence number 119 119 + // label has: src, uri, cid?, val, neg, cts, exp?, sig 120 120 + }, 121 121 + 122 122 + async query({ uriPatterns, sources, cursor, limit }) { 123 123 + // query labels matching URI patterns and sources 124 124 + // return { labels: [...], cursor: "..." } 125 125 + }, 126 126 + 127 127 + async getLatestSeq() { 128 128 + // return the highest sequence number, or 0 129 129 + }, 130 130 + 131 131 + async getRange(after, limit) { 132 132 + // return labels with seq > after, ordered by seq 133 133 + }, 134 134 + }; 135 135 + ``` 136 136 + 137 137 + `save()` receives a signed label (without `seq`) and must assign a monotonically increasing sequence 138 138 + number. `query()` returns formatted labels for the `queryLabels` endpoint. `getRange()` and 139 139 + `getLatestSeq()` are used by the subscription outbox for backfill and cursor validation.

+103

packages/servers/labeler/lib/async-buffer.ts

reviewed

··· 1 1 + import type { SimpleEventEmitter } from '@mary-ext/simple-event-emitter'; 2 2 + 3 3 + import { Queue } from './queue.ts'; 4 4 + 5 5 + /** thrown when the async buffer exceeds its maximum size. */ 6 6 + export class AsyncBufferFullError extends Error { 7 7 + constructor(maxSize: number) { 8 8 + super(`reached max buffer size: ${maxSize}`); 9 9 + } 10 10 + } 11 11 + 12 12 + /** bounded async buffer with backpressure support. */ 13 13 + export class AsyncBuffer<T> { 14 14 + #queue = new Queue<T>(); 15 15 + #closed = false; 16 16 + #deferred = Promise.withResolvers<void>(); 17 17 + 18 18 + #maxSize: number; 19 19 + 20 20 + constructor(maxSize: number) { 21 21 + this.#maxSize = maxSize; 22 22 + } 23 23 + 24 24 + push(value: T): void { 25 25 + if (this.#closed) { 26 26 + return; 27 27 + } 28 28 + 29 29 + if (this.#queue.size >= this.#maxSize) { 30 30 + this.#closed = true; 31 31 + } 32 32 + 33 33 + this.#queue.enqueue(value); 34 34 + this.#deferred.resolve(); 35 35 + } 36 36 + 37 37 + close(): void { 38 38 + if (this.#closed) { 39 39 + return; 40 40 + } 41 41 + 42 42 + this.#closed = true; 43 43 + this.#deferred.resolve(); 44 44 + } 45 45 + 46 46 + async *events(): AsyncGenerator<T> { 47 47 + while (true) { 48 48 + await this.#deferred.promise; 49 49 + 50 50 + if (this.#queue.size > this.#maxSize) { 51 51 + throw new AsyncBufferFullError(this.#maxSize); 52 52 + } 53 53 + 54 54 + const value = this.#queue.dequeue(); 55 55 + if (value !== undefined) { 56 56 + yield value; 57 57 + } else if (this.#closed) { 58 58 + return; 59 59 + } else { 60 60 + this.#deferred = Promise.withResolvers(); 61 61 + } 62 62 + } 63 63 + } 64 64 + } 65 65 + 66 66 + export interface OnOptions { 67 67 + maxSize: number; 68 68 + signal?: AbortSignal; 69 69 + } 70 70 + 71 71 + /** 72 72 + * create an async iterator from a SimpleEventEmitter with bounded buffering. 73 73 + * @param emitter event emitter 74 74 + * @param options buffer and signal options 75 75 + * @returns async iterator of event payloads 76 76 + */ 77 77 + export const on = <T>(emitter: SimpleEventEmitter<[T]>, options: OnOptions): AsyncIterableIterator<T> => { 78 78 + const { maxSize, signal } = options; 79 79 + 80 80 + signal?.throwIfAborted(); 81 81 + 82 82 + const buffer = new AsyncBuffer<T>(maxSize); 83 83 + const unsubscribe = emitter.subscribe((value) => { 84 84 + buffer.push(value); 85 85 + }); 86 86 + 87 87 + const cleanup = () => { 88 88 + unsubscribe(); 89 89 + buffer.close(); 90 90 + 91 91 + signal?.removeEventListener('abort', cleanup); 92 92 + }; 93 93 + 94 94 + signal?.addEventListener('abort', cleanup, { once: true }); 95 95 + 96 96 + return (async function* () { 97 97 + try { 98 98 + yield* buffer.events(); 99 99 + } finally { 100 100 + cleanup(); 101 101 + } 102 102 + })(); 103 103 + };

+4

packages/servers/labeler/lib/index.ts

reviewed

··· 1 1 + export { formatLabel, signLabel, type CreateLabelData, type LabelSubject } from './labels.ts'; 2 2 + export { LabelOutbox, type LabelOutboxOptions } from './outbox.ts'; 3 3 + export { Labeler, type AuthCheck, type LabelerOptions } from './labeler.ts'; 4 4 + export type { LabelQueryParams, LabelQueryResult, LabelStore, SavedLabel } from './store.ts';

+257

packages/servers/labeler/lib/labeler.test.ts

reviewed

··· 1 1 + import type { ComAtprotoLabelDefs } from '@atcute/atproto'; 2 2 + import { Secp256k1PrivateKey } from '@atcute/crypto'; 3 3 + import type { Did } from '@atcute/lexicons'; 4 4 + import { XRPCRouter } from '@atcute/xrpc-server'; 5 5 + 6 6 + import { describe, expect, it } from 'vitest'; 7 7 + 8 8 + import { Labeler, type LabelerOptions } from './labeler.ts'; 9 9 + import { formatLabel } from './labels.ts'; 10 10 + import type { LabelStore, SavedLabel } from './store.ts'; 11 11 + 12 12 + const TEST_DID = 'did:plc:testlabeler' as Did; 13 13 + 14 14 + const getTestKey = async () => { 15 15 + const keyBytes = new Uint8Array(32); 16 16 + keyBytes[0] = 1; 17 17 + return Secp256k1PrivateKey.importRaw(keyBytes); 18 18 + }; 19 19 + 20 20 + const createMockStore = (): LabelStore & { labels: SavedLabel[] } => { 21 21 + const labels: SavedLabel[] = []; 22 22 + 23 23 + return { 24 24 + labels, 25 25 + async save(label) { 26 26 + const seq = labels.length + 1; 27 27 + const saved = { ...label, seq }; 28 28 + labels.push(saved); 29 29 + return saved; 30 30 + }, 31 31 + async query({ uriPatterns, sources, cursor, limit }) { 32 32 + let result = [...labels]; 33 33 + 34 34 + // filter by URI patterns 35 35 + if (uriPatterns.length > 0 && !uriPatterns.includes('*')) { 36 36 + result = result.filter((l) => 37 37 + uriPatterns.some((pattern) => { 38 38 + if (pattern.endsWith('*')) { 39 39 + return l.uri.startsWith(pattern.slice(0, -1)); 40 40 + } 41 41 + return l.uri === pattern; 42 42 + }), 43 43 + ); 44 44 + } 45 45 + 46 46 + // filter by sources 47 47 + if (sources.length > 0) { 48 48 + result = result.filter((l) => sources.includes(l.src)); 49 49 + } 50 50 + 51 51 + // filter by cursor 52 52 + if (cursor > 0) { 53 53 + result = result.filter((l) => l.seq > cursor); 54 54 + } 55 55 + 56 56 + // apply limit 57 57 + result = result.slice(0, limit); 58 58 + 59 59 + const formatted: ComAtprotoLabelDefs.Label[] = result.map(formatLabel); 60 60 + const lastSeq = result.at(-1)?.seq ?? 0; 61 61 + 62 62 + return { labels: formatted, cursor: String(lastSeq) }; 63 63 + }, 64 64 + async getLatestSeq() { 65 65 + return labels.at(-1)?.seq ?? 0; 66 66 + }, 67 67 + async getRange(after, limit) { 68 68 + const result = labels.filter((l) => l.seq > after); 69 69 + return limit !== undefined ? result.slice(0, limit) : result; 70 70 + }, 71 71 + }; 72 72 + }; 73 73 + 74 74 + const createMockWebSocket = () => ({ 75 75 + async upgrade() { 76 76 + return undefined; 77 77 + }, 78 78 + }); 79 79 + 80 80 + const createTestRouter = async (options?: Partial<LabelerOptions>) => { 81 81 + const key = await getTestKey(); 82 82 + const store = createMockStore(); 83 83 + 84 84 + const labeler = new Labeler({ 85 85 + did: TEST_DID, 86 86 + key, 87 87 + store, 88 88 + ...options, 89 89 + }); 90 90 + 91 91 + const router = new XRPCRouter({ websocket: createMockWebSocket() }); 92 92 + labeler.register(router); 93 93 + 94 94 + return { labeler, store, router }; 95 95 + }; 96 96 + 97 97 + describe('Labeler', () => { 98 98 + describe('createLabel', () => { 99 99 + it('should create and save a label', async () => { 100 100 + const { labeler, store } = await createTestRouter(); 101 101 + 102 102 + const saved = await labeler.createLabel({ 103 103 + uri: 'did:plc:target', 104 104 + val: 'spam', 105 105 + }); 106 106 + 107 107 + expect(saved.seq).toBe(1); 108 108 + expect(saved.src).toBe(TEST_DID); 109 109 + expect(saved.uri).toBe('did:plc:target'); 110 110 + expect(saved.val).toBe('spam'); 111 111 + expect(saved.neg).toBe(false); 112 112 + expect(saved.sig.byteLength).toBe(64); 113 113 + expect(store.labels).toHaveLength(1); 114 114 + }); 115 115 + 116 116 + it('should create labels with custom src', async () => { 117 117 + const { labeler } = await createTestRouter(); 118 118 + 119 119 + const saved = await labeler.createLabel({ 120 120 + uri: 'did:plc:target', 121 121 + val: 'spam', 122 122 + src: 'did:plc:custom', 123 123 + }); 124 124 + 125 125 + expect(saved.src).toBe('did:plc:custom'); 126 126 + }); 127 127 + }); 128 128 + 129 129 + describe('createLabels', () => { 130 130 + it('should create and negate labels for a subject', async () => { 131 131 + const { labeler, store } = await createTestRouter(); 132 132 + 133 133 + const results = await labeler.createLabels( 134 134 + { uri: 'did:plc:target' }, 135 135 + { create: ['spam', 'nsfw'], negate: ['misleading'] }, 136 136 + ); 137 137 + 138 138 + expect(results).toHaveLength(3); 139 139 + expect(results[0]!.val).toBe('spam'); 140 140 + expect(results[0]!.neg).toBe(false); 141 141 + expect(results[1]!.val).toBe('nsfw'); 142 142 + expect(results[1]!.neg).toBe(false); 143 143 + expect(results[2]!.val).toBe('misleading'); 144 144 + expect(results[2]!.neg).toBe(true); 145 145 + expect(store.labels).toHaveLength(3); 146 146 + }); 147 147 + }); 148 148 + 149 149 + describe('queryLabels endpoint', () => { 150 150 + it('should return labels matching URI patterns', async () => { 151 151 + const { labeler, router } = await createTestRouter(); 152 152 + 153 153 + await labeler.createLabel({ uri: 'did:plc:user1', val: 'spam' }); 154 154 + await labeler.createLabel({ uri: 'did:plc:user2', val: 'nsfw' }); 155 155 + 156 156 + const response = await router.fetch( 157 157 + new Request('http://localhost/xrpc/com.atproto.label.queryLabels?uriPatterns=*'), 158 158 + ); 159 159 + 160 160 + expect(response.status).toBe(200); 161 161 + 162 162 + const body = await response.json(); 163 163 + expect(body.labels).toHaveLength(2); 164 164 + expect(body.cursor).toBeDefined(); 165 165 + }); 166 166 + 167 167 + it('should paginate with cursor', async () => { 168 168 + const { labeler, router } = await createTestRouter(); 169 169 + 170 170 + await labeler.createLabel({ uri: 'did:plc:user1', val: 'spam' }); 171 171 + await labeler.createLabel({ uri: 'did:plc:user2', val: 'nsfw' }); 172 172 + 173 173 + const response = await router.fetch( 174 174 + new Request('http://localhost/xrpc/com.atproto.label.queryLabels?uriPatterns=*&limit=1'), 175 175 + ); 176 176 + 177 177 + const body = await response.json(); 178 178 + expect(body.labels).toHaveLength(1); 179 179 + expect(body.labels[0].val).toBe('spam'); 180 180 + 181 181 + // fetch next page 182 182 + const response2 = await router.fetch( 183 183 + new Request( 184 184 + `http://localhost/xrpc/com.atproto.label.queryLabels?uriPatterns=*&limit=1&cursor=${body.cursor}`, 185 185 + ), 186 186 + ); 187 187 + 188 188 + const body2 = await response2.json(); 189 189 + expect(body2.labels).toHaveLength(1); 190 190 + expect(body2.labels[0].val).toBe('nsfw'); 191 191 + }); 192 192 + }); 193 193 + 194 194 + describe('emitEvent endpoint', () => { 195 195 + it('should not register without auth', async () => { 196 196 + const { router } = await createTestRouter(); 197 197 + 198 198 + const response = await router.fetch( 199 199 + new Request('http://localhost/xrpc/tools.ozone.moderation.emitEvent', { 200 200 + method: 'POST', 201 201 + headers: { 'content-type': 'application/json' }, 202 202 + body: JSON.stringify({ 203 203 + event: { 204 204 + $type: 'tools.ozone.moderation.defs#modEventLabel', 205 205 + createLabelVals: ['spam'], 206 206 + negateLabelVals: [], 207 207 + }, 208 208 + subject: { 209 209 + $type: 'com.atproto.admin.defs#repoRef', 210 210 + did: 'did:plc:target', 211 211 + }, 212 212 + createdBy: TEST_DID, 213 213 + }), 214 214 + }), 215 215 + ); 216 216 + 217 217 + expect(response.status).toBe(404); 218 218 + }); 219 219 + 220 220 + it('should reject when auth returns false', async () => { 221 221 + const { router } = await createTestRouter({ 222 222 + auth: () => false, 223 223 + }); 224 224 + 225 225 + const response = await router.fetch( 226 226 + new Request('http://localhost/xrpc/tools.ozone.moderation.emitEvent', { 227 227 + method: 'POST', 228 228 + headers: { 'content-type': 'application/json' }, 229 229 + body: JSON.stringify({ 230 230 + event: { 231 231 + $type: 'tools.ozone.moderation.defs#modEventLabel', 232 232 + createLabelVals: ['spam'], 233 233 + negateLabelVals: [], 234 234 + }, 235 235 + subject: { 236 236 + $type: 'com.atproto.admin.defs#repoRef', 237 237 + did: 'did:plc:target', 238 238 + }, 239 239 + createdBy: TEST_DID, 240 240 + }), 241 241 + }), 242 242 + ); 243 243 + 244 244 + expect(response.status).toBe(401); 245 245 + }); 246 246 + }); 247 247 + 248 248 + describe('register', () => { 249 249 + it('should return 404 for unknown routes', async () => { 250 250 + const { router } = await createTestRouter(); 251 251 + 252 252 + const response = await router.fetch(new Request('http://localhost/xrpc/com.atproto.nonexistent')); 253 253 + 254 254 + expect(response.status).toBe(404); 255 255 + }); 256 256 + }); 257 257 + });

+221

packages/servers/labeler/lib/labeler.ts

reviewed

··· 1 1 + import { ComAtprotoLabelQueryLabels, ComAtprotoLabelSubscribeLabels } from '@atcute/atproto'; 2 2 + import type { PrivateKey } from '@atcute/crypto'; 3 3 + import type { Did } from '@atcute/lexicons'; 4 4 + import { ToolsOzoneModerationEmitEvent } from '@atcute/ozone'; 5 5 + import { AuthRequiredError, InvalidRequestError, type XRPCRouter, json } from '@atcute/xrpc-server'; 6 6 + 7 7 + import { SimpleEventEmitter } from '@mary-ext/simple-event-emitter'; 8 8 + 9 9 + import { formatLabel, signLabel, type CreateLabelData, type LabelSubject } from './labels.ts'; 10 10 + import { LabelOutbox } from './outbox.ts'; 11 11 + import type { LabelStore, SavedLabel } from './store.ts'; 12 12 + 13 13 + type Promisable<T> = T | Promise<T>; 14 14 + 15 15 + /** 16 16 + * auth callback for the `emitEvent` endpoint. 17 17 + * receives the request; return `false` to reject. 18 18 + */ 19 19 + export type AuthCheck = (request: Request) => Promisable<boolean>; 20 20 + 21 21 + export interface LabelerOptions { 22 22 + /** DID of the labeler account */ 23 23 + did: Did; 24 24 + /** private signing key */ 25 25 + key: PrivateKey; 26 26 + /** label storage backend */ 27 27 + store: LabelStore; 28 28 + /** authenticate `emitEvent` requests. if not provided, the endpoint is not registered */ 29 29 + auth?: AuthCheck; 30 30 + /** maximum outbox buffer size per subscription */ 31 31 + maxBufferSize?: number; 32 32 + } 33 33 + 34 34 + /** labeler server core. */ 35 35 + export class Labeler { 36 36 + /** labeler DID */ 37 37 + readonly did: Did; 38 38 + 39 39 + #store: LabelStore; 40 40 + #key: PrivateKey; 41 41 + 42 42 + #auth: AuthCheck | undefined; 43 43 + 44 44 + #emitter = new SimpleEventEmitter<[SavedLabel]>(); 45 45 + #maxBufferSize: number; 46 46 + 47 47 + constructor(options: LabelerOptions) { 48 48 + this.did = options.did; 49 49 + 50 50 + this.#store = options.store; 51 51 + this.#key = options.key; 52 52 + 53 53 + this.#auth = options.auth; 54 54 + 55 55 + this.#maxBufferSize = options.maxBufferSize ?? 500; 56 56 + } 57 57 + 58 58 + /** 59 59 + * register labeler routes on a router. 60 60 + * registers queryLabels and subscribeLabels always; 61 61 + * emitEvent is only registered if `auth` was provided. 62 62 + * @param router the router to register on 63 63 + */ 64 64 + register(router: XRPCRouter): void { 65 65 + this.#registerQueryLabels(router); 66 66 + this.#registerSubscribeLabels(router); 67 67 + 68 68 + if (this.#auth !== undefined) { 69 69 + this.#registerEmitEvent(router, this.#auth); 70 70 + } 71 71 + } 72 72 + 73 73 + /** 74 74 + * create and save a single label. 75 75 + * @param data label creation data 76 76 + * @returns the saved label 77 77 + */ 78 78 + async createLabel(data: CreateLabelData): Promise<SavedLabel> { 79 79 + const signed = await signLabel(data, this.did, this.#key); 80 80 + const saved = await this.#store.save(signed); 81 81 + 82 82 + this.#emitter.emit(saved); 83 83 + return saved; 84 84 + } 85 85 + 86 86 + /** 87 87 + * create and save multiple labels for a subject. 88 88 + * @param subject the label subject (URI + optional CID) 89 89 + * @param labels label values to create and/or negate 90 90 + * @returns all created labels 91 91 + */ 92 92 + async createLabels( 93 93 + subject: LabelSubject, 94 94 + labels: { create?: string[]; negate?: string[]; exp?: string }, 95 95 + ): Promise<SavedLabel[]> { 96 96 + const result: SavedLabel[] = []; 97 97 + 98 98 + if (labels.create) { 99 99 + for (const val of labels.create) { 100 100 + const saved = await this.createLabel({ ...subject, val, exp: labels.exp }); 101 101 + result.push(saved); 102 102 + } 103 103 + } 104 104 + 105 105 + if (labels.negate) { 106 106 + for (const val of labels.negate) { 107 107 + const saved = await this.createLabel({ ...subject, val, neg: true }); 108 108 + result.push(saved); 109 109 + } 110 110 + } 111 111 + 112 112 + return result; 113 113 + } 114 114 + 115 115 + #registerQueryLabels(router: XRPCRouter): void { 116 116 + const store = this.#store; 117 117 + 118 118 + router.addQuery(ComAtprotoLabelQueryLabels, { 119 119 + handler: async ({ params }) => { 120 120 + const result = await store.query({ 121 121 + uriPatterns: params.uriPatterns, 122 122 + sources: params.sources ?? [], 123 123 + cursor: params.cursor !== undefined ? parseInt(params.cursor, 10) || 0 : 0, 124 124 + limit: params.limit, 125 125 + }); 126 126 + 127 127 + return json(result); 128 128 + }, 129 129 + }); 130 130 + } 131 131 + 132 132 + #registerSubscribeLabels(router: XRPCRouter): void { 133 133 + const store = this.#store; 134 134 + const emitter = this.#emitter; 135 135 + const maxBufferSize = this.#maxBufferSize; 136 136 + 137 137 + router.addSubscription(ComAtprotoLabelSubscribeLabels, { 138 138 + async *handler({ params, signal }) { 139 139 + const { cursor } = params; 140 140 + 141 141 + if (cursor !== undefined) { 142 142 + const latestSeq = await store.getLatestSeq(); 143 143 + if (cursor > latestSeq) { 144 144 + throw new InvalidRequestError({ 145 145 + error: 'FutureCursor', 146 146 + description: `cursor is in the future`, 147 147 + }); 148 148 + } 149 149 + } 150 150 + 151 151 + const outbox = new LabelOutbox(store, emitter, { maxBufferSize }); 152 152 + 153 153 + for await (const label of outbox.events(cursor, signal)) { 154 154 + yield { 155 155 + $type: 'com.atproto.label.subscribeLabels#labels', 156 156 + seq: label.seq, 157 157 + labels: [formatLabel(label)], 158 158 + }; 159 159 + } 160 160 + }, 161 161 + }); 162 162 + } 163 163 + 164 164 + #registerEmitEvent(router: XRPCRouter, auth: AuthCheck): void { 165 165 + router.addProcedure(ToolsOzoneModerationEmitEvent, { 166 166 + handler: async ({ request, input }) => { 167 167 + if (!(await auth(request))) { 168 168 + throw new AuthRequiredError({ description: `unauthorized` }); 169 169 + } 170 170 + 171 171 + const { event, subject, subjectBlobCids = [], createdBy } = input; 172 172 + 173 173 + if (event.$type !== 'tools.ozone.moderation.defs#modEventLabel') { 174 174 + throw new InvalidRequestError({ description: `unsupported event type` }); 175 175 + } 176 176 + 177 177 + if (!event.createLabelVals?.length && !event.negateLabelVals?.length) { 178 178 + throw new InvalidRequestError({ description: `must provide at least one label value` }); 179 179 + } 180 180 + 181 181 + const uri = 182 182 + subject.$type === 'com.atproto.admin.defs#repoRef' 183 183 + ? subject.did 184 184 + : subject.$type === 'com.atproto.repo.strongRef' 185 185 + ? subject.uri 186 186 + : undefined; 187 187 + 188 188 + if (uri === undefined) { 189 189 + throw new InvalidRequestError({ description: `invalid subject` }); 190 190 + } 191 191 + 192 192 + const cid = subject.$type === 'com.atproto.repo.strongRef' ? subject.cid : undefined; 193 193 + 194 194 + const labelSubject: LabelSubject = { uri }; 195 195 + if (cid !== undefined) { 196 196 + labelSubject.cid = cid; 197 197 + } 198 198 + 199 199 + let exp: string | undefined; 200 200 + if (event.durationInHours !== undefined) { 201 201 + exp = new Date(Date.now() + event.durationInHours * 60 * 60 * 1000).toISOString(); 202 202 + } 203 203 + 204 204 + const labels = await this.createLabels(labelSubject, { 205 205 + create: event.createLabelVals, 206 206 + negate: event.negateLabelVals, 207 207 + exp, 208 208 + }); 209 209 + 210 210 + return json({ 211 211 + id: labels[0]!.seq, 212 212 + event, 213 213 + subject, 214 214 + subjectBlobCids, 215 215 + createdBy, 216 216 + createdAt: new Date().toISOString(), 217 217 + }); 218 218 + }, 219 219 + }); 220 220 + } 221 221 + }

+145

packages/servers/labeler/lib/labels.test.ts

reviewed

··· 1 1 + import { fromBytes, isBytes } from '@atcute/cbor'; 2 2 + import { Secp256k1PrivateKey } from '@atcute/crypto'; 3 3 + import type { Did } from '@atcute/lexicons'; 4 4 + 5 5 + import { describe, expect, it } from 'vitest'; 6 6 + 7 7 + import { formatLabel, signLabel } from './labels.ts'; 8 8 + import type { SavedLabel } from './store.ts'; 9 9 + 10 10 + const TEST_DID = 'did:plc:test1234' as Did; 11 11 + 12 12 + const getTestKey = async () => { 13 13 + // deterministic test key (32 bytes) 14 14 + const keyBytes = new Uint8Array(32); 15 15 + keyBytes[0] = 1; 16 16 + return Secp256k1PrivateKey.importRaw(keyBytes); 17 17 + }; 18 18 + 19 19 + describe('signLabel', () => { 20 20 + it('should produce a 64-byte compact secp256k1 signature', async () => { 21 21 + const key = await getTestKey(); 22 22 + 23 23 + const signed = await signLabel({ uri: 'did:plc:target', val: 'spam' }, TEST_DID, key); 24 24 + 25 25 + expect(signed.sig).toBeInstanceOf(Uint8Array); 26 26 + expect(signed.sig.byteLength).toBe(64); 27 27 + expect(signed.src).toBe(TEST_DID); 28 28 + expect(signed.uri).toBe('did:plc:target'); 29 29 + expect(signed.val).toBe('spam'); 30 30 + expect(signed.neg).toBe(false); 31 31 + expect(signed.cts).toBeDefined(); 32 32 + }); 33 33 + 34 34 + it('should use provided cts and src', async () => { 35 35 + const key = await getTestKey(); 36 36 + const cts = '2025-01-01T00:00:00.000Z'; 37 37 + 38 38 + const signed = await signLabel( 39 39 + { uri: 'did:plc:target', val: 'spam', cts, src: 'did:plc:other' }, 40 40 + TEST_DID, 41 41 + key, 42 42 + ); 43 43 + 44 44 + expect(signed.cts).toBe(cts); 45 45 + expect(signed.src).toBe('did:plc:other'); 46 46 + }); 47 47 + 48 48 + it('should produce consistent label fields for the same input', async () => { 49 49 + const key = await getTestKey(); 50 50 + const cts = '2025-01-01T00:00:00.000Z'; 51 51 + const data = { uri: 'did:plc:target', val: 'spam', cts }; 52 52 + 53 53 + const signed1 = await signLabel(data, TEST_DID, key); 54 54 + const signed2 = await signLabel(data, TEST_DID, key); 55 55 + 56 56 + // label fields should be identical 57 57 + expect(signed1.src).toBe(signed2.src); 58 58 + expect(signed1.uri).toBe(signed2.uri); 59 59 + expect(signed1.val).toBe(signed2.val); 60 60 + expect(signed1.neg).toBe(signed2.neg); 61 61 + expect(signed1.cts).toBe(signed2.cts); 62 62 + 63 63 + // both signatures should be valid 64-byte compact signatures 64 64 + expect(signed1.sig.byteLength).toBe(64); 65 65 + expect(signed2.sig.byteLength).toBe(64); 66 66 + }); 67 67 + 68 68 + it('should include optional fields when provided', async () => { 69 69 + const key = await getTestKey(); 70 70 + 71 71 + const signed = await signLabel( 72 72 + { 73 73 + uri: 'at://did:plc:target/app.bsky.feed.post/abc', 74 74 + val: 'nudity', 75 75 + cid: 'bafyreib2rxk3rybk3aobmv5cjuql3setrnhfekwxbdmg7il4q2hr3hilqe', 76 76 + neg: true, 77 77 + exp: '2026-01-01T00:00:00.000Z', 78 78 + }, 79 79 + TEST_DID, 80 80 + key, 81 81 + ); 82 82 + 83 83 + expect(signed.cid).toBe('bafyreib2rxk3rybk3aobmv5cjuql3setrnhfekwxbdmg7il4q2hr3hilqe'); 84 84 + expect(signed.neg).toBe(true); 85 85 + expect(signed.exp).toBe('2026-01-01T00:00:00.000Z'); 86 86 + }); 87 87 + }); 88 88 + 89 89 + describe('formatLabel', () => { 90 90 + it('should convert sig to Bytes wrapper and set ver to 1', () => { 91 91 + const saved: SavedLabel = { 92 92 + seq: 1, 93 93 + src: TEST_DID, 94 94 + uri: 'did:plc:target', 95 95 + val: 'spam', 96 96 + neg: false, 97 97 + cts: '2025-01-01T00:00:00.000Z', 98 98 + sig: new Uint8Array(64), 99 99 + }; 100 100 + 101 101 + const formatted = formatLabel(saved); 102 102 + 103 103 + expect(formatted.ver).toBe(1); 104 104 + expect(formatted.src).toBe(TEST_DID); 105 105 + expect(isBytes(formatted.sig)).toBe(true); 106 106 + expect(fromBytes(formatted.sig)).toEqual(saved.sig); 107 107 + }); 108 108 + 109 109 + it('should include optional fields when present', () => { 110 110 + const saved: SavedLabel = { 111 111 + seq: 1, 112 112 + src: TEST_DID, 113 113 + uri: 'at://did:plc:target/app.bsky.feed.post/abc', 114 114 + val: 'nudity', 115 115 + neg: true, 116 116 + cts: '2025-01-01T00:00:00.000Z', 117 117 + cid: 'bafyreib2rxk3rybk3aobmv5cjuql3setrnhfekwxbdmg7il4q2hr3hilqe', 118 118 + exp: '2026-01-01T00:00:00.000Z', 119 119 + sig: new Uint8Array(64), 120 120 + }; 121 121 + 122 122 + const formatted = formatLabel(saved); 123 123 + 124 124 + expect(formatted.cid).toBe(saved.cid); 125 125 + expect(formatted.exp).toBe(saved.exp); 126 126 + expect(formatted.neg).toBe(true); 127 127 + }); 128 128 + 129 129 + it('should omit optional fields when absent', () => { 130 130 + const saved: SavedLabel = { 131 131 + seq: 1, 132 132 + src: TEST_DID, 133 133 + uri: 'did:plc:target', 134 134 + val: 'spam', 135 135 + neg: false, 136 136 + cts: '2025-01-01T00:00:00.000Z', 137 137 + sig: new Uint8Array(64), 138 138 + }; 139 139 + 140 140 + const formatted = formatLabel(saved); 141 141 + 142 142 + expect('cid' in formatted).toBe(false); 143 143 + expect('exp' in formatted).toBe(false); 144 144 + }); 145 145 + });

+124

packages/servers/labeler/lib/labels.ts

reviewed

··· 1 1 + import type { ComAtprotoLabelDefs } from '@atcute/atproto'; 2 2 + import { encode, toBytes } from '@atcute/cbor'; 3 3 + import type { PrivateKey } from '@atcute/crypto'; 4 4 + import type { Did, GenericUri } from '@atcute/lexicons'; 5 5 + 6 6 + import type { SavedLabel } from './store.ts'; 7 7 + 8 8 + const LABEL_VERSION = 1; 9 9 + 10 10 + /** data for creating a new label. */ 11 11 + export interface CreateLabelData { 12 12 + /** AT URI of the record, repository (account), or other resource */ 13 13 + uri: string; 14 14 + /** CID specifying the version of `uri` to label */ 15 15 + cid?: string; 16 16 + /** the label value */ 17 17 + val: string; 18 18 + /** whether this label negates a previous label */ 19 19 + neg?: boolean; 20 20 + /** creation timestamp (ISO 8601). defaults to current time */ 21 21 + cts?: string; 22 22 + /** expiration timestamp (ISO 8601) */ 23 23 + exp?: string; 24 24 + /** DID of the label source. defaults to the labeler's DID */ 25 25 + src?: string; 26 26 + } 27 27 + 28 28 + /** subject of a label: a URI with optional CID. */ 29 29 + export interface LabelSubject { 30 30 + uri: string; 31 31 + cid?: string; 32 32 + } 33 33 + 34 34 + interface UnsignedLabel { 35 35 + ver: number; 36 36 + src: string; 37 37 + uri: string; 38 38 + cid?: string; 39 39 + val: string; 40 40 + neg: boolean; 41 41 + cts: string; 42 42 + exp?: string; 43 43 + } 44 44 + 45 45 + /** 46 46 + * create an unsigned label object with version and defaults. 47 47 + * @param data label creation data 48 48 + * @param src default source DID 49 49 + * @returns unsigned label ready for signing 50 50 + */ 51 51 + const toUnsignedLabel = (data: CreateLabelData, src: Did): UnsignedLabel => { 52 52 + const label: UnsignedLabel = { 53 53 + ver: LABEL_VERSION, 54 54 + src: data.src ?? src, 55 55 + uri: data.uri, 56 56 + val: data.val, 57 57 + neg: data.neg ?? false, 58 58 + cts: data.cts ?? new Date().toISOString(), 59 59 + }; 60 60 + 61 61 + if (data.cid !== undefined) { 62 62 + label.cid = data.cid; 63 63 + } 64 64 + if (data.exp !== undefined) { 65 65 + label.exp = data.exp; 66 66 + } 67 67 + 68 68 + return label; 69 69 + }; 70 70 + 71 71 + /** 72 72 + * CBOR-encode and sign a label. 73 73 + * @param data label creation data 74 74 + * @param src default source DID 75 75 + * @param key private key for signing 76 76 + * @returns the label fields and signature, ready for storage 77 77 + */ 78 78 + export const signLabel = async ( 79 79 + data: CreateLabelData, 80 80 + src: Did, 81 81 + key: PrivateKey, 82 82 + ): Promise<Omit<SavedLabel, 'seq'>> => { 83 83 + const label = toUnsignedLabel(data, src); 84 84 + const bytes = encode(label); 85 85 + const sig = await key.sign(bytes); 86 86 + 87 87 + return { 88 88 + src: label.src, 89 89 + uri: label.uri, 90 90 + cid: label.cid, 91 91 + val: label.val, 92 92 + neg: label.neg, 93 93 + cts: label.cts, 94 94 + exp: label.exp, 95 95 + sig: sig, 96 96 + }; 97 97 + }; 98 98 + 99 99 + /** 100 100 + * format a saved label for wire transmission. 101 101 + * converts raw sig bytes to CBOR Bytes wrapper. 102 102 + * @param label saved label from store 103 103 + * @returns formatted label for XRPC responses 104 104 + */ 105 105 + export const formatLabel = (label: SavedLabel): ComAtprotoLabelDefs.Label => { 106 106 + const formatted: ComAtprotoLabelDefs.Label = { 107 107 + ver: 1, 108 108 + src: label.src as Did, 109 109 + uri: label.uri as GenericUri, 110 110 + val: label.val, 111 111 + neg: label.neg, 112 112 + cts: label.cts, 113 113 + sig: toBytes(label.sig), 114 114 + }; 115 115 + 116 116 + if (label.cid !== undefined) { 117 117 + formatted.cid = label.cid; 118 118 + } 119 119 + if (label.exp !== undefined) { 120 120 + formatted.exp = label.exp; 121 121 + } 122 122 + 123 123 + return formatted; 124 124 + };

+191

packages/servers/labeler/lib/outbox.test.ts

reviewed

··· 1 1 + import { XRPCSubscriptionError } from '@atcute/xrpc-server'; 2 2 + 3 3 + import { SimpleEventEmitter } from '@mary-ext/simple-event-emitter'; 4 4 + import { describe, expect, it } from 'vitest'; 5 5 + 6 6 + import { AsyncBuffer, AsyncBufferFullError } from './async-buffer.ts'; 7 7 + import { LabelOutbox } from './outbox.ts'; 8 8 + import type { LabelStore, SavedLabel } from './store.ts'; 9 9 + 10 10 + const makeSavedLabel = (seq: number): SavedLabel => ({ 11 11 + seq, 12 12 + src: 'did:plc:labeler', 13 13 + uri: 'did:plc:target', 14 14 + val: 'spam', 15 15 + neg: false, 16 16 + cts: new Date().toISOString(), 17 17 + sig: new Uint8Array(64), 18 18 + }); 19 19 + 20 20 + const createMockStore = (labels: SavedLabel[] = []): LabelStore => { 21 21 + return { 22 22 + async save(label) { 23 23 + const seq = labels.length + 1; 24 24 + const saved = { ...label, seq }; 25 25 + labels.push(saved); 26 26 + return saved; 27 27 + }, 28 28 + async query() { 29 29 + return { labels: [], cursor: '0' }; 30 30 + }, 31 31 + async getLatestSeq() { 32 32 + return labels.at(-1)?.seq ?? 0; 33 33 + }, 34 34 + async getRange(after, limit) { 35 35 + const result = labels.filter((l) => l.seq > after); 36 36 + return limit !== undefined ? result.slice(0, limit) : result; 37 37 + }, 38 38 + }; 39 39 + }; 40 40 + 41 41 + describe('AsyncBuffer', () => { 42 42 + it('should throw AsyncBufferFullError on overflow', async () => { 43 43 + const buffer = new AsyncBuffer<number>(2); 44 44 + 45 45 + // push 4 items (exceeding max of 2) 46 46 + buffer.push(1); 47 47 + buffer.push(2); 48 48 + buffer.push(3); 49 49 + buffer.push(4); 50 50 + 51 51 + const collected: number[] = []; 52 52 + 53 53 + await expect(async () => { 54 54 + for await (const value of buffer.events()) { 55 55 + collected.push(value); 56 56 + } 57 57 + }).rejects.toThrow(AsyncBufferFullError); 58 58 + }); 59 59 + }); 60 60 + 61 61 + describe('LabelOutbox', () => { 62 62 + it('should backfill from cursor', async () => { 63 63 + const labels = Array.from({ length: 5 }, (_, i) => makeSavedLabel(i + 1)); 64 64 + const store = createMockStore(labels); 65 65 + const emitter = new SimpleEventEmitter<[SavedLabel]>(); 66 66 + 67 67 + const outbox = new LabelOutbox(store, emitter); 68 68 + const ac = new AbortController(); 69 69 + 70 70 + const collected: SavedLabel[] = []; 71 71 + 72 72 + setTimeout(() => ac.abort(), 100); 73 73 + 74 74 + for await (const label of outbox.events(0, ac.signal)) { 75 75 + collected.push(label); 76 76 + if (collected.length === 5) { 77 77 + ac.abort(); 78 78 + break; 79 79 + } 80 80 + } 81 81 + 82 82 + expect(collected).toHaveLength(5); 83 83 + expect(collected.map((l) => l.seq)).toEqual([1, 2, 3, 4, 5]); 84 84 + }); 85 85 + 86 86 + it('should tail live events with no cursor', async () => { 87 87 + const store = createMockStore(); 88 88 + const emitter = new SimpleEventEmitter<[SavedLabel]>(); 89 89 + 90 90 + const outbox = new LabelOutbox(store, emitter); 91 91 + const ac = new AbortController(); 92 92 + 93 93 + const collected: SavedLabel[] = []; 94 94 + 95 95 + const iter = outbox.events(undefined, ac.signal); 96 96 + 97 97 + // emit after a tick so the outbox is tailing 98 98 + queueMicrotask(() => { 99 99 + emitter.emit(makeSavedLabel(1)); 100 100 + }); 101 101 + 102 102 + for await (const label of iter) { 103 103 + collected.push(label); 104 104 + if (collected.length === 1) { 105 105 + ac.abort(); 106 106 + break; 107 107 + } 108 108 + } 109 109 + 110 110 + expect(collected).toHaveLength(1); 111 111 + expect(collected[0]!.seq).toBe(1); 112 112 + }); 113 113 + 114 114 + it('should backfill then tail live events', async () => { 115 115 + const labels = [makeSavedLabel(1), makeSavedLabel(2)]; 116 116 + const store = createMockStore(labels); 117 117 + const emitter = new SimpleEventEmitter<[SavedLabel]>(); 118 118 + 119 119 + const outbox = new LabelOutbox(store, emitter); 120 120 + const ac = new AbortController(); 121 121 + 122 122 + const collected: SavedLabel[] = []; 123 123 + 124 124 + // emit a new label shortly after start so the outbox picks it up during tailing 125 125 + setTimeout(() => { 126 126 + emitter.emit(makeSavedLabel(3)); 127 127 + }, 20); 128 128 + 129 129 + for await (const label of outbox.events(0, ac.signal)) { 130 130 + collected.push(label); 131 131 + if (collected.length === 3) { 132 132 + ac.abort(); 133 133 + break; 134 134 + } 135 135 + } 136 136 + 137 137 + expect(collected.map((l) => l.seq)).toEqual([1, 2, 3]); 138 138 + }); 139 139 + 140 140 + it('should stop on abort signal', async () => { 141 141 + const store = createMockStore(); 142 142 + const emitter = new SimpleEventEmitter<[SavedLabel]>(); 143 143 + 144 144 + const outbox = new LabelOutbox(store, emitter); 145 145 + const ac = new AbortController(); 146 146 + 147 147 + const collected: SavedLabel[] = []; 148 148 + 149 149 + setTimeout(() => ac.abort(), 50); 150 150 + 151 151 + for await (const label of outbox.events(undefined, ac.signal)) { 152 152 + collected.push(label); 153 153 + } 154 154 + 155 155 + expect(collected).toHaveLength(0); 156 156 + }); 157 157 + 158 158 + it('should wrap AsyncBufferFullError as ConsumerTooSlow', async () => { 159 159 + const store = createMockStore(); 160 160 + const emitter = new SimpleEventEmitter<[SavedLabel]>(); 161 161 + 162 162 + const outbox = new LabelOutbox(store, emitter, { maxBufferSize: 2 }); 163 163 + const ac = new AbortController(); 164 164 + 165 165 + let error: unknown; 166 166 + const done = (async () => { 167 167 + try { 168 168 + for await (const _label of outbox.events(undefined, ac.signal)) { 169 169 + // stall the consumer — don't break, just wait for more 170 170 + // while the producer floods the buffer below 171 171 + await new Promise((resolve) => setTimeout(resolve, 50)); 172 172 + } 173 173 + } catch (err) { 174 174 + error = err; 175 175 + } 176 176 + })(); 177 177 + 178 178 + // wait for the outbox to start tailing 179 179 + await new Promise((resolve) => setTimeout(resolve, 10)); 180 180 + 181 181 + // flood the buffer well beyond capacity 182 182 + for (let i = 1; i <= 10; i++) { 183 183 + emitter.emit(makeSavedLabel(i)); 184 184 + } 185 185 + 186 186 + await done; 187 187 + 188 188 + expect(error).toBeInstanceOf(XRPCSubscriptionError); 189 189 + expect((error as XRPCSubscriptionError).error).toBe('ConsumerTooSlow'); 190 190 + }); 191 191 + });

+111

packages/servers/labeler/lib/outbox.ts

reviewed

··· 1 1 + import { XRPCSubscriptionError } from '@atcute/xrpc-server'; 2 2 + 3 3 + import type { SimpleEventEmitter } from '@mary-ext/simple-event-emitter'; 4 4 + 5 5 + import { AsyncBufferFullError, on } from './async-buffer.ts'; 6 6 + import type { LabelStore, SavedLabel } from './store.ts'; 7 7 + 8 8 + const BACKFILL_PAGE_SIZE = 500; 9 9 + 10 10 + export interface LabelOutboxOptions { 11 11 + maxBufferSize?: number; 12 12 + } 13 13 + 14 14 + /** 15 15 + * outbox for streaming labels with push-pull semantics. 16 16 + * 17 17 + * handles backfill from cursor, catch-up, and live tailing 18 18 + * with backpressure via bounded async buffer. 19 19 + */ 20 20 + export class LabelOutbox { 21 21 + #store: LabelStore; 22 22 + #emitter: SimpleEventEmitter<[label: SavedLabel]>; 23 23 + #maxBufferSize: number; 24 24 + 25 25 + constructor( 26 26 + store: LabelStore, 27 27 + emitter: SimpleEventEmitter<[label: SavedLabel]>, 28 28 + options: LabelOutboxOptions = {}, 29 29 + ) { 30 30 + this.#store = store; 31 31 + this.#emitter = emitter; 32 32 + this.#maxBufferSize = options.maxBufferSize ?? 500; 33 33 + } 34 34 + 35 35 + /** 36 36 + * stream labels, optionally backfilling from a cursor. 37 37 + * @param cursor sequence number to backfill from (exclusive), or undefined for live-only 38 38 + * @param signal abort signal to stop streaming 39 39 + * @returns async iterator of saved labels 40 40 + */ 41 41 + async *events(cursor: number | undefined, signal: AbortSignal): AsyncGenerator<SavedLabel> { 42 42 + let lastBackfillSeq = -1; 43 43 + let caughtUp = cursor === undefined; 44 44 + 45 45 + // backfill phase: dump stored labels in pages 46 46 + if (!caughtUp) { 47 47 + while (true) { 48 48 + const events = await this.#store.getRange( 49 49 + lastBackfillSeq > -1 ? lastBackfillSeq : cursor!, 50 50 + BACKFILL_PAGE_SIZE, 51 51 + ); 52 52 + 53 53 + if (events.length === 0) { 54 54 + break; 55 55 + } 56 56 + 57 57 + yield* events; 58 58 + signal.throwIfAborted(); 59 59 + 60 60 + lastBackfillSeq = events.at(-1)!.seq; 61 61 + 62 62 + // stop when close to the head 63 63 + const latestSeq = await this.#store.getLatestSeq(); 64 64 + if (latestSeq - lastBackfillSeq < BACKFILL_PAGE_SIZE / 2) { 65 65 + break; 66 66 + } 67 67 + } 68 68 + 69 69 + signal.throwIfAborted(); 70 70 + } 71 71 + 72 72 + // start listening before reading the gap 73 73 + const tail = on(this.#emitter, { signal, maxSize: this.#maxBufferSize }); 74 74 + 75 75 + // catch-up phase: read any labels between backfill end and tail start 76 76 + if (!caughtUp) { 77 77 + const events = await this.#store.getRange(lastBackfillSeq > -1 ? lastBackfillSeq : cursor!); 78 78 + 79 79 + if (events.length > 0) { 80 80 + yield* events; 81 81 + signal.throwIfAborted(); 82 82 + 83 83 + lastBackfillSeq = events.at(-1)!.seq; 84 84 + } 85 85 + } 86 86 + 87 87 + // tail phase: stream live events, deduplicating the catch-up overlap 88 88 + try { 89 89 + for await (const event of tail) { 90 90 + if (!caughtUp) { 91 91 + if (event.seq <= lastBackfillSeq) { 92 92 + continue; 93 93 + } 94 94 + 95 95 + caughtUp = true; 96 96 + } 97 97 + 98 98 + yield event; 99 99 + } 100 100 + } catch (err) { 101 101 + if (err instanceof AsyncBufferFullError) { 102 102 + throw new XRPCSubscriptionError({ 103 103 + error: 'ConsumerTooSlow', 104 104 + description: `stream consumer too slow`, 105 105 + }); 106 106 + } 107 107 + 108 108 + throw err; 109 109 + } 110 110 + } 111 111 + }

+55

packages/servers/labeler/lib/queue.ts

reviewed

··· 1 1 + interface Node<T> { 2 2 + value: T; 3 3 + next: Node<T> | undefined; 4 4 + } 5 5 + 6 6 + /** a FIFO queue backed by a linked list. */ 7 7 + export class Queue<T> { 8 8 + #head: Node<T> | undefined; 9 9 + #tail: Node<T> | undefined; 10 10 + #size: number = 0; 11 11 + 12 12 + /** number of items in the queue */ 13 13 + get size(): number { 14 14 + return this.#size; 15 15 + } 16 16 + 17 17 + /** 18 18 + * add a value to the end of the queue. 19 19 + * @param value value to enqueue 20 20 + */ 21 21 + enqueue(value: T): void { 22 22 + const node: Node<T> = { value, next: undefined }; 23 23 + const tail = this.#tail; 24 24 + 25 25 + if (tail !== undefined) { 26 26 + tail.next = node; 27 27 + } else { 28 28 + this.#head = node; 29 29 + } 30 30 + 31 31 + this.#tail = node; 32 32 + this.#size++; 33 33 + } 34 34 + 35 35 + /** 36 36 + * remove and return the first value from the queue. 37 37 + * @returns first queued value, or undefined if empty 38 38 + */ 39 39 + dequeue(): T | undefined { 40 40 + const head = this.#head; 41 41 + if (head === undefined) { 42 42 + return; 43 43 + } 44 44 + 45 45 + const next = head.next; 46 46 + 47 47 + this.#head = next; 48 48 + if (next === undefined) { 49 49 + this.#tail = undefined; 50 50 + } 51 51 + 52 52 + this.#size--; 53 53 + return head.value; 54 54 + } 55 55 + }

+68

packages/servers/labeler/lib/store.ts

reviewed

··· 1 1 + import type { ComAtprotoLabelDefs } from '@atcute/atproto'; 2 2 + 3 3 + /** label as stored in the database, with sequence number and raw sig bytes. */ 4 4 + export interface SavedLabel { 5 5 + /** sequence number / ID of the label */ 6 6 + seq: number; 7 7 + /** DID of the actor who created this label */ 8 8 + src: string; 9 9 + /** AT URI of the record, repository (account), or other resource */ 10 10 + uri: string; 11 11 + /** CID specifying the version of `uri` to label */ 12 12 + cid?: string; 13 13 + /** the label value */ 14 14 + val: string; 15 15 + /** whether this label negates a previous label */ 16 16 + neg: boolean; 17 17 + /** creation timestamp (ISO 8601) */ 18 18 + cts: string; 19 19 + /** expiration timestamp (ISO 8601) */ 20 20 + exp?: string; 21 21 + /** signature bytes */ 22 22 + sig: Uint8Array; 23 23 + } 24 24 + 25 25 + /** parameters for querying labels. */ 26 26 + export interface LabelQueryParams { 27 27 + uriPatterns: string[]; 28 28 + sources: string[]; 29 29 + cursor: number; 30 30 + limit: number; 31 31 + } 32 32 + 33 33 + /** result of a label query. */ 34 34 + export interface LabelQueryResult { 35 35 + labels: ComAtprotoLabelDefs.Label[]; 36 36 + cursor: string; 37 37 + } 38 38 + 39 39 + /** pluggable storage backend for labels. */ 40 40 + export interface LabelStore { 41 41 + /** 42 42 + * save a signed label and assign it a sequence number. 43 43 + * @param label the signed label to save 44 44 + * @returns the saved label with sequence number 45 45 + */ 46 46 + save(label: Omit<SavedLabel, 'seq'>): Promise<SavedLabel>; 47 47 + 48 48 + /** 49 49 + * query labels matching the given parameters. 50 50 + * @param params query parameters 51 51 + * @returns matching labels and cursor 52 52 + */ 53 53 + query(params: LabelQueryParams): Promise<LabelQueryResult>; 54 54 + 55 55 + /** 56 56 + * get the latest sequence number. 57 57 + * @returns the highest sequence number, or 0 if none 58 58 + */ 59 59 + getLatestSeq(): Promise<number>; 60 60 + 61 61 + /** 62 62 + * get a range of labels after a given sequence number. 63 63 + * @param after sequence number to start after (exclusive) 64 64 + * @param limit maximum number of labels to return 65 65 + * @returns labels in the range 66 66 + */ 67 67 + getRange(after: number, limit?: number): Promise<SavedLabel[]>; 68 68 + }

+42

packages/servers/labeler/package.json

reviewed

··· 1 1 + { 2 2 + "name": "@atcute/labeler", 3 3 + "version": "0.1.0", 4 4 + "description": "sign and emit AT Protocol labels", 5 5 + "license": "0BSD", 6 6 + "repository": { 7 7 + "url": "https://github.com/mary-ext/atcute", 8 8 + "directory": "packages/servers/labeler" 9 9 + }, 10 10 + "files": [ 11 11 + "dist/", 12 12 + "lib/", 13 13 + "!lib/**/*.bench.ts", 14 14 + "!lib/**/*.test.ts" 15 15 + ], 16 16 + "type": "module", 17 17 + "exports": { 18 18 + ".": "./dist/index.js" 19 19 + }, 20 20 + "publishConfig": { 21 21 + "access": "public" 22 22 + }, 23 23 + "scripts": { 24 24 + "build": "tsgo --project tsconfig.build.json", 25 25 + "test": "vitest", 26 26 + "prepublish": "rm -rf dist; pnpm run build" 27 27 + }, 28 28 + "dependencies": { 29 29 + "@atcute/cbor": "workspace:^", 30 30 + "@atcute/crypto": "workspace:^", 31 31 + "@atcute/lexicons": "workspace:^", 32 32 + "@atcute/xrpc-server": "workspace:^", 33 33 + "@mary-ext/simple-event-emitter": "^1.0.1" 34 34 + }, 35 35 + "devDependencies": { 36 36 + "@atcute/atproto": "workspace:^", 37 37 + "@atcute/labeler": "file:", 38 38 + "@atcute/ozone": "workspace:^", 39 39 + "@types/node": "^25.2.3", 40 40 + "vitest": "^4.0.18" 41 41 + } 42 42 + }

+4

packages/servers/labeler/tsconfig.build.json

reviewed

··· 1 1 + { 2 2 + "extends": "./tsconfig.json", 3 3 + "exclude": ["**/*.test.ts"] 4 4 + }

+27

packages/servers/labeler/tsconfig.json

reviewed

··· 1 1 + { 2 2 + "compilerOptions": { 3 3 + "types": ["node"], 4 4 + "rootDir": "lib/", 5 5 + "outDir": "dist/", 6 6 + "esModuleInterop": true, 7 7 + "skipLibCheck": true, 8 8 + "target": "ESNext", 9 9 + "allowJs": true, 10 10 + "resolveJsonModule": true, 11 11 + "moduleDetection": "force", 12 12 + "isolatedModules": true, 13 13 + "verbatimModuleSyntax": true, 14 14 + "allowImportingTsExtensions": true, 15 15 + "rewriteRelativeImportExtensions": true, 16 16 + "strict": true, 17 17 + "noImplicitOverride": true, 18 18 + "noUnusedLocals": true, 19 19 + "noUnusedParameters": true, 20 20 + "noFallthroughCasesInSwitch": true, 21 21 + "module": "NodeNext", 22 22 + "sourceMap": true, 23 23 + "declaration": true, 24 24 + "declarationMap": true 25 25 + }, 26 26 + "include": ["lib"] 27 27 + }

+58 -7

pnpm-lock.yaml

reviewed

··· 728 728 prebuildify: 729 729 specifier: ^6.0.1 730 730 version: 6.0.1 731 731 - vitest: 732 732 - specifier: ^4.0.18 733 733 - version: 4.0.18(@types/node@25.2.1)(@vitest/browser-playwright@4.0.18)(jiti@2.6.1)(tsx@4.20.6)(yaml@2.8.0) 734 731 735 732 packages/misc/uint8array: 736 733 devDependencies: ··· 798 795 '@atcute/oauth-crypto': 799 796 specifier: workspace:^ 800 797 version: link:../crypto 801 801 - devDependencies: 802 802 - vitest: 803 803 - specifier: ^4.0.18 804 804 - version: 4.0.18(@types/node@25.2.1)(@vitest/browser-playwright@4.0.18)(jiti@2.6.1)(tsx@4.20.6)(yaml@2.8.0) 805 798 806 799 packages/oauth/node-client: 807 800 dependencies: ··· 918 911 specifier: ^4.0.18 919 912 version: 4.0.18(@types/node@25.2.3)(@vitest/browser-playwright@4.0.18)(jiti@2.6.1)(tsx@4.20.6)(yaml@2.8.0) 920 913 914 914 + packages/servers/labeler: 915 915 + dependencies: 916 916 + '@atcute/cbor': 917 917 + specifier: workspace:^ 918 918 + version: link:../../utilities/cbor 919 919 + '@atcute/crypto': 920 920 + specifier: workspace:^ 921 921 + version: link:../../utilities/crypto 922 922 + '@atcute/identity': 923 923 + specifier: workspace:^ 924 924 + version: link:../../identity/identity 925 925 + '@atcute/identity-resolver': 926 926 + specifier: workspace:^ 927 927 + version: link:../../identity/identity-resolver 928 928 + '@atcute/lexicons': 929 929 + specifier: workspace:^ 930 930 + version: link:../../lexicons/lexicons 931 931 + '@atcute/xrpc-server': 932 932 + specifier: workspace:^ 933 933 + version: link:../xrpc-server 934 934 + '@mary-ext/simple-event-emitter': 935 935 + specifier: ^1.0.1 936 936 + version: 1.0.1 937 937 + devDependencies: 938 938 + '@atcute/atproto': 939 939 + specifier: workspace:^ 940 940 + version: link:../../definitions/atproto 941 941 + '@atcute/labeler': 942 942 + specifier: 'file:' 943 943 + version: file:packages/servers/labeler 944 944 + '@atcute/ozone': 945 945 + specifier: workspace:^ 946 946 + version: link:../../definitions/ozone 947 947 + '@types/node': 948 948 + specifier: ^25.2.3 949 949 + version: 25.2.3 950 950 + vitest: 951 951 + specifier: ^4.0.18 952 952 + version: 4.0.18(@types/node@25.2.3)(@vitest/browser-playwright@4.0.18)(jiti@2.6.1)(tsx@4.20.6)(yaml@2.8.0) 953 953 + 921 954 packages/servers/xrpc-server: 922 955 dependencies: 923 956 '@atcute/cbor': ··· 1245 1278 '@atcute/frontpage@file:packages/definitions/frontpage': 1246 1279 resolution: {directory: packages/definitions/frontpage, type: directory} 1247 1280 1281 1281 + '@atcute/labeler@file:packages/servers/labeler': 1282 1282 + resolution: {directory: packages/servers/labeler, type: directory} 1283 1283 + 1248 1284 '@atcute/leaflet@file:packages/definitions/leaflet': 1249 1285 resolution: {directory: packages/definitions/leaflet, type: directory} 1250 1286 ··· 2096 2132 2097 2133 '@mary-ext/simple-event-emitter@1.0.0': 2098 2134 resolution: {integrity: sha512-meA/zJZKIN1RVBNEYIbjufkUrW7/tRjHH60FjolpG1ixJKo76TB208qefQLNdOVDA7uIG0CGEDuhmMirtHKLAg==} 2135 2135 + 2136 2136 + '@mary-ext/simple-event-emitter@1.0.1': 2137 2137 + resolution: {integrity: sha512-9+VvZisxZ/gSg+JJH7hmXaA8Qj42Qjz3O58RSB+INYc8iLA0icATZxHB9vKbj59ojDGZjO3hCKzMXocx3L0H8w==} 2099 2138 2100 2139 '@mitata/counters@0.0.8': 2101 2140 resolution: {integrity: sha512-f11w0Y1ETFlarDP7CePj8Z+y8Gv5Ax4gMxWsEwrqh0kH/YIY030Ezx5SUJeQg0YPTZ2OHKGcLG1oGJbIqHzaJA==} ··· 4664 4703 '@atcute/atproto': link:packages/definitions/atproto 4665 4704 '@atcute/lexicons': link:packages/lexicons/lexicons 4666 4705 4706 4706 + '@atcute/labeler@file:packages/servers/labeler': 4707 4707 + dependencies: 4708 4708 + '@atcute/cbor': link:packages/utilities/cbor 4709 4709 + '@atcute/crypto': link:packages/utilities/crypto 4710 4710 + '@atcute/identity': link:packages/identity/identity 4711 4711 + '@atcute/identity-resolver': link:packages/identity/identity-resolver 4712 4712 + '@atcute/lexicons': link:packages/lexicons/lexicons 4713 4713 + '@atcute/xrpc-server': link:packages/servers/xrpc-server 4714 4714 + '@mary-ext/simple-event-emitter': 1.0.1 4715 4715 + 4667 4716 '@atcute/leaflet@file:packages/definitions/leaflet': 4668 4717 dependencies: 4669 4718 '@atcute/atproto': link:packages/definitions/atproto ··· 6120 6169 yocto-queue: 1.2.2 6121 6170 6122 6171 '@mary-ext/simple-event-emitter@1.0.0': {} 6172 6172 + 6173 6173 + '@mary-ext/simple-event-emitter@1.0.1': {} 6123 6174 6124 6175 '@mitata/counters@0.0.8': {} 6125 6176