Feat: [WIP] - API

.gitignore

@@ -1,3 +1,5 @@
 *.log
 node_modules/
 .cache/
+.env
+api/data/

api/README

@@ -0,0 +1,124 @@
+# API Prototype PR
+
+## Overview
+
+This is a prototype PR of an API that manages cryptocurrency metadata, current prices, and timeseries data. The API uses:
+
+- **PM2 Cluster** for load balancing and auto-restart.
+- **Local Caching** for metadata (managed in a dedicated module).
+- **Elasticsearch** for historical timeseries data.
+- **Redis** for current price data.
+
+## Features
+
+- **PM2 Cluster Setup**  
+  The API is designed to run in a PM2 cluster. Instance 0 is responsible for fetching the initial metadata refresh (if enabled) while all instances serve the endpoints.
+
+- **Local Cache Management**  
+  Metadata is cached locally in a file (e.g., `api/data/metadata.json`). The cache is maintained via a dedicated module (`cache/metadataCache.js`).  
+  **Note:** The metadata refresh cron is now managed externally (e.g., via Coolify) using a separate script.
+
+- **Normalized PID Mapping**  
+  Coin IDs (pids) are normalized using the helper `normalizeCoinId` (from `utils/index.js`) and a mapping is maintained between the input pid and its normalized value. A maximum of 100 pids per request is allowed.
+
+## Endpoints
+
+### 1. Metadata Route
+
+**GET** `/api/coins/metadata?pid=bitcoin,ethereum`
+
+- **Description:**  
+  Retrieves metadata for the specified coin IDs.
+
+- **Process:**  
+  - The API checks the local cache for each pid (using normalized values).
+  - If metadata for a pid is missing, it returns `null` (no fallback to Elasticsearch is performed).
+
+### 2. Current Price Route
+
+**GET** `/api/coins/current?pid=bitcoin,ethereum&withMetadata=true&withTTL=true`
+
+- **Description:**  
+  Retrieves the current price of the specified coins from Redis.
+
+- **Options:**  
+  - `withTTL=true`: Includes TTL information.
+
+- **Process:**  
+  - The API queries Redis using keys formatted as `price_<normalized_pid>`.
+
+### 3. Timeseries Route
+
+**GET** `/api/coins/timeseries?pid=bitcoin,ethereum&startDate=2023-01-01&endDate=2023-01-31&scale=5m`  
+or  
+**GET** `/api/coins/timeseries?pid=bitcoin,ethereum&timestamp=1673000000`  
+
+- **Description:**  
+  Retrieves historical timeseries data for the specified coin IDs.
+
+- **Modes:**  
+  - **Timestamp mode:**  
+    If a `timestamp` (in seconds) is provided, the API returns, for each coin, the document whose `ts` is closest to that timestamp.
+
+  - **Range mode:**  
+    If `startDate` and `endDate` are provided, the API returns all matching records (grouped by coin).
+
+  - **Scale mode:**  
+    If a `scale` (e.g., "1m", "5m", "1h", "1d") is provided along with `startDate` and `endDate`, the API performs an aggregation using a `date_histogram`.  
+    The response includes, for each coin, an array of objects per interval containing:
+    - `timestamp` (in seconds)
+    - `avg_price`
+    - `min_price`
+    - `max_price`
+    - `count`
+
+  - **Note:**  
+    A check is performed to avoid exceeding the maximum number of buckets (e.g., 5,000). If too many buckets would be created, an error is returned.
+
+### 4. Earliest Timestamp Route
+
+**GET** `/api/coins/earliest?pid=bitcoin,ethereum`
+
+- **Description:**  
+  Retrieves, for each coin, the earliest available timeseries record (i.e., the record with the lowest timestamp).
+
+- **Process:**  
+  - For each provided pid, the API queries Elasticsearch for the document with the smallest `ts`.
+
+### 5. Percentage Change Route
+
+**GET** `/api/coins/percentage-change?pid=bitcoin,ethereum&timestamp=1656944730&period=3600&lookForward=true`
+
+- **Description:**  
+  Calculates the percentage change in price for each coin between two points in time.
+
+- **Parameters:**  
+  - `timestamp`: A reference timestamp in seconds (t₀).
+  - `period`: The period (in seconds) over which to calculate the change.
+  - `lookForward`:  
+    - If `true`, the change is measured from t₀ to t₀ + period (i.e., forward in time).  
+    - If `false` or not provided, the change is measured from t₀ to t₀ - period (i.e., backward in time).
+
+- **Process:**  
+  - For each coin, the API retrieves the document closest to t₀ and another at t₀ + period (or t₀ - period), then calculates the percentage change as:  
+    ```
+    ((price_t1 - price_t0) / price_t0) * 100
+    ```
+
+## Setup Instructions
+
+1. Install dependencies:
+   ```sh
+   npm install
+   ```
+
+2. Start the PM2 cluster:
+  ```sh
+   pm2 start ecosystem.config.js
+  ```
+
+## Env Dependencies
+  ```
+  REDIS_CLIENT_CONFIG=
+  COINS_ELASTICSEARCH_CONFIG=
+  ```

