Firehose Identity event (#2208)

* add new identity event * add note for tombstone * send identity evts * add emission of identity evts * emit handle event on activate account & fix subscribeRepos * add time to evt * update indexer for tests * rm logs
bluesky-social · Feb 21, 2024 · 1a12c7e · 1a12c7e
1 parent 30b05a7
commit 1a12c7e
Show file tree

Hide file tree

Showing 23 changed files with 306 additions and 41 deletions.
diff --git a/lexicons/com/atproto/sync/subscribeRepos.json b/lexicons/com/atproto/sync/subscribeRepos.json
@@ -17,7 +17,14 @@
       "message": {
         "schema": {
           "type": "union",
-          "refs": ["#commit", "#handle", "#migrate", "#tombstone", "#info"]
+          "refs": [
+            "#commit",
+            "#identity",
+            "#handle",
+            "#migrate",
+            "#tombstone",
+            "#info"
+          ]
         }
       },
       "errors": [
@@ -104,9 +111,19 @@
         }
       }
     },
+    "identity": {
+      "type": "object",
+      "description": "Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache.",
+      "required": ["seq", "did", "time"],
+      "properties": {
+        "seq": { "type": "integer" },
+        "did": { "type": "string", "format": "did" },
+        "time": { "type": "string", "format": "datetime" }
+      }
+    },
     "handle": {
       "type": "object",
-      "description": "Represents an update of the account's handle, or transition to/from invalid state.",
+      "description": "Represents an update of the account's handle, or transition to/from invalid state. NOTE: Will be deprecated in favor of #identity.",
       "required": ["seq", "did", "handle", "time"],
       "properties": {
         "seq": { "type": "integer" },
@@ -117,7 +134,7 @@
     },
     "migrate": {
       "type": "object",
-      "description": "Represents an account moving from one PDS instance to another. NOTE: not implemented; full account migration may introduce a new message instead.",
+      "description": "Represents an account moving from one PDS instance to another. NOTE: not implemented; account migration uses #identity instead",
       "required": ["seq", "did", "migrateTo", "time"],
       "nullable": ["migrateTo"],
       "properties": {
@@ -129,7 +146,7 @@
     },
     "tombstone": {
       "type": "object",
-      "description": "Indicates that an account has been deleted.",
+      "description": "Indicates that an account has been deleted. NOTE: may be deprecated in favor of #identity or a future #account event",
       "required": ["seq", "did", "time"],
       "properties": {
         "seq": { "type": "integer" },

diff --git a/packages/api/src/client/lexicons.ts b/packages/api/src/client/lexicons.ts
@@ -4579,6 +4579,7 @@ export const schemaDict = {
             type: 'union',
             refs: [
               'lex:com.atproto.sync.subscribeRepos#commit',
+              'lex:com.atproto.sync.subscribeRepos#identity',
               'lex:com.atproto.sync.subscribeRepos#handle',
               'lex:com.atproto.sync.subscribeRepos#migrate',
               'lex:com.atproto.sync.subscribeRepos#tombstone',
@@ -4685,10 +4686,29 @@ export const schemaDict = {
           },
         },
       },
+      identity: {
+        type: 'object',
+        description:
+          "Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache.",
+        required: ['seq', 'did', 'time'],
+        properties: {
+          seq: {
+            type: 'integer',
+          },
+          did: {
+            type: 'string',
+            format: 'did',
+          },
+          time: {
+            type: 'string',
+            format: 'datetime',
+          },
+        },
+      },
       handle: {
         type: 'object',
         description:
-          "Represents an update of the account's handle, or transition to/from invalid state.",
+          "Represents an update of the account's handle, or transition to/from invalid state. NOTE: Will be deprecated in favor of #identity.",
         required: ['seq', 'did', 'handle', 'time'],
         properties: {
           seq: {
@@ -4711,7 +4731,7 @@ export const schemaDict = {
       migrate: {
         type: 'object',
         description:
-          'Represents an account moving from one PDS instance to another. NOTE: not implemented; full account migration may introduce a new message instead.',
+          'Represents an account moving from one PDS instance to another. NOTE: not implemented; account migration uses #identity instead',
         required: ['seq', 'did', 'migrateTo', 'time'],
         nullable: ['migrateTo'],
         properties: {
@@ -4733,7 +4753,8 @@ export const schemaDict = {
       },
       tombstone: {
         type: 'object',
-        description: 'Indicates that an account has been deleted.',
+        description:
+          'Indicates that an account has been deleted. NOTE: may be deprecated in favor of #identity or a future #account event',
         required: ['seq', 'did', 'time'],
         properties: {
           seq: {
@@ -7087,7 +7108,8 @@ export const schemaDict = {
             },
             tags: {
               type: 'array',
-              description: 'Additional non-inline tags describing this post.',
+              description:
+                'Additional hashtags, in addition to any included in post text and facets.',
               maxLength: 8,
               items: {
                 type: 'string',

diff --git a/packages/api/src/client/types/app/bsky/feed/post.ts b/packages/api/src/client/types/app/bsky/feed/post.ts
@@ -32,7 +32,7 @@ export interface Record {
   labels?:
     | ComAtprotoLabelDefs.SelfLabels
     | { $type: string; [k: string]: unknown }
-  /** Additional non-inline tags describing this post. */
+  /** Additional hashtags, in addition to any included in post text and facets. */
   tags?: string[]
   /** Client-declared timestamp when this post was originally created. */
   createdAt: string

diff --git a/packages/api/src/client/types/com/atproto/sync/subscribeRepos.ts b/packages/api/src/client/types/com/atproto/sync/subscribeRepos.ts
@@ -46,7 +46,27 @@ export function validateCommit(v: unknown): ValidationResult {
   return lexicons.validate('com.atproto.sync.subscribeRepos#commit', v)
 }
 
-/** Represents an update of the account's handle, or transition to/from invalid state. */
+/** Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache. */
+export interface Identity {
+  seq: number
+  did: string
+  time: string
+  [k: string]: unknown
+}
+
+export function isIdentity(v: unknown): v is Identity {
+  return (
+    isObj(v) &&
+    hasProp(v, '$type') &&
+    v.$type === 'com.atproto.sync.subscribeRepos#identity'
+  )
+}
+
+export function validateIdentity(v: unknown): ValidationResult {
+  return lexicons.validate('com.atproto.sync.subscribeRepos#identity', v)
+}
+
+/** Represents an update of the account's handle, or transition to/from invalid state. NOTE: Will be deprecated in favor of #identity. */
 export interface Handle {
   seq: number
   did: string
@@ -67,7 +87,7 @@ export function validateHandle(v: unknown): ValidationResult {
   return lexicons.validate('com.atproto.sync.subscribeRepos#handle', v)
 }
 
-/** Represents an account moving from one PDS instance to another. NOTE: not implemented; full account migration may introduce a new message instead. */
+/** Represents an account moving from one PDS instance to another. NOTE: not implemented; account migration uses #identity instead */
 export interface Migrate {
   seq: number
   did: string
@@ -88,7 +108,7 @@ export function validateMigrate(v: unknown): ValidationResult {
   return lexicons.validate('com.atproto.sync.subscribeRepos#migrate', v)
 }
 
-/** Indicates that an account has been deleted. */
+/** Indicates that an account has been deleted. NOTE: may be deprecated in favor of #identity or a future #account event */
 export interface Tombstone {
   seq: number
   did: string

diff --git a/packages/bsky/src/indexer/subscription.ts b/packages/bsky/src/indexer/subscription.ts
@@ -145,6 +145,8 @@ export class IndexerSubscription {
         await this.handleCommit(msg)
       } else if (message.isHandle(msg)) {
         await this.handleUpdateHandle(msg)
+      } else if (message.isIdentity(msg)) {
+        await this.handleIdentityEvt(msg)
       } else if (message.isTombstone(msg)) {
         await this.handleTombstone(msg)
       } else if (message.isMigrate(msg)) {
@@ -244,6 +246,10 @@ export class IndexerSubscription {
     await this.indexingSvc.indexHandle(msg.did, msg.time, true)
   }
 
+  private async handleIdentityEvt(msg: message.Identity) {
+    await this.indexingSvc.indexHandle(msg.did, msg.time, true)
+  }
+
   private async handleTombstone(msg: message.Tombstone) {
     await this.indexingSvc.tombstoneActor(msg.did)
   }

diff --git a/packages/bsky/src/ingester/subscription.ts b/packages/bsky/src/ingester/subscription.ts
@@ -156,6 +156,8 @@ function getMessageDetails(msg: Message):
     return { seq: msg.seq, repo: msg.repo, message: msg }
   } else if (message.isHandle(msg)) {
     return { seq: msg.seq, repo: msg.did, message: msg }
+  } else if (message.isIdentity(msg)) {
+    return { seq: msg.seq, repo: msg.did, message: msg }
   } else if (message.isMigrate(msg)) {
     return { seq: msg.seq, repo: msg.did, message: msg }
   } else if (message.isTombstone(msg)) {

diff --git a/packages/bsky/src/lexicon/lexicons.ts b/packages/bsky/src/lexicon/lexicons.ts
@@ -4579,6 +4579,7 @@ export const schemaDict = {
             type: 'union',
             refs: [
               'lex:com.atproto.sync.subscribeRepos#commit',
+              'lex:com.atproto.sync.subscribeRepos#identity',
               'lex:com.atproto.sync.subscribeRepos#handle',
               'lex:com.atproto.sync.subscribeRepos#migrate',
               'lex:com.atproto.sync.subscribeRepos#tombstone',
@@ -4685,10 +4686,29 @@ export const schemaDict = {
           },
         },
       },
+      identity: {
+        type: 'object',
+        description:
+          "Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache.",
+        required: ['seq', 'did', 'time'],
+        properties: {
+          seq: {
+            type: 'integer',
+          },
+          did: {
+            type: 'string',
+            format: 'did',
+          },
+          time: {
+            type: 'string',
+            format: 'datetime',
+          },
+        },
+      },
       handle: {
         type: 'object',
         description:
-          "Represents an update of the account's handle, or transition to/from invalid state.",
+          "Represents an update of the account's handle, or transition to/from invalid state. NOTE: Will be deprecated in favor of #identity.",
         required: ['seq', 'did', 'handle', 'time'],
         properties: {
           seq: {
@@ -4711,7 +4731,7 @@ export const schemaDict = {
       migrate: {
         type: 'object',
         description:
-          'Represents an account moving from one PDS instance to another. NOTE: not implemented; full account migration may introduce a new message instead.',
+          'Represents an account moving from one PDS instance to another. NOTE: not implemented; account migration uses #identity instead',
         required: ['seq', 'did', 'migrateTo', 'time'],
         nullable: ['migrateTo'],
         properties: {
@@ -4733,7 +4753,8 @@ export const schemaDict = {
       },
       tombstone: {
         type: 'object',
-        description: 'Indicates that an account has been deleted.',
+        description:
+          'Indicates that an account has been deleted. NOTE: may be deprecated in favor of #identity or a future #account event',
         required: ['seq', 'did', 'time'],
         properties: {
           seq: {
@@ -7087,7 +7108,8 @@ export const schemaDict = {
             },
             tags: {
               type: 'array',
-              description: 'Additional non-inline tags describing this post.',
+              description:
+                'Additional hashtags, in addition to any included in post text and facets.',
               maxLength: 8,
               items: {
                 type: 'string',

diff --git a/packages/bsky/src/lexicon/types/app/bsky/feed/post.ts b/packages/bsky/src/lexicon/types/app/bsky/feed/post.ts
@@ -32,7 +32,7 @@ export interface Record {
   labels?:
     | ComAtprotoLabelDefs.SelfLabels
     | { $type: string; [k: string]: unknown }
-  /** Additional non-inline tags describing this post. */
+  /** Additional hashtags, in addition to any included in post text and facets. */
   tags?: string[]
   /** Client-declared timestamp when this post was originally created. */
   createdAt: string

diff --git a/packages/bsky/src/lexicon/types/com/atproto/sync/subscribeRepos.ts b/packages/bsky/src/lexicon/types/com/atproto/sync/subscribeRepos.ts
@@ -15,6 +15,7 @@ export interface QueryParams {
 
 export type OutputSchema =
   | Commit
+  | Identity
   | Handle
   | Migrate
   | Tombstone
@@ -71,7 +72,27 @@ export function validateCommit(v: unknown): ValidationResult {
   return lexicons.validate('com.atproto.sync.subscribeRepos#commit', v)
 }
 
-/** Represents an update of the account's handle, or transition to/from invalid state. */
+/** Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache. */
+export interface Identity {
+  seq: number
+  did: string
+  time: string
+  [k: string]: unknown
+}
+
+export function isIdentity(v: unknown): v is Identity {
+  return (
+    isObj(v) &&
+    hasProp(v, '$type') &&
+    v.$type === 'com.atproto.sync.subscribeRepos#identity'
+  )
+}
+
+export function validateIdentity(v: unknown): ValidationResult {
+  return lexicons.validate('com.atproto.sync.subscribeRepos#identity', v)
+}
+
+/** Represents an update of the account's handle, or transition to/from invalid state. NOTE: Will be deprecated in favor of #identity. */
 export interface Handle {
   seq: number
   did: string
@@ -92,7 +113,7 @@ export function validateHandle(v: unknown): ValidationResult {
   return lexicons.validate('com.atproto.sync.subscribeRepos#handle', v)
 }
 
-/** Represents an account moving from one PDS instance to another. NOTE: not implemented; full account migration may introduce a new message instead. */
+/** Represents an account moving from one PDS instance to another. NOTE: not implemented; account migration uses #identity instead */
 export interface Migrate {
   seq: number
   did: string
@@ -113,7 +134,7 @@ export function validateMigrate(v: unknown): ValidationResult {
   return lexicons.validate('com.atproto.sync.subscribeRepos#migrate', v)
 }
 
-/** Indicates that an account has been deleted. */
+/** Indicates that an account has been deleted. NOTE: may be deprecated in favor of #identity or a future #account event */
 export interface Tombstone {
   seq: number
   did: string