kafka produce

coins/package.json

@@ -53,6 +53,7 @@
     "@sentry/tracing": "^6.19.7",
     "@solana/web3.js": "^1.73.3",
     "@supercharge/promise-pool": "^2.1.0",
+    "ajv": "^6.12.6",
     "axios": "^1.6.7",
     "crypto-js": "^4.2.0",
     "dayjs": "^1.11.10",
@@ -62,6 +63,7 @@
     "graphql": "^16.6.0",
     "graphql-request": "^5.1.0",
     "ioredis": "^5.3.2",
+    "kafkajs": "^2.2.4",
     "node-fetch": "^2.7.0",
     "p-limit": "^3.1.0",
     "path": "^0.12.7",

coins/src/adapters/bridges/index.ts

@@ -38,8 +38,8 @@ import fraxtal from "./fraxtal";
 import symbiosis from "./symbiosis";
 import fuel from "./fuel";
 import zircuit from "./zircuit";
-import morph from './morph'
-import aptos from './aptosFa'
+import morph from "./morph";
+import aptos from "./aptosFa";
 
 export type Token =
   | {
@@ -108,11 +108,12 @@ export const bridges = [
   fuel,
   zircuit,
   morph,
-  aptos
+  aptos,
 ].map(normalizeBridgeResults) as Bridge[];
 
 import { batchGet, batchWrite } from "../../utils/shared/dynamodb";
 import { getCurrentUnixTimestamp } from "../../utils/date";
+import produceKafkaTopics from "../../utils/coins3/produce";
 
 const craftToPK = (to: string) => (to.includes("#") ? to : `asset#${to}`);
 
@@ -146,7 +147,7 @@ async function _storeTokensOfBridge(bridge: Bridge) {
   const redirectMap: { [redirect: string]: string } = {};
 
   const toRecords = await batchGet(
-    unlisted.map((t) => ({
+    tokens.map((t) => ({
       PK: craftToPK(t.to),
       SK: 0,
     })),
@@ -173,9 +174,26 @@ async function _storeTokensOfBridge(bridge: Bridge) {
     if (record.price) toAddressToRecord[toPK] = record.PK;
   });
 
+  const writes2: any[] = [];
   const writes: any[] = [];
   await Promise.all(
-    unlisted.map(async (token) => {
+    tokens.map(async (token) => {
+      // coins 3
+      if ("decimals" in token && "symbol" in token) {
+        const toRecord = toRecords.find((r: any) => r.PK == token.to);
+        writes2.push({
+          PK: `asset#${token.from}`,
+          timestamp: toRecord.timestamp,
+          decimals: token.decimals,
+          symbol: token.symbol,
+          confidence: 0.97,
+          adapter: "bridges",
+          price: toRecord.price,
+        });
+      }
+
+      if (!unlisted.includes(token)) return;
+
       const finalPK = toAddressToRecord[craftToPK(token.to)];
       if (finalPK === undefined) return;
 
@@ -206,50 +224,8 @@ async function _storeTokensOfBridge(bridge: Bridge) {
     }),
   );
 
-  // const writes2: Coin[] = [];
-  // const data = await readCoins2(
-  //   tokens.map((t: Token) => ({
-  //     key: t.to.includes("coingecko#") ? t.to.replace("#", ":") : t.to,
-  //     timestamp: getCurrentUnixTimestamp(),
-  //   })),
-  // );
-  // tokens.map(async (token) => {
-  //   const to = token.to.includes("coingecko#")
-  //     ? token.to.replace("#", ":")
-  //     : token.to;
-  //   if (!(to in data)) return;
-  //   let PK: string = token.from.includes("coingecko#")
-  //     ? token.from.replace("#", ":")
-  //     : token.from.substring(token.from.indexOf("#") + 1);
-  //   const chain = PK.split(":")[0];
-  //   let decimals: number, symbol: string;
-  //   if ("getAllInfo" in token) {
-  //     try {
-  //       const newToken = await token.getAllInfo();
-  //       decimals = newToken.decimals;
-  //       symbol = newToken.symbol;
-  //     } catch (e) {
-  //       console.log("Skipping token", PK, e);
-  //       return;
-  //     }
-  //   } else {
-  //     decimals = token.decimals;
-  //     symbol = token.symbol;
-  //   }
-  //   writes2.push({
-  //     timestamp: getCurrentUnixTimestamp(),
-  //     price: data[to].price,
-  //     confidence: Math.min(data[to].confidence, 0.9),
-  //     key: PK,
-  //     chain,
-  //     adapter: "bridges",
-  //     symbol,
-  //     decimals,
-  //   });
-  // });
-
   await batchWrite(writes, true);
-  // await batchWrite2(writes2, true, undefined, `bridge index ${i}`);
+  await produceKafkaTopics(writes2);
   return tokens;
 }
 export async function storeTokens() {

coins/src/adapters/utils/database.ts

@@ -16,6 +16,7 @@ import { batchWrite2, translateItems } from "../../../coins2";
 const confidenceThreshold: number = 0.3;
 import pLimit from "p-limit";
 import { sliceIntoChunks } from "@defillama/sdk/build/util";
+import produceKafkaTopics from "../../utils/coins3/produce";
 
 const rateLimited = pLimit(10);
 process.env.tableName = "prod-coins-table";
@@ -396,6 +397,7 @@ export async function batchWriteWithAlerts(
   const filteredItems: AWS.DynamoDB.DocumentClient.PutItemInputAttributeMap[] =
     await checkMovement(items, previousItems);
   await batchWrite(filteredItems, failOnError);
+  await produceKafkaTopics(filteredItems.filter((i) => i.SK == 0) as any[]);
 }
 export async function batchWrite2WithAlerts(
   items: AWS.DynamoDB.DocumentClient.PutItemInputAttributeMap[],

coins/src/scripts/defiCoins.ts

@@ -1,7 +1,6 @@
 require("dotenv").config();
 import {
   batchWriteWithAlerts,
-  batchWrite2WithAlerts,
   filterWritesWithLowConfidence,
 } from "../adapters/utils/database";
 import { withTimeout } from "../../../defi/src/utils/shared/withTimeout";
@@ -53,9 +52,9 @@ async function storeDefiCoins() {
               true,
             ),
           ]);
-          await batchWrite2WithAlerts(
-            resultsWithoutDuplicates.slice(i, i + step),
-          );
+          // await batchWrite2WithAlerts(
+          //   resultsWithoutDuplicates.slice(i, i + step),
+          // );
         }
       } catch (e) {
         console.error(`ERROR: ${adapterKey} adapter failed ${e}`);

coins/src/utils/coins3/cli/consume.ts

@@ -0,0 +1,25 @@
+import { topics } from "../jsonValidation";
+import { getConsumer } from "../kafka";
+
+const run = async () => {
+  await Promise.all(
+    topics.map(async (topic) => {
+      const consumer = await getConsumer(topic);
+      await consumer.subscribe({ topic, fromBeginning: true });
+
+      await consumer.run({
+        eachMessage: async ({ topic, partition, message }: any) => {
+          console.log({
+            topic,
+            partition,
+            offset: message.offset,
+            value: message.value.toString(),
+          });
+        },
+      });
+    }),
+  );
+};
+
+run().catch(console.error);
+// ts-node coins/src/utils/coins3/cli/consume.ts

coins/src/utils/coins3/cli/newTopic.ts

@@ -0,0 +1,26 @@
+import { getKafka } from "../kafka";
+
+const createTopic = async () => {
+  const kafka = getKafka();
+  const admin = kafka.admin();
+  await admin.connect();
+
+  // Create a topic
+  await admin.createTopics({
+    topics: [
+      {
+        topic: "current", // Replace with your topic name
+        numPartitions: 1, // Number of partitions
+        replicationFactor: 1, // Replication factor
+      },
+    ],
+  });
+
+  console.log("Topic created successfully");
+
+  // Disconnect the admin client
+  await admin.disconnect();
+};
+
+createTopic().catch(console.error);
+// ts-node coins/src/utils/coins3/cli/newTopic.ts

coins/src/utils/coins3/jsonValidation.ts

@@ -0,0 +1,61 @@
+import Ajv from "ajv";
+
+type Schema = {
+  type: string;
+  properties: { [prop: string]: { type: string } };
+  required: string[];
+  additionalProperties: Boolean;
+};
+
+const schemas: { [topic: string]: Schema } = {
+  metadata: {
+    type: "object",
+    properties: {
+      decimals: { type: "integer" },
+      symbol: { type: "string" },
+      address: { type: "string" },
+      pid: { type: "string" },
+      chain: { type: "string" },
+      source: { type: "string" },
+    },
+    required: ["decimals", "symbol", "address", "pid", "chain"],
+    additionalProperties: false,
+  },
+  current: {
+    type: "object",
+    properties: {
+      pid: { type: "string" },
+      price: { type: "number" },
+      confidence: { type: "number" },
+      source: { type: "string" },
+    },
+    required: ["pid", "price", "confidence", "source"],
+    additionalProperties: false,
+  },
+  timeseries: {
+    type: "object",
+    properties: {
+      ts: { type: "integer" },
+      pid: { type: "string" },
+      price: { type: "number" },
+      confidence: { type: "number" },
+      source: { type: "string" },
+    },
+    required: ["ts", "pid", "price", "confidence", "source"],
+    additionalProperties: false,
+  },
+};
+
+const ajv = new Ajv();
+
+export function validate(data: object, topic: Topic) {
+  const comp = ajv.compile(schemas[topic]);
+  const valid = comp(data);
+  if (!valid) {
+    console.log(comp.errors);
+    throw new Error(`${topic} validation error`);
+  }
+}
+
+export const topics: string[] = Object.keys(schemas);
+export type Topic = keyof typeof schemas;