api/cache/metadataCache.js

@@ -0,0 +1,114 @@
+const fs = require('fs');
+const path = require('path');
+const { getClient } = require('../../db/elastic');
+
+let localMetadataCache = {};
+
+const DATA_DIR = path.join(__dirname, '../data');
+const METADATA_JSON_PATH = path.join(DATA_DIR, 'metadata.json');
+
+function ensureCacheFileExists() {
+  if (!fs.existsSync(DATA_DIR)) {
+    fs.mkdirSync(DATA_DIR, { recursive: true });
+    console.log("Data directory created:", DATA_DIR);
+  }
+  if (!fs.existsSync(METADATA_JSON_PATH)) {
+    fs.writeFileSync(METADATA_JSON_PATH, '{}', 'utf8');
+    console.log("Created empty metadata file:", METADATA_JSON_PATH);
+  }
+}
+
+function loadLocalMetadataCache() {
+  try {
+    ensureCacheFileExists();
+    const data = fs.readFileSync(METADATA_JSON_PATH, 'utf8');
+    localMetadataCache = JSON.parse(data);
+    console.log("Local metadata cache reloaded.");
+  } catch (err) {
+    console.error("Error loading metadata cache file:", err);
+    localMetadataCache = {};
+  }
+}
+
+function saveLocalMetadataCache() {
+  ensureCacheFileExists();
+  fs.writeFile(
+    METADATA_JSON_PATH,
+    JSON.stringify(localMetadataCache, null, 2),
+    'utf8',
+    (err) => {
+      if (err) {
+        console.error("Failed to write local metadata cache file:", err);
+      } else {
+        console.log(`Cache saved to ${METADATA_JSON_PATH}`);
+      }
+    }
+  );
+}
+
+
+function initCacheWatcher() {
+  loadLocalMetadataCache();
+
+  fs.watchFile(METADATA_JSON_PATH, { interval: 1000 }, (curr, prev) => {
+    if (curr.mtimeMs !== prev.mtimeMs) {
+      console.log("Detected change in metadata file, reloading cache...");
+      loadLocalMetadataCache();
+    }
+  });
+}
+
+function getLocalCache() {
+  return localMetadataCache;
+}
+
+function updateLocalCache(newCache) {
+  localMetadataCache = newCache;
+  saveLocalMetadataCache();
+}
+
+async function getAllMetadata() {
+  const client = getClient();
+  if (!client) {
+    throw new Error('No Elasticsearch client available.');
+  }
+
+  const pageSize = 100000;
+  const allDocs = [];
+
+  let response = await client.search({
+    index: 'coins-metadata',
+    scroll: '1m',
+    size: pageSize,
+    body: { query: { match_all: {} } }
+  });
+
+  while (response.hits && response.hits.hits.length > 0) {
+    allDocs.push(...response.hits.hits.map(hit => hit._source));
+    response = await client.scroll({
+      scroll_id: response._scroll_id,
+      scroll: '1m'
+    });
+  }
+  return allDocs;
+}
+
+async function refreshLocalMetadataFromES() {
+  const allData = await getAllMetadata();
+  console.log(`Fetched ${allData.length} metadata documents from ES.`);
+
+  const newCache = { ...localMetadataCache };
+  for (const item of allData) {
+    if (item.pid) {
+      newCache[item.pid.toLowerCase()] = item;
+    }
+  }
+  updateLocalCache(newCache);
+}
+
+module.exports = {
+  initCacheWatcher,
+  getLocalCache,
+  updateLocalCache,
+  refreshLocalMetadataFromES
+};

api/controllers/coins.js

