bunch of fixes

Dockerfile

···
 
 COPY *.py ./
 
-ENV METRICS_PORT=6009
-ENV EMBEDDING_DEVICE=cuda
-ENV NVIDIA_VISIBLE_DEVICES=all
-ENV NVIDIA_DRIVER_CAPABILITIES=compute,utility
-
-EXPOSE 6009
-
 ENTRYPOINT ["/usr/bin/dumb-init", "--"]
 CMD ["uv", "run", "main.py"]

config.py

···
 
 
 class Config(BaseSettings):
-    ch_host: str = "localhost"
-    ch_port: int = 8123
-    ch_user: str = "default"
-    ch_pass: str = "clickhouse"
+    clickhouse_host: str = "localhost"
+    clickhouse_port: int = 8123
+    clickhouse_user: str = "default"
+    clickhouse_pass: str = "clickhouse"
 
     batch_size: int = 1000
 

docker-compose.yaml

···
     hostname: zookeeper
     container_name: zookeeper
     ports:
-      - "2181:2181"
+      - "127.0.0.1:2181:2181"
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
···
     depends_on:
       - zookeeper
     ports:
-      - "9092:9092"
-      - "9101:9101"
+      - "127.0.0.1:9092:9092"
+      - "127.0.0.1:9101:9101"
     environment:
       KAFKA_BROKER_ID: 1
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
···
       kafka:
         condition: service_healthy
     ports:
-      - "2480:2480"
-      - "6010:6010"
+      - "127.0.0.1:2480:2480"
+      - "127.0.0.1:6010:6010"
     environment:
       TAP_BIND: ":2480"
       TAP_FULL_NETWORK: true
       TAP_DISABLE_ACKS: false
       TAP_COLLECTION_FILTERS: "app.bsky.graph.follow"
       TAP_METRICS_LISTEN: ":6010"
+      TAP_RESYNC_PARALLELISM: 50
     volumes:
       - tap-data:/data
     restart: unless-stopped
···
         condition: service_healthy
     ports:
       # metrics port
-      - "6011:6009"
+      - "127.0.0.1:6011:6009"
     command: ["tap-mode"]
     environment:
       ATKAFKA_TAP_HOST: "ws://tap:2480"
···
   indexer:
     build: .
     ports:
-      - "6012:6009"
+      - "127.0.0.1:8050:8050"
     depends_on:
       clickhouse:
         condition: service_healthy
···
         condition: service_healthy
     command: ["uv", "run", "indexer.py"]
     environment:
-      KAFKA_BOOTSTRAP_SERVERS: "localhost:9092"
+      KAFKA_BOOTSTRAP_SERVERS: '["kafka:29092"]'
       KAFKA_INPUT_TOPIC: "tap-events"
       CLICKHOUSE_HOST: "clickhouse"
     restart: unless-stopped

indexer.py

···
                 column_names=["did", "subject", "uri", "created_at"],
             )
 
+            self.client.insert(
+                "follows_reverse",
+                follows_data,
+                column_names=["did", "subject", "uri", "created_at"],
+            )
+
             status = "ok"
         except Exception as e:
             # TODO: handle errors gracefully
···
 
             if op.action == "update":
                 kind = "update"
-                status = "ok"
-                return
             elif op.action == "create":
                 kind = "create"
-
                 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
                 )
-
                 self.indexer.insert_follow(follow)
             else:
                 kind = "delete"
-
-                kind = "delete"
-
                 unfollow = Unfollow(uri=op.uri, created_at=datetime.now())
-
                 self.indexer.insert_unfollow(unfollow)
 
             status = "ok"
···
     )
 
     indexer = FollowIndexer(
-        clickhouse_host=ch_host or CONFIG.ch_host,
-        clickhouse_port=ch_port or CONFIG.ch_port,
-        clickhouse_user=ch_user or CONFIG.ch_user,
-        clickhouse_pass=ch_pass or CONFIG.ch_pass,
+        clickhouse_host=ch_host or CONFIG.clickhouse_host,
+        clickhouse_port=ch_port or CONFIG.clickhouse_port,
+        clickhouse_user=ch_user or CONFIG.clickhouse_user,
+        clickhouse_pass=ch_pass or CONFIG.clickhouse_pass,
         batch_size=batch_size or CONFIG.batch_size,
     )
     indexer.init_schema()