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

@@ -112,6 +112,7 @@ export const bridges = [
 
 import { batchGet, batchWrite } from "../../utils/shared/dynamodb";
 import { getCurrentUnixTimestamp } from "../../utils/date";
+import produceKafkaTopics from "../../utils/coins3/produce";
 import { chainsThatShouldNotBeLowerCased } from "../../utils/shared/constants";
 
 const craftToPK = (to: string) => (to.includes("#") ? to : `asset#${to}`);
@@ -206,50 +207,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(writes, ["coins-metadata"]);
   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";
 import { chainsThatShouldNotBeLowerCased } from "../../utils/shared/constants";
 
 const rateLimited = pLimit(10);
@@ -412,6 +413,7 @@ export async function batchWriteWithAlerts(
   const filteredItems: AWS.DynamoDB.DocumentClient.PutItemInputAttributeMap[] =
     await checkMovement(items, previousItems);
   await batchWrite(filteredItems, failOnError);
+  await produceKafkaTopics(filteredItems as any[]);
 }
 export async function batchWrite2WithAlerts(
   items: AWS.DynamoDB.DocumentClient.PutItemInputAttributeMap[],

coins/src/scripts/coingecko.ts

@@ -13,6 +13,7 @@ import { CgEntry, Write } from "../adapters/utils/dbInterfaces";
 import { batchReadPostgres, getRedisConnection } from "../../coins2";
 import chainToCoingeckoId from "../../../common/chainToCoingeckoId";
 import { decimals, symbol } from "@defillama/sdk/build/erc20";
+import produceKafkaTopics, { Dynamo } from "../utils/coins3/produce";
 import { PublicKey } from "@solana/web3.js";
 import { getConnection } from "../adapters/solana/utils";
 import { chainsThatShouldNotBeLowerCased } from "../utils/shared/constants";
@@ -65,32 +66,43 @@ interface IdToSymbol {
 }
 
 async function storeCoinData(coinData: Write[]) {
-  return batchWrite(
-    coinData
-      .map((c) => ({
-        PK: c.PK,
-        SK: 0,
-        price: c.price,
-        mcap: c.mcap,
-        timestamp: c.timestamp,
-        symbol: c.symbol,
-        confidence: c.confidence,
-      }))
-      .filter((c: Write) => c.symbol != null),
-    false,
-  );
-}
-
-async function storeHistoricalCoinData(coinData: Write[]) {
-  return batchWrite(
-    coinData.map((c) => ({
-      SK: c.SK,
+  const items = coinData
+    .map((c) => ({
       PK: c.PK,
+      SK: 0,
       price: c.price,
+      mcap: c.mcap,
+      timestamp: c.timestamp,
+      symbol: c.symbol,
       confidence: c.confidence,
-    })),
-    false,
-  );
+    }))
+    .filter((c: Write) => c.symbol != null);
+  await Promise.all([
+    produceKafkaTopics(
+      items.map((i) => ({ adapter: "coingecko", decimals: 0, ...i } as Dynamo)),
+    ),
+    batchWrite(items, false),
+  ]);
+}
+
+async function storeHistoricalCoinData(coinData: Write[]) {
+  const items = coinData.map((c) => ({
+    SK: c.SK,
+    PK: c.PK,
+    price: c.price,
+    confidence: c.confidence,
+  }));
+  await Promise.all([
+    produceKafkaTopics(
+      items.map((i) => ({
+        adapter: "coingecko",
+        timestamp: i.SK,
+        ...i,
+      })) as Dynamo[],
+      ["coins-timeseries"],
+    ),
+    batchWrite(items, false),
+  ]);
 }
 
 let solanaTokens: Promise<any>;
@@ -276,6 +288,7 @@ async function getAndStoreCoins(coins: Coin[], rejected: Coin[]) {
     pricesAndMcaps[c.PK] = { price: c.price, mcap: c.mcap };
   });
 
+  const kafkaItems: any[] = [];
   await Promise.all(
     filteredCoins.map(async (coin) =>
       iterateOverPlatforms(
@@ -303,22 +316,30 @@ async function getAndStoreCoins(coins: Coin[], rejected: Coin[]) {
             return;
           }
 
-          await ddb.put({
+          const item = {
             PK,
             SK: 0,
             created,
             decimals,
             symbol,
             redirect: cgPK(coin.id),
             confidence: 0.99,
-          });
+          };
+          kafkaItems.push(item);
+          await ddb.put(item);
         },
         coinPlatformData,
       ),
     ),
   );
 
-  await deleteStaleKeysPromise;
+  await Promise.all([
+    produceKafkaTopics(
+      kafkaItems.map((i) => ({ adapter: "coingecko", ...i })),
+      ["coins-metadata"],
+    ),
+    deleteStaleKeysPromise,
+  ]);
 }
 
 const HOUR = 3600;
@@ -355,20 +376,27 @@ async function getAndStoreHourly(
     (c: any) => c.timestamp,
   );
 
-  await batchWrite(
-    coinData.prices
-      .filter((price) => {
-        const ts = toUNIXTimestamp(price[0]);
-        return !writtenTimestamps[ts];
-      })
-      .map((price) => ({
-        SK: toUNIXTimestamp(price[0]),
-        PK,
-        price: price[1],
-        confidence: 0.99,
-      })),
-    false,
-  );
+  const items = coinData.prices
+    .filter((price) => {
+      const ts = toUNIXTimestamp(price[0]);
+      return !writtenTimestamps[ts];
+    })
+    .map((price) => ({
+      SK: toUNIXTimestamp(price[0]),
+      PK,
+      price: price[1],
+      confidence: 0.99,
+    }));
+
+  await Promise.all([
+    produceKafkaTopics(
+      items.map(
+        (i) => ({ adapter: "coingecko", timestamp: i.SK, ...i }),
+        ["coins-timeseries"],
+      ),
+    ),
+    batchWrite(items, false),
+  ]);
 }
 
 async function fetchCoingeckoData(

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