@@ -0,0 +1,119 @@
+const { getCurrentCoin } = require('../services/getCurrentCoin')
+const { getCoinsEarliest } = require('../services/getEarliestCoin')
+const { getCoinMetadata } = require('../services/getMetadataCoin')
+const { getTimeseries } = require('../services/getTimeseriesCoin')
+const { getPercentageChange } = require('../services/getPercentageChangeCoin')
+
+function sendResponse(res, status, data) {
+  res.statusCode = status;
+  res.header('Content-Type', 'application/json');
+  res.send(JSON.stringify(data));
+}
+
+/**
+ * GET /metadata
+ * Example: /api/coins/metadata?pid=bitcoin,ethereum
+ * This endpoint retrieves metadata from the local cache or Elasticsearch for the given pid(s).
+ */
+async function getCoinsMetadata(req, res) {
+  try {
+    const { pid } = req.query;
+    if (!pid) {
+      return sendResponse(res, 400, { error: "Missing 'pid' query parameter." });
+    }
+    const data = getCoinMetadata({ pid });
+    return sendResponse(res, 200, data);
+  } catch (error) {
+    return sendResponse(res, 400, { error: error.message || 'Internal Server Error' });
+  }
+}
+
+/**
+ * GET /current
+ * Example: /api/coins/current?pid=bitcoin,ethereum&withMetadata=true&withTTL=true
+ * This endpoint retrieves current coin data from Redis for the given pid(s).
+ * It can also include metadata or TTL if specified in the query.
+ */
+async function getCoinsCurrent(req, res) {
+  try {
+    const { pid, withTTL = 'false' } = req.query;
+    if (!pid) {
+      return sendResponse(res, 400, { error: "Missing 'pid' query parameter." });
+    }
+    const includeTTL = (withTTL === 'true');
+    const data = await getCurrentCoin(pid, { withTTL: includeTTL });
+    return sendResponse(res, 200, data);
+  } catch (error) {
+    return sendResponse(res, 400, { error: error.message || 'Internal Server Error' });
+  }
+}
+
+/**
+ * GET /timeseries
+ * Example: /api/coins/timeseries?pid=bitcoin&startDate=2023-01-01&endDate=2023-01-31
+ * Or:      /api/coins/timeseries?pid=bitcoin,ethereum&timestamp=1673000000
+ *
+ * pid is required.
+ * If 'timestamp' is provided, for each pid we return the single document whose 'ts' is closest to that timestamp.
+ * If 'timestamp' is not provided, we fetch all documents (optionally constrained by startDate/endDate), grouped by pid and sorted by ts ascending.
+ */
+async function getCoinsTimeseries(req, res) {
+  try {
+    const { pid, startDate, endDate, timestamp, scale } = req.query;
+    if (!pid) {
+      return sendResponse(res, 400, { error: "Missing 'pid' query parameter." });
+    }
+    const data = await getTimeseries({ pid, startDate, endDate, timestamp, scale });
+    return sendResponse(res, 200, data);
+  } catch (error) {
+    return sendResponse(res, 400, { error: error.message || 'Internal Server Error' });
+  }
+}
+
+/**
+ * GET /earliest
+ * Example: /api/coins/earliest?pid=bitcoin,ethereum
+ * Returns, for each coin, the earliest record (i.e. with the lowest timestamp).
+ */
+async function getCoinsFirstTimestamp(req, res) {
+  try {
+    const { pid } = req.query;
+    if (!pid) {
+      return sendResponse(res, 400, { error: "Missing 'pid' query parameter." });
+    }
+    const data = await getCoinsEarliest({ pid });
+    return sendResponse(res, 200, data);
+  } catch (error) {
+    return sendResponse(res, 400, { error: error.message || 'Internal Server Error' });
+  }
+}
+
+/**
+ * GET /percentage-change
+ * Example: /api/coins/percentage-change?pid=bitcoin,ethereum&timestamp=1656944730&period=3600&lookForward=true
+ * Parameters:
+ * - pid: list of coins.
+ * - timestamp: a reference timestamp in seconds.
+ * - period: period (in seconds) to calculate the change.
+ * - lookForward: if true, change is from t0 to t0 + period; otherwise from t0 to t0 - period.
+ */
+async function getCoinsPercentageChange(req, res) {
+  try {
+    const { pid, timestamp, period, lookForward } = req.query;
+    if (!pid || !timestamp || !period) {
+      return sendResponse(res, 400, { error: "Missing required query parameters: pid, timestamp, period." });
+    }
+    const data = await getPercentageChange({ pid, timestamp, period, lookForward });
+    return sendResponse(res, 200, data);
+  } catch (error) {
+    return sendResponse(res, 400, { error: error.message || 'Internal Server Error' });
+  }
+}
+
+module.exports = {
+  getCoinsMetadata,
+  getCoinsCurrent,
+  getCoinsTimeseries,
+  getCoinsFirstTimestamp,
+  getCoinsPercentageChange
+};