fix(credits): read stream amounts from Superfluid periods instead of reconstructing them

backend/src/analytics-cron.ts

@@ -0,0 +1,64 @@
+import { Env } from "./env.js";
+import { runAnalyticsAggregation } from "./analytics.js";
+
+export const ANALYTICS_REFRESH_COOLDOWN_MS = 60 * 60 * 1000;
+export const ANALYTICS_REFRESH_LAST_RUN_KEY = "analytics:refresh:last-run-at";
+export const ANALYTICS_CRON_BACKFILL_DAYS = 40;
+export const ANALYTICS_CRON_BACKFILL_CURSOR_KEY = "analytics:cron:backfill-cursor";
+
+export async function backFillAnalytics(env: Env, maxRunsPerTick = 2) {
+  const kv = env.ANTSEED_KV;
+  const summaries: Array<{ currentDate: string; finalizedDates: string[] }> = [];
+  const runAt = new Date();
+  const todayDate = runAt.toISOString().slice(0, 10);
+  let cursorDate = await readAnalyticsBackfillCursorDate(env.ANTSEED_KV);
+  if (!cursorDate) {
+    cursorDate = dateDaysAgo(runAt, ANALYTICS_CRON_BACKFILL_DAYS);
+    await kv.put(ANALYTICS_CRON_BACKFILL_CURSOR_KEY, cursorDate);
+  }
+  // TODO: replace bounded loop with persisted backfill cursor when we need broader historical catch-up.
+  for (let i = 0; i < maxRunsPerTick; i += 1) {
+    console.log(`analytics backfill run ${i + 1} of ${maxRunsPerTick}, cursorDate=${cursorDate}`);
+    const aggregationRunAt = cursorDate === todayDate ? runAt : new Date(`${cursorDate}T23:59:59.999Z`);
+    const analyticsSummary = await runAnalyticsAggregation(env, aggregationRunAt);
+    console.log(`analytics backfill run ${i + 1} of ${maxRunsPerTick}, cursorDate=${cursorDate} completed`, analyticsSummary);
+    summaries.push({
+      currentDate: analyticsSummary.currentDate,
+      finalizedDates: analyticsSummary.finalizedDates
+    });
+    if (cursorDate === todayDate) {
+      break;
+    }
+    const nextCursorDate = nextUtcDate(cursorDate);
+    cursorDate = nextCursorDate && nextCursorDate <= todayDate ? nextCursorDate : todayDate;
+    await kv.put(ANALYTICS_CRON_BACKFILL_CURSOR_KEY, cursorDate);
+  }
+  return summaries;
+}
+
+export async function readAnalyticsRefreshTimestamp(kv: Pick<KVNamespace, "get">): Promise<number | undefined> {
+  const raw = await kv.get(ANALYTICS_REFRESH_LAST_RUN_KEY);
+  if (!raw) return undefined;
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed) || parsed < 0) return undefined;
+  return parsed;
+}
+
+export async function readAnalyticsBackfillCursorDate(kv: Pick<KVNamespace, "get">): Promise<string | undefined> {
+  const raw = await kv.get(ANALYTICS_CRON_BACKFILL_CURSOR_KEY);
+  if (!raw || !/^\d{4}-\d{2}-\d{2}$/.test(raw)) return undefined;
+  return raw;
+}
+
+export function dateDaysAgo(date: Date, days: number): string {
+  const shifted = new Date(date.getTime());
+  shifted.setUTCDate(shifted.getUTCDate() - days);
+  return shifted.toISOString().slice(0, 10);
+}
+
+export function nextUtcDate(date: string): string | undefined {
+  const parsed = new Date(`${date}T00:00:00.000Z`);
+  if (Number.isNaN(parsed.getTime())) return undefined;
+  parsed.setUTCDate(parsed.getUTCDate() + 1);
+  return parsed.toISOString().slice(0, 10);
+}

backend/src/funding.ts

