feat: migrated to redis

Author: alexcambose

apps/insights/package.json

@@ -28,10 +28,12 @@
     "@pythnetwork/known-publishers": "workspace:*",
     "@react-hookz/web": "catalog:",
     "@solana/web3.js": "catalog:",
+    "async-cache-dedupe": "^3.0.0",
     "bs58": "catalog:",
     "clsx": "catalog:",
     "cryptocurrency-icons": "catalog:",
     "dnum": "catalog:",
+    "ioredis": "^5.7.0",
     "lightweight-charts": "catalog:",
     "motion": "catalog:",
     "next": "catalog:",

apps/insights/src/app/price-feeds/[slug]/layout.ts

@@ -2,7 +2,7 @@ import type { Metadata } from "next";
 import { notFound } from "next/navigation";
 import type { ReactNode } from "react";
 
-import { getFeedsCached } from "../../../server/pyth";
+import { getFeeds } from "../../../server/pyth/get-feeds";
 import { Cluster } from "../../../services/pyth";
 
 export { PriceFeedLayout as default } from "../../../components/PriceFeed/layout";
@@ -20,7 +20,7 @@ export const generateMetadata = async ({
 }: Props): Promise<Metadata> => {
   const [{ slug }, feeds] = await Promise.all([
     params,
-    getFeedsCached(Cluster.Pythnet),
+    getFeeds(Cluster.Pythnet),
   ]);
   const symbol = decodeURIComponent(slug);
   const feed = feeds.find((item) => item.symbol === symbol);

apps/insights/src/components/Overview/index.tsx

@@ -9,7 +9,7 @@ import PriceFeedsLight from "./price-feeds-light.svg";
 import PublishersDark from "./publishers-dark.svg";
 import PublishersLight from "./publishers-light.svg";
 import { TabList } from "./tab-list";
-import { getFeedsCached } from "../../server/pyth";
+import { getFeeds } from "../../server/pyth/get-feeds";
 import { Cluster } from "../../services/pyth";
 import {
   totalVolumeTraded,
@@ -24,7 +24,7 @@ import { FormattedDate } from "../FormattedDate";
 import { FormattedNumber } from "../FormattedNumber";
 
 export const Overview = async () => {
-  const priceFeeds = await getFeedsCached(Cluster.Pythnet);
+  const priceFeeds = await getFeeds(Cluster.Pythnet);
   const today = new Date();
   const feedCounts = [
     ...activeFeeds.map(({ date, numFeeds }) => ({

apps/insights/src/components/PriceFeed/get-feed.tsx

@@ -1,10 +1,10 @@
 import { notFound } from "next/navigation";
 
-import { getFeedsCached } from "../../server/pyth";
+import { getFeeds } from "../../server/pyth/get-feeds";
 import { Cluster } from "../../services/pyth";
 
 export const getFeed = async (params: Promise<{ slug: string }>) => {
-  const [{ slug }, feeds] = await Promise.all([params, getFeedsCached(Cluster.Pythnet)]);
+  const [{ slug }, feeds] = await Promise.all([params, getFeeds(Cluster.Pythnet)]);
   const symbol = decodeURIComponent(slug);
   return {
     feeds,

apps/insights/src/components/PriceFeed/publishers.tsx

@@ -2,7 +2,8 @@ import { lookup as lookupPublisher } from "@pythnetwork/known-publishers";
 import { notFound } from "next/navigation";
 
 import { getRankingsBySymbolCached } from '../../server/clickhouse';
-import { getFeedsCached, getPublishersForFeedCached } from "../../server/pyth";
+import { getPublishersForFeedCached } from "../../server/pyth";
+import { getFeeds } from "../../server/pyth/get-feeds";
 import {
   Cluster,
   ClusterToName,
@@ -29,8 +30,8 @@ export const Publishers = async ({ params }: Props) => {
     pythnetPublishers,
     pythtestConformancePublishers,
   ] = await Promise.all([
-    getFeedsCached(Cluster.Pythnet),
-    getFeedsCached(Cluster.PythtestConformance),
+    getFeeds(Cluster.Pythnet),
+    getFeeds(Cluster.PythtestConformance),
     getPublishers(Cluster.Pythnet, symbol),
     getPublishers(Cluster.PythtestConformance, symbol),
   ]);
@@ -96,7 +97,7 @@ const getPublishers = async (cluster: Cluster, symbol: string) => {
   ]);
 
   return (
-    publishers?.map((publisher) => {
+    publishers.map((publisher) => {
       const ranking = rankings.find(
         (ranking) =>
           ranking.publisher === publisher &&
@@ -110,6 +111,6 @@ const getPublishers = async (cluster: Cluster, symbol: string) => {
         cluster,
         knownPublisher: lookupPublisher(publisher),
       };
-    }) ?? []
+    })
   );
 };

apps/insights/src/components/PriceFeeds/index.tsx

@@ -19,7 +19,7 @@ import { AssetClassTable } from "./asset-class-table";
 import { ComingSoonList } from "./coming-soon-list";
 import styles from "./index.module.scss";
 import { PriceFeedsCard } from "./price-feeds-card";
-import { getFeedsCached } from "../../server/pyth";
+import { getFeeds } from "../../server/pyth/get-feeds";
 import { Cluster } from "../../services/pyth";
 import { priceFeeds as priceFeedsStaticConfig } from "../../static-data/price-feeds";
 import { activeChains } from "../../static-data/stats";
@@ -290,7 +290,7 @@ const FeaturedFeedsCard = <T extends ElementType>({
 );
 
 const getPriceFeeds = async () => {
-  const priceFeeds = await getFeedsCached(Cluster.Pythnet);
+  const priceFeeds = await getFeeds(Cluster.Pythnet);
   const activeFeeds = priceFeeds.filter((feed) => isActive(feed));
   const comingSoon = priceFeeds.filter((feed) => !isActive(feed));
   return { activeFeeds, comingSoon };

apps/insights/src/components/Root/index.tsx

@@ -11,7 +11,7 @@ import {
 } from "../../config/server";
 import { LivePriceDataProvider } from "../../hooks/use-live-price-data";
 import { getPublishersCached } from '../../server/clickhouse';
-import { getFeedsCached } from '../../server/pyth';
+import { getFeeds } from '../../server/pyth/get-feeds';
 import { Cluster } from "../../services/pyth";
 import { PriceFeedIcon } from "../PriceFeedIcon";
 import { PublisherIcon } from "../PublisherIcon";
@@ -75,7 +75,7 @@ const getPublishersForSearchDialog = async (cluster: Cluster) => {
 };
 
 const getFeedsForSearchDialog = async (cluster: Cluster) => {
-  const feeds = await getFeedsCached(cluster);
+  const feeds = await getFeeds(cluster);
 
   return feeds.map((feed) => ({
     symbol: feed.symbol,

apps/insights/src/config/server.ts

@@ -2,6 +2,7 @@
 // and load all env variables.
 /* eslint-disable n/no-process-env */
 
+import { Redis } from 'ioredis';
 import "server-only";
 
 /**
@@ -56,3 +57,21 @@ export const SOLANA_RPC =
 
 export const ENABLE_ACCESSIBILITY_REPORTING =
   !IS_PRODUCTION_SERVER && !process.env.DISABLE_ACCESSIBILITY_REPORTING;
+
+
+export async function getRedis(): Promise<Redis> {
+  const host = process.env.REDIS_HOST;
+  const port = process.env.REDIS_PORT;
+  const password = process.env.REDIS_PASSWORD;
+  if (!host || !port) {
+    throw new Error('REDIS_HOST, and REDIS_PORT must be set');
+  }
+  const client = new Redis({ 
+    username: 'default',
+    password: password ?? '',
+    host,
+    port: Number.parseInt(port),
+  });
+  await client.set('test2', Math.random().toString());
+  return client;
+}

apps/insights/src/server/clickhouse.ts

@@ -1,55 +1,45 @@
 
 import { getPublisherAverageScoreHistory, getPublisherRankingHistory, getPublishers, getRankingsByPublisher, getRankingsBySymbol } from "../services/clickhouse";
 import { Cluster } from "../services/pyth";
-import { createChunkedCacheFetcher, fetchAllChunks, timeFunction } from '../utils/cache';
+import { createChunkedCacheFetcher, fetchAllChunks } from '../utils/cache';
 
 
 const _getRankingsBySymbol = createChunkedCacheFetcher(async (symbol: string) => {
   return getRankingsBySymbol(symbol);
 }, 'getRankingsBySymbol');
 
 export const getRankingsBySymbolCached = async (symbol: string) => {
-  return timeFunction(async () => {
     return fetchAllChunks<ReturnType<typeof getRankingsBySymbol>, [string]>(_getRankingsBySymbol, symbol);
-  }, 'getRankingsBySymbolCached');
 };
 
 const _getRankingsByPublisher = createChunkedCacheFetcher(async (publisherKey: string) => {
   return getRankingsByPublisher(publisherKey);
 }, 'getRankingsByPublisher');
 
 export const getRankingsByPublisherCached = async (publisherKey: string) => {
-  return timeFunction(async () => {
     return fetchAllChunks<ReturnType<typeof getRankingsByPublisher>, [string]>(_getRankingsByPublisher, publisherKey);
-  }, 'getRankingsByPublisherCached');
 };
 
 const _getPublisherAverageScoreHistory = createChunkedCacheFetcher(async (cluster: Cluster, key: string) => {
   return getPublisherAverageScoreHistory(cluster, key);
 }, 'getPublisherAverageScoreHistory');
 
 export const getPublisherAverageScoreHistoryCached = async (cluster: Cluster, key: string) => {
-  return timeFunction(async () => {
     return fetchAllChunks<ReturnType<typeof getPublisherAverageScoreHistory>, [Cluster, string]>(_getPublisherAverageScoreHistory, cluster, key);
-  }, 'getPublisherAverageScoreHistoryCached');
 };
 
 const _getPublisherRankingHistory = createChunkedCacheFetcher(async (cluster: Cluster, key: string) => {
   return getPublisherRankingHistory(cluster, key);
 }, 'getPublisherRankingHistory');
 
 export const getPublisherRankingHistoryCached = async (cluster: Cluster, key: string) => {
-  return timeFunction(async () => {
     return fetchAllChunks<ReturnType<typeof getPublisherRankingHistory>, [Cluster, string]>(_getPublisherRankingHistory, cluster, key);
-  }, 'getPublisherRankingHistoryCached');
 };
 
 const _getPublishers = createChunkedCacheFetcher(async (cluster: Cluster) => {
   return getPublishers(cluster);
 }, 'getPublishers');
 
 export const getPublishersCached = async (cluster: Cluster) => {
-  return timeFunction(async () => {
     return fetchAllChunks<ReturnType<typeof getPublishers>, [Cluster]>(_getPublishers, cluster);
-  }, 'getPublishersCached');
 }

apps/insights/src/server/pyth.ts

@@ -1,112 +1,24 @@
-import type { PythHttpClientResult } from '@pythnetwork/client/lib/PythHttpClient';
-import type { z } from 'zod';
+import { Cluster, priceFeedsSchema } from "../services/pyth";
+import { getFeeds } from './pyth/get-feeds';
+import { getPublishersForCluster } from './pyth/get-publishers-for-cluster';
 
-import { Cluster, clients, priceFeedsSchema } from "../services/pyth";
-import { createChunkedCacheFetcher, fetchAllChunks, timeFunction } from '../utils/cache';
-
-
-type CacheEntry = {
-  data?: PythHttpClientResult;
-  timestamp?: number;
-  promise?: Promise<PythHttpClientResult>;
-};
-
-const dataCache = new Map<Cluster, CacheEntry>();
-const CACHE_EXPIRY_MS = 24 * 60 * 60; // 1 day in seconds
-
-const getDataCached = async (cluster: Cluster) => {
-  const now = Date.now();
-  const cached = dataCache.get(cluster);
-  
-  // Check if cache exists and is not expired
-  if (cached?.data && cached.timestamp && (now - cached.timestamp) < CACHE_EXPIRY_MS * 1000) {
-    return cached.data;
-  }
-  
-  // Check if there's already a pending request
-  if (cached?.promise) {
-    return cached.promise;
-  }
-  
-  // eslint-disable-next-line no-console
-  console.log('fetching fresh FULL data');
-  
-  // Create a new promise for the request
-  const promise = clients[cluster].getData().then((data) => {
-    // Store the result in cache
-    dataCache.set(cluster, { data, timestamp: now });
-    return data;
-  });
-  
-  // Store the promise in cache to prevent duplicate requests
-  dataCache.set(cluster, { promise });
-  
-  return promise;
-};
-
-const fetchFeeds = createChunkedCacheFetcher(async (cluster: Cluster) => {
-  const unfilteredData = await getDataCached(cluster);
-  // eslint-disable-next-line no-console
-  console.log('fetchFeeds called');
-  const filteredData = unfilteredData.symbols
-      .filter(
-        (symbol) =>
-          unfilteredData.productFromSymbol.get(symbol)?.display_symbol !== undefined
-      )
-      .map((symbol) => ({
-        symbol,
-        product: unfilteredData.productFromSymbol.get(symbol),
-        price: {
-          ...unfilteredData.productPrice.get(symbol),
-          priceComponents:
-            unfilteredData.productPrice
-              .get(symbol)
-              ?.priceComponents.map(({ publisher }) => ({
-                publisher: publisher.toBase58(),
-              })) ?? [],
-        },
-      }));
-  const parsedData = priceFeedsSchema.parse(filteredData);
-  return parsedData;
-}, 'getFeeds');
-
-const fetchPublishers = createChunkedCacheFetcher(async (cluster: Cluster) => {
-  const data = await getDataCached(cluster);
-  // eslint-disable-next-line no-console
-  console.log('fetchPublishers called');
-  const result: Record<string, string[]> = {};
-  for (const key of data.productPrice.keys()) {
-    const price = data.productPrice.get(key);
-    result[key] = price?.priceComponents.map(({ publisher }) => publisher.toBase58()) ?? [];
-  }
-  return result;
-}, 'fetchPublishers');
-
-export const getFeedsCached = async (cluster: Cluster) => {
-  return timeFunction(async () => {
-    return fetchAllChunks<z.infer<typeof priceFeedsSchema>, [Cluster]>(fetchFeeds,  cluster);
-  }, 'getFeedsCached');
-};
-
-export const getPublishersForFeedCached = async (cluster: Cluster, symbol: string) => {
-  const data = await timeFunction(async () => {
-    return fetchAllChunks<Record<string, string[]>, [Cluster]>(fetchPublishers, cluster);
-  }, 'getPublishersForFeedCached');
-  return data[symbol]
-};
+// Convenience helpers matching your previous functions
+export async function getPublishersForFeedCached(
+  cluster: Cluster,
+  symbol: string
+) {
+  const map = await getPublishersForCluster(cluster);
+  return map[symbol] ?? [];
+}
 
-export const getFeedsForPublisherCached = async (
+export async function getFeedsForPublisherCached(
   cluster: Cluster,
   publisher: string
-) => {
-  const data = await timeFunction(async () => {
-    return getFeedsCached(cluster);
-  }, 'getFeedsForPublisherCached');
+) {
+  const data = await getFeeds(cluster);
   return priceFeedsSchema.parse(
     data.filter(({ price }) =>
-      price.priceComponents.some(
-        (component) => component.publisher.toString() === publisher
-      )
+      price.priceComponents.some((c) => c.publisher === publisher)
     )
   );
-};
+}