Fix Alpha Vantage rate-limit failures in the daily alerts worker

apps/alerts/src/alerts/sp500-close.test.ts

@@ -1,3 +1,4 @@
+import { AlphaVantageClient } from "@repo/alpha-vantage";
 import type { DailyBar, DailyTimeSeries } from "@repo/alpha-vantage/types";
 import type { Logger } from "@repo/logger";
 import { beforeEach, describe, expect, it, vi } from "vitest";
@@ -68,7 +69,11 @@ const env = {
   ALPHA_VANTAGE_API_KEY: "test-key",
 };
 
-const run = () => sp500Close.run({ env, logger: createMockLogger() });
+const run = () => {
+  const logger = createMockLogger();
+  const alphaVantage = new AlphaVantageClient("test-key", logger);
+  return sp500Close.run({ env, logger, alphaVantage });
+};
 
 describe("sp500Close", () => {
   beforeEach(() => {

apps/alerts/src/alerts/sp500-close.ts

@@ -1,5 +1,3 @@
-import { AlphaVantageClient } from "@repo/alpha-vantage";
-
 import { fetchSp500, SP500_SYMBOL } from "./sp500";
 import type { Alert } from "./types";
 
@@ -10,11 +8,11 @@ import type { Alert } from "./types";
 const sp500Close: Alert = {
   name: "sp500-close",
   cron: "0 8 * * *", // reuses the existing daily 08:00 UTC trigger
-  run: async ({ env, logger }) => {
-    const client = new AlphaVantageClient(env.ALPHA_VANTAGE_API_KEY, logger);
+  run: async ({ logger, alphaVantage }) => {
     // Free daily `compact` closes + the free weekly series for the ATH baseline
-    // (daily `full` history is a premium feature). See `fetchSp500`.
-    const { bars, priorHigh } = await fetchSp500(client);
+    // (daily `full` history is a premium feature). See `fetchSp500`. The client
+    // is shared across alerts so the fetch de-duplicates within the run.
+    const { bars, priorHigh } = await fetchSp500(alphaVantage);
     if (bars.length === 0) {
       logger.warn("no SPY bars returned", { symbol: SP500_SYMBOL });
       return null;

apps/alerts/src/alerts/sp500-drawdown.test.ts

@@ -1,3 +1,4 @@
+import { AlphaVantageClient } from "@repo/alpha-vantage";
 import type { DailyBar, DailyTimeSeries } from "@repo/alpha-vantage/types";
 import type { Logger } from "@repo/logger";
 import { beforeEach, describe, expect, it, vi } from "vitest";
@@ -74,7 +75,11 @@ const env = {
   ALPHA_VANTAGE_API_KEY: "test-key",
 };
 
-const run = () => sp500Drawdown.run({ env, logger: createMockLogger() });
+const run = () => {
+  const logger = createMockLogger();
+  const alphaVantage = new AlphaVantageClient("test-key", logger);
+  return sp500Drawdown.run({ env, logger, alphaVantage });
+};
 
 describe("sp500Drawdown", () => {
   beforeEach(() => {

apps/alerts/src/alerts/sp500-drawdown.ts

@@ -1,4 +1,3 @@
-import { AlphaVantageClient } from "@repo/alpha-vantage";
 import type { DailyBar } from "@repo/alpha-vantage/types";
 
 import { fetchSp500 } from "./sp500";
@@ -67,11 +66,11 @@ const formatPercent = (fraction: number): string =>
 const sp500Drawdown: Alert = {
   name: "sp500-drawdown",
   cron: "0 8 * * *", // reuses the existing daily 08:00 UTC trigger
-  run: async ({ env, logger }) => {
-    const client = new AlphaVantageClient(env.ALPHA_VANTAGE_API_KEY, logger);
+  run: async ({ logger, alphaVantage }) => {
     // Free daily `compact` closes + the free weekly series for the ATH baseline
-    // (daily `full` history is a premium feature). See `fetchSp500`.
-    const { bars, priorHigh } = await fetchSp500(client);
+    // (daily `full` history is a premium feature). See `fetchSp500`. The client
+    // is shared across alerts so the fetch de-duplicates within the run.
+    const { bars, priorHigh } = await fetchSp500(alphaVantage);
 
     if (bars.length < 2) {
       logger.warn("need at least 2 SPY bars for drawdown", {

apps/alerts/src/alerts/types.ts

@@ -1,10 +1,17 @@
+import type { AlphaVantageClient } from "@repo/alpha-vantage";
 import type { Logger } from "@repo/logger";
 
 import type { Env } from "../types/env";
 
 type AlertContext = {
   env: Env;
   logger: Logger;
+  /**
+   * Shared Alpha Vantage client for the run. Constructed once by the scheduled
+   * handler so alerts fetching the same series de-duplicate to one HTTP request
+   * (see `AlphaVantageClient`), staying under the free-tier burst limit.
+   */
+  alphaVantage: AlphaVantageClient;
 };
 
 type Alert = {

apps/alerts/src/scheduled.test.ts

@@ -27,8 +27,15 @@ const spySeries = {
   ],
 };
 
+// Every `new AlphaVantageClient(...)` pushes here so the test can assert the
+// handler builds exactly one client and shares it across the due alerts.
+const clientInstances: object[] = [];
+
 vi.mock("@repo/alpha-vantage", () => ({
   AlphaVantageClient: class {
+    constructor() {
+      clientInstances.push(this);
+    }
     getDailyTimeSeries = () => Promise.resolve(spySeries);
     getWeeklyTimeSeries = () => Promise.resolve(spySeries);
   },
@@ -52,6 +59,7 @@ describe("handleScheduled", () => {
   beforeEach(() => {
     sendMessage.mockReset();
     sendMessage.mockResolvedValue({ ok: true });
+    clientInstances.length = 0;
   });
 
   it("sends the alert messages when the daily cron fires", async () => {
@@ -71,6 +79,14 @@ describe("handleScheduled", () => {
     }
   });
 
+  it("builds one Alpha Vantage client and shares it across the due alerts", async () => {
+    await handleScheduled(makeEvent("0 8 * * *"), env, ctx);
+
+    // A single shared client means both SPY alerts de-duplicate their fetches
+    // instead of racing the free-tier burst limit with separate clients.
+    expect(clientInstances).toHaveLength(1);
+  });
+
   it("does not send anything when no alert matches the fired cron", async () => {
     await handleScheduled(makeEvent("*/5 * * * *"), env, ctx);
 

apps/alerts/src/scheduled.ts

@@ -1,3 +1,4 @@
+import { AlphaVantageClient } from "@repo/alpha-vantage";
 import { Logger } from "@repo/logger";
 import { TelegramService } from "@repo/telegram";
 
@@ -23,9 +24,16 @@ const handleScheduled = async (
 
   logger.info("running alerts", { cron: event.cron, count: due.length });
 
+  // One client for the whole run so alerts fetching the same series share a
+  // single request instead of racing the free-tier burst limit.
+  const alphaVantage = new AlphaVantageClient(
+    parsed.ALPHA_VANTAGE_API_KEY,
+    logger
+  );
+
   const results = await Promise.allSettled(
     due.map(async (alert) => {
-      const message = await alert.run({ env: parsed, logger });
+      const message = await alert.run({ env: parsed, logger, alphaVantage });
       if (message === null) {
         logger.info("alert produced no message", { alert: alert.name });
         return;

packages/alpha-vantage/src/alpha-vantage.test.ts

@@ -76,7 +76,11 @@ describe("AlphaVantageClient", () => {
 
   beforeEach(() => {
     mockFetch.mockReset();
-    client = new AlphaVantageClient("test-key", createMockLogger());
+    // Throttling off by default so most tests stay fast; the serialization test
+    // opts back in with its own interval.
+    client = new AlphaVantageClient("test-key", createMockLogger(), {
+      minRequestIntervalMs: 0,
+    });
   });
 
   describe("getDailyTimeSeries", () => {
@@ -183,4 +187,106 @@ describe("AlphaVantageClient", () => {
       );
     });
   });
+
+  describe("request de-duplication", () => {
+    it("collapses concurrent identical requests into a single fetch", async () => {
+      mockFetch.mockResolvedValue(createJsonResponse(dailySuccessBody));
+
+      const [a, b] = await Promise.all([
+        client.getDailyTimeSeries("SPY"),
+        client.getDailyTimeSeries("SPY"),
+      ]);
+
+      expect(mockFetch).toHaveBeenCalledTimes(1);
+      expect(a).toBe(b);
+    });
+
+    it("treats omitted and explicit compact output size as one request", async () => {
+      mockFetch.mockResolvedValue(createJsonResponse(dailySuccessBody));
+
+      await Promise.all([
+        client.getDailyTimeSeries("SPY"),
+        client.getDailyTimeSeries("SPY", { outputSize: "compact" }),
+      ]);
+
+      expect(mockFetch).toHaveBeenCalledTimes(1);
+    });
+
+    it("de-duplicates daily and weekly concurrently as separate requests", async () => {
+      // A fresh Response per call (bodies are single-read) with the shape the
+      // called endpoint expects.
+      mockFetch.mockImplementation((url: unknown) =>
+        Promise.resolve(
+          createJsonResponse(
+            String(url).includes("TIME_SERIES_WEEKLY")
+              ? weeklySuccessBody
+              : dailySuccessBody
+          )
+        )
+      );
+
+      await Promise.all([
+        client.getDailyTimeSeries("SPY"),
+        client.getWeeklyTimeSeries("SPY"),
+      ]);
+
+      expect(mockFetch).toHaveBeenCalledTimes(2);
+    });
+
+    it("does not cache results — a call after the first settles refetches", async () => {
+      // Fresh Response per call so both sequential reads succeed.
+      mockFetch.mockImplementation(() =>
+        Promise.resolve(createJsonResponse(dailySuccessBody))
+      );
+
+      await client.getDailyTimeSeries("SPY");
+      await client.getDailyTimeSeries("SPY");
+
+      // In-flight de-dup only: sequential calls each hit the network so the data
+      // stays fresh for a reused client.
+      expect(mockFetch).toHaveBeenCalledTimes(2);
+    });
+
+    it("does not let a failed request poison a later identical call", async () => {
+      mockFetch.mockRejectedValueOnce(new Error("network down"));
+      mockFetch.mockResolvedValueOnce(createJsonResponse(dailySuccessBody));
+
+      await expect(client.getDailyTimeSeries("SPY")).rejects.toThrow(
+        "network down"
+      );
+
+      const series = await client.getDailyTimeSeries("SPY");
+
+      expect(series.symbol).toBe("SPY");
+      expect(mockFetch).toHaveBeenCalledTimes(2);
+    });
+
+    it("serializes distinct requests spaced by the configured interval", async () => {
+      const interval = 50;
+      const throttled = new AlphaVantageClient("test-key", createMockLogger(), {
+        minRequestIntervalMs: interval,
+      });
+      const starts: number[] = [];
+      mockFetch.mockImplementation((url: unknown) => {
+        starts.push(Date.now());
+        return Promise.resolve(
+          createJsonResponse(
+            String(url).includes("TIME_SERIES_WEEKLY")
+              ? weeklySuccessBody
+              : dailySuccessBody
+          )
+        );
+      });
+
+      // Fired in the same tick — without throttling both would hit the network
+      // simultaneously and risk the per-second burst limit.
+      await Promise.all([
+        throttled.getDailyTimeSeries("SPY"),
+        throttled.getWeeklyTimeSeries("SPY"),
+      ]);
+
+      expect(starts).toHaveLength(2);
+      expect(starts[1] - starts[0]).toBeGreaterThanOrEqual(interval - 5);
+    });
+  });
 });