Record proofs

lexicons/com/atproto/sync/getRecord.json

@@ -0,0 +1,23 @@
+{
+  "lexicon": 1,
+  "id": "com.atproto.sync.getRecord",
+  "defs": {
+    "main": {
+      "type": "query",
+      "description": "Gets blocks needed for existence or non-existence of record.",
+      "parameters": {
+        "type": "params",
+        "required": ["did", "collection", "rkey"],
+        "properties": {
+          "did": {"type": "string", "description": "The DID of the repo."},
+          "collection": {"type": "string" },
+          "rkey": {"type": "string" },
+          "commit": {"type": "string", "description": "An optional past commit CID."}
+        }
+      },
+      "output": {
+        "encoding": "application/cbor"
+      }
+    }
+  }
+}

packages/api/src/client/index.ts

@@ -39,6 +39,7 @@ import * as ComAtprotoSessionRefresh from './types/com/atproto/session/refresh'
 import * as ComAtprotoSyncGetCheckout from './types/com/atproto/sync/getCheckout'
 import * as ComAtprotoSyncGetCommitPath from './types/com/atproto/sync/getCommitPath'
 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 ComAtprotoSyncUpdateRepo from './types/com/atproto/sync/updateRepo'
 import * as AppBskyActorCreateScene from './types/app/bsky/actor/createScene'
@@ -116,6 +117,7 @@ export * as ComAtprotoSessionRefresh from './types/com/atproto/session/refresh'
 export * as ComAtprotoSyncGetCheckout from './types/com/atproto/sync/getCheckout'
 export * as ComAtprotoSyncGetCommitPath from './types/com/atproto/sync/getCommitPath'
 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 ComAtprotoSyncUpdateRepo from './types/com/atproto/sync/updateRepo'
 export * as AppBskyActorCreateScene from './types/app/bsky/actor/createScene'
@@ -615,6 +617,17 @@ export class SyncNS {
       })
   }
 
+  getRecord(
+    params?: ComAtprotoSyncGetRecord.QueryParams,
+    opts?: ComAtprotoSyncGetRecord.CallOptions,
+  ): Promise<ComAtprotoSyncGetRecord.Response> {
+    return this._service.xrpc
+      .call('com.atproto.sync.getRecord', params, undefined, opts)
+      .catch((e) => {
+        throw ComAtprotoSyncGetRecord.toKnownErr(e)
+      })
+  }
+
   getRepo(
     params?: ComAtprotoSyncGetRepo.QueryParams,
     opts?: ComAtprotoSyncGetRepo.CallOptions,

packages/api/src/client/lexicons.ts

@@ -1332,6 +1332,40 @@ export const schemaDict = {
       },
     },
   },
