Allow syncing a commit range (#535)

allow syncing a commit range

lexicons/com/atproto/sync/getRepo.json

@@ -10,7 +10,8 @@
         "required": ["did"],
         "properties": {
           "did": {"type": "string", "description": "The DID of the repo."},
-          "from": {"type": "string", "description": "A past commit CID."}
+          "earliest": {"type": "string", "description": "The earliest commit in the commit range (not inclusive)"},
+          "latest": {"type": "string", "description": "The latest commit you in the commit range (inclusive"}
         }
       },
       "output": {

packages/api/src/client/lexicons.ts

@@ -2155,9 +2155,15 @@ export const schemaDict = {
               type: 'string',
               description: 'The DID of the repo.',
             },
-            from: {
+            earliest: {
               type: 'string',
-              description: 'A past commit CID.',
+              description:
+                'The earliest commit in the commit range (not inclusive)',
+            },
+            latest: {
+              type: 'string',
+              description:
+                'The latest commit you in the commit range (inclusive',
             },
           },
         },

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

@@ -9,8 +9,10 @@ import { lexicons } from '../../../../lexicons'
 export interface QueryParams {
   /** The DID of the repo. */
   did: string
-  /** A past commit CID. */
-  from?: string
+  /** The earliest commit in the commit range (not inclusive) */
+  earliest?: string
+  /** The latest commit you in the commit range (inclusive */
+  latest?: string
 }
 
 export type InputSchema = undefined

packages/pds/src/api/com/atproto/sync.ts

@@ -43,14 +43,16 @@ export default function (server: Server, ctx: AppContext) {
   })
 
   server.com.atproto.sync.getRepo(async ({ params }) => {
-    const { did, from = null } = params
+    const { did } = params
     const storage = new SqlRepoStorage(ctx.db, did)
-    const head = await storage.getHead()
-    if (head === null) {
+    const earliest = params.earliest ? CID.parse(params.earliest) : null
+    const latest = params.latest
+      ? CID.parse(params.latest)
+      : await storage.getHead()
+    if (latest === null) {
       throw new InvalidRequestError(`Could not find repo for DID: ${did}`)
     }
-    const fromCid = from ? CID.parse(from) : null
-    const diff = await repo.getDiff(storage, head, fromCid)
+    const diff = await repo.getDiff(storage, latest, earliest)
     return {
       encoding: 'application/vnd.ipld.car',
       body: Buffer.from(diff),

packages/pds/src/lexicon/lexicons.ts

@@ -2155,9 +2155,15 @@ export const schemaDict = {
               type: 'string',
               description: 'The DID of the repo.',
             },
-            from: {
+            earliest: {
               type: 'string',
-              description: 'A past commit CID.',
+              description:
+                'The earliest commit in the commit range (not inclusive)',
+            },
+            latest: {
+              type: 'string',
+              description:
+                'The latest commit you in the commit range (inclusive',
             },
           },
         },

packages/pds/src/lexicon/types/com/atproto/sync/getRepo.ts

@@ -11,8 +11,10 @@ import { HandlerAuth } from '@atproto/xrpc-server'
 export interface QueryParams {
   /** The DID of the repo. */
   did: string
-  /** A past commit CID. */
-  from?: string
+  /** The earliest commit in the commit range (not inclusive) */
+  earliest?: string
+  /** The latest commit you in the commit range (inclusive */
+  latest?: string
 }
 
 export type InputSchema = undefined

packages/pds/tests/sync.test.ts

@@ -97,7 +97,7 @@ describe('repo sync', () => {
 
     const car = await agent.api.com.atproto.sync.getRepo({
       did,
-      from: currRoot?.toString(),
+      earliest: currRoot?.toString(),
     })
     const currRepo = await repo.Repo.load(storage, currRoot)
     const synced = await repo.loadDiff(
@@ -137,6 +137,47 @@ describe('repo sync', () => {
     expect(partialCommitPath.data.commits).toEqual(localStr.slice(3, 16))
   })
 
+  it('syncs commit range', async () => {
+    const local = await storage.getCommits(currRoot as CID, null)
+    if (!local) {
+      throw new Error('Could not get local commit path')
+    }
+    const memoryStore = new MemoryBlockstore()
+    // first we load some baseline data (needed for parsing range)
+    const first = await agent.api.com.atproto.sync.getRepo({
+      did,
+      latest: local[2].commit.toString(),
+    })
+    const firstParsed = await repo.readCar(new Uint8Array(first.data))
+    memoryStore.putMany(firstParsed.blocks)
+
+    // then we load some commit range
+    const second = await agent.api.com.atproto.sync.getRepo({
+      did,
+      earliest: local[2].commit.toString(),
+      latest: local[15].commit.toString(),
+    })
+    const secondParsed = await repo.readCar(new Uint8Array(second.data))
+    memoryStore.putMany(secondParsed.blocks)
+
+    // then we verify we have all the commits in the range
+    const commits = await memoryStore.getCommits(
+      local[15].commit,
+      local[2].commit,
+    )
+    if (!commits) {
+      throw new Error('expected commits to be defined')
+    }
+    const localSlice = local.slice(2, 15)
+    expect(commits.length).toBe(localSlice.length)
+    for (let i = 0; i < commits.length; i++) {
+      const fromRemote = commits[i]
+      const fromLocal = localSlice[i]
+      expect(fromRemote.commit.equals(fromLocal.commit))
+      expect(fromRemote.blocks.equals(fromLocal.blocks))
+    }
+  })
+
   it('sync a repo checkout', async () => {
     const car = await agent.api.com.atproto.sync.getCheckout({ did })
     const checkoutStorage = new MemoryBlockstore()