convert to work with tap

indexer.py

···
 
 from config import CONFIG
 from metrics import prom_metrics
-from models import AtKafkaEvent, Follow, FollowRecord, Unfollow
+from models import Follow, FollowRecord, TapEvent, Unfollow
 
 logging.basicConfig(
     level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
···
         kind = "unk"
 
         try:
-            evt = AtKafkaEvent.model_validate(message.value)
+            evt = TapEvent.model_validate(message.value)
 
-            if not evt.operation or evt.operation.collection != "app.bsky.graph.follow":
+            if not evt.record or evt.record.collection != "app.bsky.graph.follow":
                 kind = "skipped"
                 status = "ok"
                 return
 
-            op = evt.operation
+            op = evt.record
+            uri = f"at://{op.did}/{op.collection}/{op.rkey}"
 
             if op.action == "update":
                 kind = "update"
···
                 rec = FollowRecord.model_validate(op.record)
                 created_at = isoparse(rec.created_at)
                 follow = Follow(
-                    uri=op.uri, did=evt.did, subject=rec.subject, created_at=created_at
+                    uri=uri, did=op.did, subject=rec.subject, created_at=created_at
                 )
                 self.indexer.insert_follow(follow)
             else:
                 kind = "delete"
-                unfollow = Unfollow(uri=op.uri, created_at=datetime.now())
+
+                unfollow = Unfollow(uri=uri, created_at=datetime.now())
+
                 self.indexer.insert_unfollow(unfollow)
 
             status = "ok"

models.py

···
 from datetime import datetime
-from typing import Any, Dict, Optional, List
+from typing import Any, Dict, Optional
 
 from pydantic import BaseModel, Field
 
 
-class AtKafkaOp(BaseModel):
-    action: str
+class RecordEvent(BaseModel):
+    """
+    A model for record events that come from Tap, in Kafka mode
+    """
+
+    live: bool
+    did: str
+    rev: str
     collection: str
     rkey: str
-    uri: str
-    cid: str
-    path: str
+    action: str
     record: Optional[Dict[str, Any]]
+    cid: Optional[str]
 
 
-class AtKafkaIdentity(BaseModel):
-    seq: int
-    handle: str
+class IdentityEvent(BaseModel):
+    """
+    A model for identity events taht come from Tap, in Kafka mode
+    """
 
+    live: bool
+    handle: Optional[str]
+    is_active: bool
+    status: str
 
-class AtKafkaInfo(BaseModel):
-    name: str
-    message: Optional[str] = None
 
+class TapEvent(BaseModel):
+    """
+    The base model for events that come from Tap, in Kafka mode
+    """
 
-class AtKafkaAccount(BaseModel):
-    active: bool
-    seq: int
-    status: Optional[str] = None
+    id: int
+    type: str
+    record: RecordEvent
+    identity: IdentityEvent
 
 
-class DIDDocument(BaseModel):
-    context: Optional[List[Any]] = Field(None, alias="@context")
-    id: Optional[str] = None
-    also_known_as: Optional[List[Any]] = Field(None, alias="alsoKnownAs")
-    verification_method: Optional[List[Any]] = Field(None, alias="verificationMethod")
-    service: Optional[List[Any]] = None
-
-    class Config:
-        populate_by_name = True
-
-
-class ProfileViewDetailed(BaseModel):
-    did: str
-    handle: str
-    display_name: Optional[str] = Field(None, alias="displayName")
-    description: Optional[str] = None
-    avatar: Optional[str] = None
-    banner: Optional[str] = None
-    followers_count: Optional[int] = Field(None, alias="followersCount")
-    follows_count: Optional[int] = Field(None, alias="followsCount")
-    posts_count: Optional[int] = Field(None, alias="postsCount")
-    indexed_at: Optional[str] = Field(None, alias="indexedAt")
-    viewer: Optional[Dict[str, Any]] = None
-    labels: Optional[List[Any]] = None
-
-    class Config:
-        populate_by_name = True
-
-
-class EventMetadata(BaseModel):
-    did_document: Optional[DIDDocument] = Field(None, alias="didDocument")
-    pds_host: Optional[str] = Field(None, alias="pdsHost")
-    handle: Optional[str] = None
-    did_created_at: Optional[str] = Field(None, alias="didCreatedAt")
-    account_age: Optional[int] = Field(None, alias="accountAge")
-    profile: Optional[ProfileViewDetailed] = None
-
-    class Config:
-        populate_by_name = True
-
-
-class AtKafkaEvent(BaseModel):
-    did: str
-    timestamp: str
-    metadata: Optional[EventMetadata] = Field(None, alias="eventMetadata")
-    operation: Optional[AtKafkaOp] = None
-    account: Optional[AtKafkaAccount] = None
-    identity: Optional[AtKafkaIdentity] = None
-    info: Optional[AtKafkaInfo] = None
-
-    class Config:
-        populate_by_name = True
+# class AtKafkaOp(BaseModel):
+#     action: str
+#     collection: str
+#     rkey: str
+#     uri: str
+#     cid: str
+#     path: str
+#     record: Optional[Dict[str, Any]]
+#
+#
+# class AtKafkaIdentity(BaseModel):
+#     seq: int
+#     handle: str
+#
+#
+# class AtKafkaInfo(BaseModel):
+#     name: str
+#     message: Optional[str] = None
+#
+#
+# class AtKafkaAccount(BaseModel):
+#     active: bool
+#     seq: int
+#     status: Optional[str] = None
+#
+#
+# class DIDDocument(BaseModel):
+#     context: Optional[List[Any]] = Field(None, alias="@context")
+#     id: Optional[str] = None
+#     also_known_as: Optional[List[Any]] = Field(None, alias="alsoKnownAs")
+#     verification_method: Optional[List[Any]] = Field(None, alias="verificationMethod")
+#     service: Optional[List[Any]] = None
+#
+#     class Config:
+#         populate_by_name = True
+#
+#
+# class ProfileViewDetailed(BaseModel):
+#     did: str
+#     handle: str
+#     display_name: Optional[str] = Field(None, alias="displayName")
+#     description: Optional[str] = None
+#     avatar: Optional[str] = None
+#     banner: Optional[str] = None
+#     followers_count: Optional[int] = Field(None, alias="followersCount")
+#     follows_count: Optional[int] = Field(None, alias="followsCount")
+#     posts_count: Optional[int] = Field(None, alias="postsCount")
+#     indexed_at: Optional[str] = Field(None, alias="indexedAt")
+#     viewer: Optional[Dict[str, Any]] = None
+#     labels: Optional[List[Any]] = None
+#
+#     class Config:
+#         populate_by_name = True
+#
+#
+# class EventMetadata(BaseModel):
+#     did_document: Optional[DIDDocument] = Field(None, alias="didDocument")
+#     pds_host: Optional[str] = Field(None, alias="pdsHost")
+#     handle: Optional[str] = None
+#     did_created_at: Optional[str] = Field(None, alias="didCreatedAt")
+#     account_age: Optional[int] = Field(None, alias="accountAge")
+#     profile: Optional[ProfileViewDetailed] = None
+#
+#     class Config:
+#         populate_by_name = True
+#
+#
+# class AtKafkaEvent(BaseModel):
+#     did: str
+#     timestamp: str
+#     metadata: Optional[EventMetadata] = Field(None, alias="eventMetadata")
+#     operation: Optional[AtKafkaOp] = None
+#     account: Optional[AtKafkaAccount] = None
+#     identity: Optional[AtKafkaIdentity] = None
+#     info: Optional[AtKafkaInfo] = None
+#
+#     class Config:
+#         populate_by_name = True
 
 
 class FollowRecord(BaseModel):