+  ComAtprotoSyncGetRecord: {
+    lexicon: 1,
+    id: 'com.atproto.sync.getRecord',
+    defs: {
+      main: {
+        type: 'query',
+        description:
+          'Gets blocks needed for existence or non-existence of record.',
+        parameters: {
+          type: 'params',
+          required: ['did', 'collection', 'rkey'],
+          properties: {
+            did: {
+              type: 'string',
+              description: 'The DID of the repo.',
+            },
+            collection: {
+              type: 'string',
+            },
+            rkey: {
+              type: 'string',
+            },
+            commit: {
+              type: 'string',
+              description: 'An optional past commit CID.',
+            },
+          },
+        },
+        output: {
+          encoding: 'application/cbor',
+        },
+      },
+    },
+  },
   ComAtprotoSyncGetRepo: {
     lexicon: 1,
     id: 'com.atproto.sync.getRepo',
@@ -3607,6 +3641,7 @@ export const ids = {
   ComAtprotoSyncGetCheckout: 'com.atproto.sync.getCheckout',
   ComAtprotoSyncGetCommitPath: 'com.atproto.sync.getCommitPath',
   ComAtprotoSyncGetHead: 'com.atproto.sync.getHead',
+  ComAtprotoSyncGetRecord: 'com.atproto.sync.getRecord',
   ComAtprotoSyncGetRepo: 'com.atproto.sync.getRepo',
   ComAtprotoSyncUpdateRepo: 'com.atproto.sync.updateRepo',
   AppBskyActorCreateScene: 'app.bsky.actor.createScene',

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

@@ -0,0 +1,31 @@
+/**
+ * GENERATED CODE - DO NOT MODIFY
+ */
+import { Headers, XRPCError } from '@atproto/xrpc'
+
+export interface QueryParams {
+  /** The DID of the repo. */
+  did: string
+  collection: string
+  rkey: string
+  /** An optional past commit CID. */
+  commit?: string
+}
+
+export type InputSchema = undefined
+
+export interface CallOptions {
+  headers?: Headers
+}
+
+export interface Response {
+  success: boolean
+  headers: Headers
+  data: Uint8Array
+}
+
+export function toKnownErr(e: any) {
+  if (e instanceof XRPCError) {
+  }
+  return e
+}

packages/common/src/index.ts

@@ -3,7 +3,7 @@ export * as util from './util'
 
 export * from './util'
 export * from './tid'
-export * from './blocks'
+export * from './ipld'
 export * from './logger'
 export * from './types'
 export * from './streams'

packages/common/src/blocks.ts → packages/common/src/ipld.ts

@@ -4,8 +4,9 @@ import * as rawCodec from 'multiformats/codecs/raw'
 import { sha256 } from 'multiformats/hashes/sha2'
 import * as mf from 'multiformats'
 import * as cborCodec from '@ipld/dag-cbor'
+import { check, schema } from '.'
 
-export const valueToIpldBlock = async (data: unknown) => {
+export const dataToCborBlock = async (data: unknown) => {
   return Block.encode({
     value: data,
     codec: cborCodec,
@@ -28,22 +29,19 @@ export const sha256RawToCid = (hash: Uint8Array): CID => {
   return CID.createV1(rawCodec.code, digest)
 }
 
-export const cidForData = async (data: unknown): Promise<CID> => {
-  const block = await valueToIpldBlock(data)
+export const cidForCbor = async (data: unknown): Promise<CID> => {
+  const block = await dataToCborBlock(data)
   return block.cid
 }
 
-export const valueToIpldBytes = (value: unknown): Uint8Array => {
-  return cborCodec.encode(value)
-}
-
-export const ipldBytesToValue = (bytes: Uint8Array) => {
-  return cborCodec.decode(bytes)
-}
+export const cborEncode = cborCodec.encode
+export const cborDecode = cborCodec.decode
 
-export const ipldBytesToRecord = (bytes: Uint8Array): object => {
-  const val = ipldBytesToValue(bytes)
-  if (typeof val !== 'object' || val === null) {
+export const cborBytesToRecord = (
+  bytes: Uint8Array,
+): Record<string, unknown> => {
+  const val = cborDecode(bytes)
+  if (!check.is(val, schema.record)) {
     throw new Error(`Expected object, got: ${val}`)
   }
   return val

packages/pds/src/api/app/bsky/notification/list.ts

@@ -91,7 +91,7 @@ export default function (server: Server, ctx: AppContext) {
         },
         reason: notif.reason,
         reasonSubject: notif.reasonSubject || undefined,
-        record: common.ipldBytesToRecord(notif.recordBytes),
+        record: common.cborBytesToRecord(notif.recordBytes),
         isRead: notif.indexedAt <= user.lastSeenNotifs,
         indexedAt: notif.indexedAt,
       }))

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

@@ -1,7 +1,7 @@
 import { InvalidRequestError } from '@atproto/xrpc-server'
 import * as crypto from '@atproto/crypto'
 import * as handleLib from '@atproto/handle'
-import { cidForData } from '@atproto/common'
+import { cidForCbor } from '@atproto/common'
 import { Server, APP_BSKY_SYSTEM } from '../../../lexicon'
 import { countAll } from '../../../db/util'
 import * as lex from '../../../lexicon/lexicons'
@@ -151,7 +151,7 @@ export default function (server: Server, ctx: AppContext) {
       await repoTxn.createRepo(did, [write], now)
       await repoTxn.indexWrites([write], now)
 
-      const declarationCid = await cidForData(declaration)
+      const declarationCid = await cidForCbor(declaration)
       const access = ctx.auth.createAccessToken(did)
       const refresh = ctx.auth.createRefreshToken(did)
       await ctx.services.auth(dbTxn).grantRefreshToken(refresh.payload)

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

@@ -1,5 +1,5 @@
 import { CID } from 'multiformats/cid'
-import { Repo } from '@atproto/repo'
+import * as repo from '@atproto/repo'
 import { InvalidRequestError } from '@atproto/xrpc-server'
 import { Server } from '../../../lexicon'
 import SqlRepoStorage from '../../../sql-repo-storage'
@@ -45,13 +45,12 @@ export default function (server: Server, ctx: AppContext) {
   server.com.atproto.sync.getRepo(async ({ params }) => {
     const { did, from = null } = params
     const storage = new SqlRepoStorage(ctx.db, did)
-    const root = await storage.getHead()
-    if (root === null) {
+    const head = await storage.getHead()
+    if (head === null) {
       throw new InvalidRequestError(`Could not find repo for DID: ${did}`)
     }
-    const repo = await Repo.load(storage, root)
     const fromCid = from ? CID.parse(from) : null
-    const diff = await repo.getDiff(fromCid)
+    const diff = await repo.getDiff(storage, head, fromCid)
     return {
       encoding: 'application/cbor',
       body: Buffer.from(diff),
@@ -61,15 +60,35 @@ export default function (server: Server, ctx: AppContext) {
   server.com.atproto.sync.getCheckout(async ({ params }) => {
     const { did } = params
     const storage = new SqlRepoStorage(ctx.db, did)
-    const commit = params.commit ? CID.parse(params.commit) : undefined
-    const repo = await Repo.load(storage, commit)
-    const checkout = await repo.getCheckout()
+    const commit = params.commit
+      ? CID.parse(params.commit)
+      : await storage.getHead()
+    if (!commit) {
+      throw new InvalidRequestError(`Could not find repo for DID: ${did}`)
+    }
+    const checkout = await repo.getCheckout(storage, commit)
     return {
       encoding: 'application/cbor',
       body: Buffer.from(checkout),
     }
   })
 
+  server.com.atproto.sync.getRecord(async ({ params }) => {
+    const { did, collection, rkey } = params
+    const storage = new SqlRepoStorage(ctx.db, did)
+    const commit = params.commit
+      ? CID.parse(params.commit)
+      : await storage.getHead()
+    if (!commit) {
+      throw new InvalidRequestError(`Could not find repo for DID: ${did}`)
+    }
+    const proof = await repo.getRecords(storage, commit, [{ collection, rkey }])
+    return {
+      encoding: 'application/cbor',
+      body: Buffer.from(proof),
+    }
+  })
+
   server.com.atproto.sync.updateRepo(async () => {
     throw new InvalidRequestError('Not implemented')
   })

packages/pds/src/lexicon/index.ts

@@ -35,6 +35,7 @@ import * as ComAtprotoSessionRefresh from './types/com/atproto/session/refresh'
 import * as ComAtprotoSyncGetCheckout from './types/com/atproto/sync/getCheckout'
 import * as ComAtprotoSyncGetCommitPath from './types/com/atproto/sync/getCommitPath'
 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 ComAtprotoSyncUpdateRepo from './types/com/atproto/sync/updateRepo'
 import * as AppBskyActorCreateScene from './types/app/bsky/actor/createScene'
@@ -409,6 +410,13 @@ export class SyncNS {
     return this._server.xrpc.method(nsid, cfg)
   }
 
+  getRecord<AV extends AuthVerifier>(
+    cfg: ConfigOf<AV, ComAtprotoSyncGetRecord.Handler<ExtractAuth<AV>>>,
+  ) {
+    const nsid = 'com.atproto.sync.getRecord' // @ts-ignore
+    return this._server.xrpc.method(nsid, cfg)
+  }
+
   getRepo<AV extends AuthVerifier>(
     cfg: ConfigOf<AV, ComAtprotoSyncGetRepo.Handler<ExtractAuth<AV>>>,
   ) {

packages/pds/src/lexicon/lexicons.ts

@@ -1332,6 +1332,40 @@ export const schemaDict = {
       },
     },
   },
+  ComAtprotoSyncGetRecord: {
+    lexicon: 1,
+    id: 'com.atproto.sync.getRecord',
+    defs: {
+      main: {
+        type: 'query',
+        description:
+          'Gets blocks needed for existence or non-existence of record.',
+        parameters: {
+          type: 'params',
+          required: ['did', 'collection', 'rkey'],
+          properties: {
+            did: {
+              type: 'string',
+              description: 'The DID of the repo.',
+            },
+            collection: {
+              type: 'string',
+            },
+            rkey: {
+              type: 'string',
+            },
+            commit: {
+              type: 'string',
+              description: 'An optional past commit CID.',
+            },
+          },
+        },
+        output: {
+          encoding: 'application/cbor',
+        },
+      },
+    },
+  },
   ComAtprotoSyncGetRepo: {
     lexicon: 1,
     id: 'com.atproto.sync.getRepo',
@@ -3607,6 +3641,7 @@ export const ids = {
   ComAtprotoSyncGetCheckout: 'com.atproto.sync.getCheckout',
   ComAtprotoSyncGetCommitPath: 'com.atproto.sync.getCommitPath',
   ComAtprotoSyncGetHead: 'com.atproto.sync.getHead',
+  ComAtprotoSyncGetRecord: 'com.atproto.sync.getRecord',
   ComAtprotoSyncGetRepo: 'com.atproto.sync.getRepo',
   ComAtprotoSyncUpdateRepo: 'com.atproto.sync.updateRepo',
   AppBskyActorCreateScene: 'app.bsky.actor.createScene',

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

@@ -0,0 +1,37 @@
+/**
+ * GENERATED CODE - DO NOT MODIFY
+ */
+import express from 'express'
+import stream from 'stream'
+import { HandlerAuth } from '@atproto/xrpc-server'
+
+export interface QueryParams {
+  /** The DID of the repo. */
+  did: string
+  collection: string
+  rkey: string
+  /** An optional past commit CID. */
+  commit?: string
+}
+
+export type InputSchema = undefined
+export type HandlerInput = undefined
+
+export interface HandlerSuccess {
+  encoding: 'application/cbor'
+  body: Uint8Array | stream.Readable
+}
+
+export interface HandlerError {
+  status: number
+  message?: string
+}
+
+export type HandlerOutput = HandlerError | HandlerSuccess
+export type Handler<HA extends HandlerAuth = never> = (ctx: {
+  auth: HA
+  params: QueryParams
+  input: HandlerInput
+  req: express.Request
+  res: express.Response
+}) => Promise<HandlerOutput> | HandlerOutput