@@ -0,0 +1,99 @@
+import { AntSeedFundingVaultClient } from "./antseed-funding-vault.js";
+import { KVCreditStore } from "./kv-credit-store.js";
+import { GdCreditEntry } from "./types.js";
+import { errorMessage, logError, logInfo, logWarn, redactAddress, redactHash } from "./logging.js";
+
+export async function fundCredit(
+  entry: GdCreditEntry,
+  store: KVCreditStore,
+  antseedFundingVault: AntSeedFundingVaultClient
+): Promise<{ [key: string]: unknown }> {
+  const buyer = entry.buyerAddress || entry.account;
+  if (entry.fundingStatus === "funded") {
+    logInfo("funding.already-funded", {
+      entryId: entry.id,
+      source: entry.source,
+      account: redactAddress(entry.account),
+      buyer: redactAddress(buyer),
+      fundingTxHash: redactHash(entry.fundingTxHash)
+    });
+    return {
+      ...entry,
+      bridge: {
+        enabled: antseedFundingVault.enabled,
+        buyer,
+        amountUsd: entry.totalCreditUsd,
+        txHash: entry.fundingTxHash,
+        alreadyFunded: true
+      }
+    };
+  }
+  const principalUsd = BigInt(entry.principalUsd);
+  const bonusUsd = BigInt(entry.bonusUsd);
+  logInfo("funding.start", {
+    entryId: entry.id,
+    source: entry.source,
+    account: redactAddress(entry.account),
+    buyer: redactAddress(buyer),
+    principalUsd: entry.principalUsd,
+    bonusUsd: entry.bonusUsd,
+    totalCreditUsd: entry.totalCreditUsd
+  });
+  try {
+    const shouldCredit = principalUsd + bonusUsd > 0n;
+    const bridge = shouldCredit
+      ? await antseedFundingVault.depositForBuyerWithId(buyer, principalUsd, bonusUsd, entry.id)
+      : { enabled: antseedFundingVault.enabled, buyer, amountUsd: "0" };
+    if (!shouldCredit) {
+      logInfo("funding.skipped.zero-amount", {
+        entryId: entry.id,
+        source: entry.source,
+        account: redactAddress(entry.account),
+        buyer: redactAddress(buyer)
+      });
+    } else if (!bridge.enabled) {
+      logWarn("funding.bridge.disabled", {
+        entryId: entry.id,
+        source: entry.source,
+        buyer: redactAddress(buyer)
+      });
+    }
+    const updated = await store.markFundingResult(entry, {
+      funded: true,
+      txHash: bridge.txHash,
+      error: undefined,
+      credited: shouldCredit
+    });
+    logInfo("funding.success", {
+      entryId: entry.id,
+      source: entry.source,
+      buyer: redactAddress(buyer),
+      txHash: redactHash(bridge.txHash),
+      bridgeEnabled: bridge.enabled
+    });
+    return { ...updated, bridge };
+  } catch (error) {
+    const message = errorMessage(error);
+    logError("funding.failed", {
+      entryId: entry.id,
+      source: entry.source,
+      account: redactAddress(entry.account),
+      buyer: redactAddress(buyer),
+      message
+    });
+    const updated = await store.markFundingResult(entry, {
+      funded: false,
+      error: message
+    });
+    return {
+      ...updated,
+      depositId: entry.id,
+      bridge: {
+        enabled: antseedFundingVault.enabled,
+        buyer,
+        amountUsd: entry.totalCreditUsd,
+        error: message
+      }
+    };
+  }
+}

backend/src/http.ts

@@ -0,0 +1,78 @@
+import { Env } from "./env.js";
+import { errorMessage, logWarn } from "./logging.js";
+
+export async function parseJson(request: Request): Promise<unknown> {
+  const contentType = request.headers.get("content-type") ?? "";
+  if (!contentType.includes("application/json")) throw new Error("content-type must be application/json");
+  return request.json();
+}
+
+export async function readRequestBody(request: Pick<Request, "text">): Promise<string | undefined> {
+  try {
+    const body = await request.text();
+    if (!body) return undefined;
+    return truncate(body, 3000);
+  } catch (error) {
+    return `[unavailable: ${errorMessage(error)}]`;
+  }
+}
+
+export function truncate(value: string, maxLength: number): string {
+  if (value.length <= maxLength) return value;
+  return `${value.slice(0, maxLength)}... [truncated]`;
+}
+
+export function json(body: unknown, status = 200): Response {
+  return cors(Response.json(body, { status }));
+}
+
+export function cors(response: Response): Response {
+  const headers = new Headers(response.headers);
+  headers.set("access-control-allow-origin", "*");
+  headers.set("access-control-allow-methods", "GET,POST,DELETE,OPTIONS");
+  headers.set("access-control-allow-headers", "content-type,authorization,x-api-key,x-gooddollar-account");
+  return new Response(response.body, {
+    status: response.status,
+    statusText: response.statusText,
+    headers
+  });
+}
+
+export async function notifySlackOnRequestError(env: Env, details: { method: string; path: string; body: string | undefined; message: string }): Promise<void> {
+  if (!env.SLACK_WEBHOOK_URL) return;
+
+  try {
+    const text = [
+      "Worker request exception",
+      `method: ${details.method}`,
+      `path: ${details.path}`,
+      `body: ${formatSlackBody(details.body)}`,
+      `error: ${details.message}`
+    ].join("\n");
+
+    const response = await fetch(env.SLACK_WEBHOOK_URL, {
+      method: "POST",
+      headers: {
+        "content-type": "application/json"
+      },
+      body: JSON.stringify({ text })
+    });
+
+    if (!response.ok) {
+      logWarn("slack.notify.failed", {
+        status: response.status,
+        path: details.path
+      });
+    }
+  } catch (error) {
+    logWarn("slack.notify.error", {
+      path: details.path,
+      message: errorMessage(error)
+    });
+  }
+}
+
+export function formatSlackBody(body: string | undefined): string {
+  if (body === undefined) return "[empty]";
+  return body;
+}