Streaming repo subscriptions

lexicons/com/atproto/sync/subscribeAllRepos.json

@@ -0,0 +1,53 @@
+{
+  "lexicon": 1,
+  "id": "com.atproto.sync.subscribeAllRepos",
+  "defs": {
+    "main": {
+      "type": "subscription",
+      "description": "Subscribe to repo updates",
+      "parameters": {
+        "type": "params",
+        "properties": {
+          "backfillFrom": {
+            "type": "datetime",
+            "description": "The last known event to backfill from. Does not dedupe as there may be an overlap in timestamps."
+          }
+        }
+      },
+      "message": {
+        "schema": {
+          "type": "union",
+          "refs": ["#repoAppend", "#repoRebase"]
+        },
+        "codes": {
+          "#repoAppend": 0,
+          "#repoRebase": 1
+        }
+      }
+    },
+    "repoAppend": {
+      "type": "object",
+      "required": ["time", "repo", "commit", "blocks", "blobs"],
+      "properties": {
+        "time": {"type": "datetime"},
+        "repo": {"type": "string"},
+        "commit": {"type": "string"},
+        "prev": {"type": "string"},
+        "blocks": {"type": "unknown"},
+        "blobs": {
+          "type": "array",
+          "items": {"type": "string"}
+        }
+      }
+    },
+    "repoRebase": {
+      "type": "object",
+      "required": ["time", "repo", "commit"],
+      "properties": {
+        "time": {"type": "datetime"},
+        "repo": {"type": "string"},
+        "commit": {"type": "string"}
+      }
+    }
+  }
+}

packages/api/src/client/index.ts

@@ -53,6 +53,7 @@ import * as ComAtprotoSyncGetCommitPath from './types/com/atproto/sync/getCommit
 import * as ComAtprotoSyncGetHead from './types/com/atproto/sync/getHead'
 import * as ComAtprotoSyncGetRecord from './types/com/atproto/sync/getRecord'
 import * as ComAtprotoSyncGetRepo from './types/com/atproto/sync/getRepo'
+import * as ComAtprotoSyncSubscribeAllRepos from './types/com/atproto/sync/subscribeAllRepos'
 import * as AppBskyActorGetProfile from './types/app/bsky/actor/getProfile'
 import * as AppBskyActorGetProfiles from './types/app/bsky/actor/getProfiles'
 import * as AppBskyActorGetSuggestions from './types/app/bsky/actor/getSuggestions'
@@ -137,6 +138,7 @@ export * as ComAtprotoSyncGetCommitPath from './types/com/atproto/sync/getCommit
 export * as ComAtprotoSyncGetHead from './types/com/atproto/sync/getHead'
 export * as ComAtprotoSyncGetRecord from './types/com/atproto/sync/getRecord'
 export * as ComAtprotoSyncGetRepo from './types/com/atproto/sync/getRepo'
+export * as ComAtprotoSyncSubscribeAllRepos from './types/com/atproto/sync/subscribeAllRepos'
 export * as AppBskyActorGetProfile from './types/app/bsky/actor/getProfile'
 export * as AppBskyActorGetProfiles from './types/app/bsky/actor/getProfiles'
 export * as AppBskyActorGetSuggestions from './types/app/bsky/actor/getSuggestions'

packages/api/src/client/lexicons.ts

@@ -2173,6 +2173,81 @@ export const schemaDict = {
       },
     },
   },
+  ComAtprotoSyncSubscribeAllRepos: {
+    lexicon: 1,
+    id: 'com.atproto.sync.subscribeAllRepos',
+    defs: {
+      main: {
+        type: 'subscription',
+        description: 'Subscribe to repo updates',
+        parameters: {
+          type: 'params',
+          properties: {
+            backfillFrom: {
+              type: 'datetime',
+              description:
+                'The last known event to backfill from. Does not dedupe as there may be an overlap in timestamps.',
+            },
+          },
+        },
+        message: {
+          schema: {
+            type: 'union',
+            refs: [
+              'lex:com.atproto.sync.subscribeAllRepos#repoAppend',
+              'lex:com.atproto.sync.subscribeAllRepos#repoRebase',
+            ],
+          },
+          codes: {
+            'lex:com.atproto.sync.subscribeAllRepos#repoAppend': 0,
+            'lex:com.atproto.sync.subscribeAllRepos#repoRebase': 1,
+          },
+        },
+      },
+      repoAppend: {
+        type: 'object',
+        required: ['time', 'repo', 'commit', 'blocks', 'blobs'],
+        properties: {
+          time: {
+            type: 'datetime',
+          },
+          repo: {
+            type: 'string',
+          },
+          commit: {
+            type: 'string',
+          },
+          prev: {
+            type: 'string',
+          },
+          blocks: {
+            type: 'unknown',
+          },
+          blobs: {
+            type: 'array',
+            items: {
+              type: 'string',
+            },
+          },
+        },
+      },
+      repoRebase: {
+        type: 'object',
+        required: ['time', 'repo', 'commit'],
+        properties: {
+          time: {
+            type: 'datetime',
+          },
+          repo: {
+            type: 'string',
+          },
+          commit: {
+            type: 'string',
+          },
+        },
+      },
+    },
+  },
   AppBskyActorGetProfile: {
     lexicon: 1,
     id: 'app.bsky.actor.getProfile',
@@ -4119,6 +4194,7 @@ export const ids = {
   ComAtprotoSyncGetHead: 'com.atproto.sync.getHead',
   ComAtprotoSyncGetRecord: 'com.atproto.sync.getRecord',
   ComAtprotoSyncGetRepo: 'com.atproto.sync.getRepo',
+  ComAtprotoSyncSubscribeAllRepos: 'com.atproto.sync.subscribeAllRepos',
   AppBskyActorGetProfile: 'app.bsky.actor.getProfile',
   AppBskyActorGetProfiles: 'app.bsky.actor.getProfiles',
   AppBskyActorGetSuggestions: 'app.bsky.actor.getSuggestions',

packages/api/src/client/types/com/atproto/sync/subscribeAllRepos.ts

@@ -0,0 +1,48 @@
+/**
+ * GENERATED CODE - DO NOT MODIFY
+ */
+import { Headers, XRPCError } from '@atproto/xrpc'
+import { ValidationResult } from '@atproto/lexicon'
+import { isObj, hasProp } from '../../../../util'
+import { lexicons } from '../../../../lexicons'
+
+export interface RepoAppend {
+  time: string
+  repo: string
+  commit: string
+  prev?: string
+  blocks: {}
+  blobs: string[]
+  [k: string]: unknown
+}
+
+export function isRepoAppend(v: unknown): v is RepoAppend {
+  return (
+    isObj(v) &&
+    hasProp(v, '$type') &&
+    v.$type === 'com.atproto.sync.subscribeAllRepos#repoAppend'
+  )
+}
+
+export function validateRepoAppend(v: unknown): ValidationResult {
+  return lexicons.validate('com.atproto.sync.subscribeAllRepos#repoAppend', v)
+}
+
+export interface RepoRebase {
+  time: string
+  repo: string
+  commit: string
+  [k: string]: unknown
+}
+
+export function isRepoRebase(v: unknown): v is RepoRebase {
+  return (
+    isObj(v) &&
+    hasProp(v, '$type') &&
+    v.$type === 'com.atproto.sync.subscribeAllRepos#repoRebase'
+  )
+}
+
+export function validateRepoRebase(v: unknown): ValidationResult {
+  return lexicons.validate('com.atproto.sync.subscribeAllRepos#repoRebase', v)
+}

packages/common/src/async.ts

@@ -1,3 +1,23 @@
+import { wait } from './util'
+
+// reads values from a generator into a list
+// NOTE: does not signal generator to close. it *will* continue to produce values
+export const readFromGenerator = async <T>(
+  gen: AsyncGenerator<T>,
+  maxLength = Number.MAX_SAFE_INTEGER,
+  timeout = 1000,
+): Promise<T[]> => {
+  const evts: T[] = []
+  while (evts.length < maxLength) {
+    const maybeEvt = await Promise.race([gen.next(), wait(timeout)])
+    if (!maybeEvt) break
+    const evt = maybeEvt as IteratorResult<T>
+    if (evt.done) break
+    evts.push(evt.value)
+  }
+  return evts
+}
+
 export type Deferrable = {
   resolve: () => void
   complete: Promise<void>
@@ -22,3 +42,57 @@ export const createDeferrables = (count: number): Deferrable[] => {
 export const allComplete = async (deferrables: Deferrable[]): Promise<void> => {
   await Promise.all(deferrables.map((d) => d.complete))
 }
+
+export class AsyncBuffer<T> {
+  private buffer: T[] = []
+  private promise: Promise<void>
+  private resolve: () => void
+
+  constructor(public maxSize?: number) {
+    this.resetPromise()
+  }
+
+  get curr(): T[] {
+    return this.buffer
+  }
+
+  get size(): number {
+    return this.buffer.length
+  }
+
+  resetPromise() {
+    this.promise = new Promise<void>((r) => (this.resolve = r))
+  }
+
+  push(item: T) {
+    this.buffer.push(item)
+    this.resolve()
+  }
+
+  pushMany(items: T[]) {
+    items.forEach((i) => this.buffer.push(i))
+    this.resolve()
+  }
+
+  async *events(): AsyncGenerator<T> {
+    while (true) {
+      await this.promise
+      if (this.maxSize && this.size > this.maxSize) {
+        throw new AsyncBufferFullError(this.maxSize)
+      }
+      const [first, ...rest] = this.buffer
+      if (first) {
+        this.buffer = rest
+        yield first
+      } else {
+        this.resetPromise()
+      }
+    }
+  }
+}
+
+export class AsyncBufferFullError extends Error {
+  constructor(maxSize: number) {
+    super(`ReachedMaxBufferSize: ${maxSize}`)
+  }
+}

packages/common/src/index.ts

@@ -8,3 +8,4 @@ export * from './ipld'
 export * from './logger'
 export * from './types'
 export * from './streams'
+export * from './times'

packages/common/src/times.ts

@@ -0,0 +1,4 @@
+export const SECOND = 1000
+export const MINUTE = SECOND * 60
+export const HOUR = MINUTE * 60
+export const DAY = HOUR * 24

packages/common/tests/async.test.ts

@@ -0,0 +1,22 @@
+import { readFromGenerator, wait } from '../src'
+
+describe('async', () => {
+  describe('readFromGenerator', () => {
+    async function* waitToYield(time: number) {
+      for (let i = 0; i < 5; i++) {
+        await wait(time)
+        yield true
+      }
+    }
+
+    it('reads from generator with timeout', async () => {
+      const read = await readFromGenerator(waitToYield(100), undefined, 101)
+      expect(read).toEqual([true, true, true, true, true])
+    })
+
+    it('stops reading at timeout', async () => {
+      const read = await readFromGenerator(waitToYield(100), undefined, 99)
+      expect(read).toEqual([])
+    })
+  })
+})

packages/dev-env/src/index.ts

@@ -10,6 +10,7 @@ import * as plc from '@atproto/plc'
 import * as crypto from '@atproto/crypto'
 import AtpAgent from '@atproto/api'
 import { ServerType, ServerConfig, StartParams } from './types.js'
+import { HOUR } from '@atproto/common'
 
 interface Startable {
   start(): Promise<http.Server>
@@ -102,6 +103,8 @@ export class DevEnvServer {
               'f23ecd142835025f42c3db2cf25dd813956c178392760256211f9d315f8ab4d8',
             privacyPolicyUrl: 'https://example.com/privacy',
             termsOfServiceUrl: 'https://example.com/tos',
+            maxSubscriptionBuffer: 200,
+            repoBackfillLimitMs: HOUR,
           }),
         })
         await startServer(pds)

packages/lex-cli/src/codegen/client.ts

@@ -10,7 +10,6 @@ import {
   LexXrpcProcedure,
   LexXrpcQuery,
   LexRecord,
-  LexXrpcSubscription,
 } from '@atproto/lexicon'
 import { NSID } from '@atproto/nsid'
 import { gen, utilTs, lexiconsTs } from './common'
@@ -273,20 +272,17 @@ function genNamespaceCls(file: SourceFile, ns: DefTreeNode) {
 
   // methods
   for (const userType of ns.userTypes) {
-    if (
-      userType.def.type !== 'query' &&
-      userType.def.type !== 'subscription' &&
-      userType.def.type !== 'procedure'
-    ) {
+    if (userType.def.type !== 'query' && userType.def.type !== 'procedure') {
       continue
     }
+    const isGetReq = userType.def.type === 'query'
     const moduleName = toTitleCase(userType.nsid)
     const name = toCamelCase(NSID.parse(userType.nsid).name || '')
     const method = cls.addMethod({
       name,
       returnType: `Promise<${moduleName}.Response>`,
     })
-    if (userType.def.type === 'query' || userType.def.type === 'subscription') {
+    if (isGetReq) {
       method.addParameter({
         name: 'params?',
         type: `${moduleName}.QueryParams`,
@@ -304,7 +300,7 @@ function genNamespaceCls(file: SourceFile, ns: DefTreeNode) {
     method.setBodyText(
       [
         `return this._service.xrpc`,
-        userType.def.type === 'query'
+        isGetReq
           ? `.call('${userType.nsid}', params, undefined, opts)`
           : `.call('${userType.nsid}', opts?.qp, data, opts)`,
         `  .catch((e) => {`,