From 21783b422f64b4fd72eb4602f1fd5b96cde3d9a1 Mon Sep 17 00:00:00 2001 From: blueogin Date: Mon, 12 Oct 2026 01:14:50 -0400 Subject: [PATCH 1/2] feat: add Superfluid streaming functionality - Introduced new module `superfluid.ts` to handle fetching and processing Superfluid streams. - Implemented `fetchSuperfluidIncomingStreams` and `fetchSuperfluidStreamsForAccount` functions for retrieving stream data. - Updated `worker.ts` to integrate Superfluid streams into the crediting process, replacing the previous inline implementation. - Refactored stream credit calculation logic into `stream-credits.ts` for better modularity and maintainability. - Adjusted tests in `worker.test.ts` to accommodate changes in the worker module and ensure proper functionality. --- backend/src/analytics-cron.ts | 64 ++++ backend/src/funding.ts | 99 ++++++ backend/src/http.ts | 78 +++++ backend/src/stream-credits.ts | 237 +++++++++++++ backend/src/superfluid.ts | 180 ++++++++++ backend/src/worker.ts | 629 +--------------------------------- backend/test/worker.test.ts | 3 +- 7 files changed, 676 insertions(+), 614 deletions(-) create mode 100644 backend/src/analytics-cron.ts create mode 100644 backend/src/funding.ts create mode 100644 backend/src/http.ts create mode 100644 backend/src/stream-credits.ts create mode 100644 backend/src/superfluid.ts diff --git a/backend/src/analytics-cron.ts b/backend/src/analytics-cron.ts new file mode 100644 index 0000000..f4642e7 --- /dev/null +++ b/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): Promise { + 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): Promise { + 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); +} diff --git a/backend/src/funding.ts b/backend/src/funding.ts new file mode 100644 index 0000000..8973af8 --- /dev/null +++ b/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 + } + }; + } +} diff --git a/backend/src/http.ts b/backend/src/http.ts new file mode 100644 index 0000000..6be27fb --- /dev/null +++ b/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 { + 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): Promise { + 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 { + 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; +} diff --git a/backend/src/stream-credits.ts b/backend/src/stream-credits.ts new file mode 100644 index 0000000..d16ae6d --- /dev/null +++ b/backend/src/stream-credits.ts @@ -0,0 +1,237 @@ +import { AntSeedFundingVaultClient } from "./antseed-funding-vault.js"; +import { fetchCurrentGdPrice, fetchGoodIdRoot } from "./celo-events.js"; +import { configFromEnv } from "./env.js"; +import { fundCredit } from "./funding.js"; +import { KVCreditStore } from "./kv-credit-store.js"; +import { UserCreditProfile } from "./types.js"; +import { logInfo, logWarn, redactAddress } from "./logging.js"; +import { fetchSuperfluidIncomingStreams } from "./superfluid.js"; +import { SuperfluidIncomingStream } from "./superfluid.js"; + +/** + * Stream credit entries are keyed per account per day, so the scheduled run and a user's own + * request collapse onto one entry instead of paying the same window twice. + */ +export function createStreamFundingId(account: string, date: Date): string { + const day = date.toISOString().slice(0, 10); // YYYY-MM-DD + return `stream:${day}:${account.toLowerCase()}`; +} + +/** + * Seconds of stream to credit on this run: the window from the later of the account's last stream + * credit and the current stream revision's creation, up to `now`. + * + * The window is deliberately not capped: a long gap means the stream really did flow for that + * whole time (a stalled cron, a backfill), and the credit should reflect it. What must not happen + * is inventing a window out of a baseline we never had. When neither timestamp is usable the + * answer is 0, not "since the epoch". + */ +export function streamElapsedSeconds(now: Date, lastStreamCreditAt: string | undefined, streamLastUpdateAt?: string): number { + const lastCreditMs = lastStreamCreditAt ? Date.parse(lastStreamCreditAt) : 0; + const lastUpdateMs = streamLastUpdateAt ? Date.parse(streamLastUpdateAt) : 0; + // The later of the two wins. `lastStreamCreditAt` alone is not enough: it survives a stream being + // closed, so a user who closes a stream and opens a new one months later would be credited for + // the whole dormant gap. The stream's last on-chain update is a hard floor -- it moves forward on + // every create, rate change and close, so a stale clock can never be reached again. + // + // The trade-off: it also moves on a plain rate change, so when the resulting `StreamUpdated` + // event is not ingested (that endpoint is push-only, no retry) the window between the last credit + // and the rate change is skipped and never paid. Chosen deliberately -- it errs towards + // under-crediting rather than over-crediting. + const baselineMs = Math.max(Number.isFinite(lastCreditMs) ? lastCreditMs : 0, Number.isFinite(lastUpdateMs) ? lastUpdateMs : 0); + if (baselineMs <= 0) { + logWarn("stream.credits.unusable-baseline", { lastStreamCreditAt, streamLastUpdateAt }); + return 0; + } + return Math.max(0, Math.floor((now.getTime() - baselineMs) / 1000)); +} + +/** + * What a stream owes credit for, and over what window. + * + * Live stream: from the later of the last credit and the stream's last on-chain update, up to now, + * at the rate the subgraph reports. + * + * Closed stream: from the last credit up to the **termination**, at the rate that was flowing before + * it closed. The end has to be the termination rather than `now` -- the stream stopped carrying G$ + * then, so measuring to `now` would invent time it did not flow. + * + * NOTE: this branch has neither a floor (a closed stream's last update *is* its termination) nor a + * ceiling, so a stale `lastStreamCreditAt` priced at a remembered rate can over-credit. Tracked in + * the PR notes -- reading `streamPeriods` is the fix, not another guard. + */ +export function streamCreditAmount(input: { + now: Date; + lastStreamCreditAt: string | undefined; + stream: SuperfluidIncomingStream; + /** The rate recorded before this run zeroed it; only consulted for a closed stream. */ + previousFlowRateWeiPerSecond: bigint; +}): { elapsedSeconds: number; gdAmountWei: bigint } { + const flowRateWeiPerSecond = BigInt(input.stream.flowRateWeiPerSecond); + if (flowRateWeiPerSecond > 0n) { + const elapsedSeconds = streamElapsedSeconds(input.now, input.lastStreamCreditAt, input.stream.lastUpdateAt); + return { elapsedSeconds, gdAmountWei: flowRateWeiPerSecond * BigInt(elapsedSeconds) }; + } + + const terminatedAtMs = Date.parse(input.stream.lastUpdateAt); + const lastCreditMs = input.lastStreamCreditAt ? Date.parse(input.lastStreamCreditAt) : 0; + if (!Number.isFinite(terminatedAtMs) || terminatedAtMs <= 0 || !Number.isFinite(lastCreditMs) || lastCreditMs <= 0) { + logWarn("stream.credits.unusable-final-window", { + lastStreamCreditAt: input.lastStreamCreditAt, + terminatedAt: input.stream.lastUpdateAt + }); + return { elapsedSeconds: 0, gdAmountWei: 0n }; + } + + const elapsedSeconds = Math.max(0, Math.floor((terminatedAtMs - lastCreditMs) / 1000)); + return { elapsedSeconds, gdAmountWei: input.previousFlowRateWeiPerSecond * BigInt(elapsedSeconds) }; +} + +/** + * One pass of stream credit accrual: sync every account's recorded flow rate from the subgraph, + * then credit each stream the window it is owed and fund it. + * + * Split out of the scheduled handler so the orchestration is readable on its own and so the + * handler is left doing only what a cron entrypoint should -- wire dependencies and sequence the + * passes. Funding is fired through `ctx.waitUntil` and deliberately not awaited. + */ +export async function runStreamCreditCycle( + cfg: ReturnType, + store: KVCreditStore, + antseedFundingVault: AntSeedFundingVaultClient, + ctx: ExecutionContext, + startedAt: number +): Promise { + const gdPrice = await fetchCurrentGdPrice(cfg); + // Includes closed streams: a terminated stream reports `currentFlowRate = 0`, and seeing that is + // the only way the backend learns it stopped. The termination event that would otherwise say so + // is pushed in by `POST /v1/celo/events/record` and may never arrive. + const streams = await fetchSuperfluidIncomingStreams(cfg); + + let inactiveAccounts = 0; + let syncedFlowRates = 0; + let syncedRootFlowRates = 0; + let skippedCooldown = 0; + let skippedMinAmount = 0; + let processed = 0; + let funded = 0; + let failed = 0; + const createdAt = new Date().toISOString(); + + // Sync every account's recorded rate from the subgraph before crediting. Doing it here rather + // than inside `recordGdCredit` means it happens on every run, not only when a credit clears the + // cooldown and the minimum -- at a 4000 G$ minimum those can be a fortnight apart. Summed per + // account, because one account can hold several revisions (a closed stream and its replacement) + // and a closed row must not clobber the rate of the live one. All-closed sums to 0, which is + // how the profile stops advertising a stream that no longer exists. + const flowRateByAccount = new Map(); + for (const stream of streams) { + const account = stream.account.toLowerCase(); + flowRateByAccount.set(account, (flowRateByAccount.get(account) ?? 0n) + BigInt(stream.flowRateWeiPerSecond)); + } + // One profile read per account, reused by the crediting pass below. `getSavedUser` rather than + // `getUser` because syncing is a correction to state that already exists -- an account that has + // never been credited has no profile, and the scheduled run must not create one as a side + // effect. Its first credit will create it, in `recordGdCredit`. + const profileByAccount = new Map(); + const flowRateByRoot = new Map(); + // The rate as it stood before this run's sync. A stream that has just closed reports 0, so its + // final window has to be priced at what was flowing beforehand. + const previousFlowRateByAccount = new Map(); + for (const [account, flowRate] of flowRateByAccount) { + const profile = await store.getSavedUser(account); + if (!profile) continue; + previousFlowRateByAccount.set(account, BigInt(profile.streamFlowRateWeiPerSecond)); + if (await store.recordStreamFlowRate(profile, flowRate)) syncedFlowRates += 1; + profileByAccount.set(account, profile); + flowRateByRoot.set(profile.rootAccount, (flowRateByRoot.get(profile.rootAccount) ?? 0n) + flowRate); + } + + // A GoodID root profile aggregates its identity's wallets. The totals get there by mirroring in + // `updateUser`, which works because they accumulate -- a flow rate is absolute, so it is summed + // here instead. A root that streams itself is part of its own sum, so this runs after the + // per-account pass and takes precedence over that account's own rate. + for (const [rootAccount, totalFlowRate] of flowRateByRoot) { + const rootProfile = profileByAccount.get(rootAccount) ?? (await store.getSavedUser(rootAccount)); + if (!rootProfile) continue; + if (await store.recordStreamFlowRate(rootProfile, totalFlowRate)) syncedRootFlowRates += 1; + } + inactiveAccounts = [...flowRateByAccount.values()].filter((flowRate) => flowRate === 0n).length; + + for (const stream of streams) { + const account = stream.account.toLowerCase(); + // Falls back to the synthesized default for an account streaming for the first time, which is + // exactly the case the sync above skips. + const profile = profileByAccount.get(account) ?? (await store.getUser(stream.account)); + const now = new Date(); + // Closed streams are credited too, for the window between the last credit and the termination + // -- without this the tail of every stream is silently forfeited. + const { elapsedSeconds, gdAmountWei } = streamCreditAmount({ + now, + lastStreamCreditAt: profile.lastStreamCreditAt, + stream, + previousFlowRateWeiPerSecond: previousFlowRateByAccount.get(account) ?? 0n + }); + if (gdAmountWei === 0n) continue; + if (elapsedSeconds < 60 * 60 * 24) { + skippedCooldown += 1; + continue; + } + if (gdAmountWei < cfg.MIN_STREAM_BONUS_WEI) { + skippedMinAmount += 1; + continue; + } + const rootAccount = await fetchGoodIdRoot(stream.account, cfg); + const buyerForOperatorCheck = stream.buyerAddress || stream.account; + const hasOperatorConsent = antseedFundingVault.enabled ? (await antseedFundingVault.isBuyerOperator(buyerForOperatorCheck)).isOperator : true; + const isValidForBonus = !!rootAccount && hasOperatorConsent; + + const depositId = createStreamFundingId(stream.account, new Date(createdAt)); + const entry = await store.recordGdCredit({ + id: depositId, + account: stream.account, + rootAccount, + source: "streamCron", + gdAmountWei: BigInt(gdAmountWei), + // The account's summed rate, not this row's: an account can hold several rows, and a closed + // one reports 0. Also what gives a first-time streamer -- skipped by the sync above for + // having no profile yet -- its rate on the profile this credit creates. + flowRate: flowRateByAccount.get(account) ?? 0n, + isVerified: isValidForBonus, // if root acccount was found it is whitelisted & operator set + gdPrice, + maxBonusCapUsd: cfg.MAX_BONUS_CAP_USD, + regularBonusBps: cfg.REGULAR_BONUS_BPS, + streamingBonusBps: cfg.STREAMING_BONUS_BPS, + buyerAddress: stream.buyerAddress + }); + logInfo("streamCron.credits.entry", { + entryId: entry.id, + account: redactAddress(entry.account), + rootAccount: redactAddress(entry.rootAccount), + buyer: redactAddress(entry.buyerAddress), + elapsedSeconds, + stream, + entry, + profile + }); + processed += 1; + ctx.waitUntil( + fundCredit(entry, store, antseedFundingVault).then((result) => { + if (result.fundingStatus === "funded") funded += 1; + if (result.fundingStatus === "failed") failed += 1; + }) + ); + } + logInfo("cron.summary", { + streamCount: streams.length, + processed, + skippedCooldown, + skippedMinAmount, + inactiveAccounts, + syncedFlowRates, + syncedRootFlowRates, + funded, + failed, + elapsedMs: Date.now() - startedAt + }); +} diff --git a/backend/src/superfluid.ts b/backend/src/superfluid.ts new file mode 100644 index 0000000..f4c4a1f --- /dev/null +++ b/backend/src/superfluid.ts @@ -0,0 +1,180 @@ +import { z } from "zod"; +import { decodeBuyerFromUserData } from "./celo-events.js"; +import { configFromEnv } from "./env.js"; +import { errorMessage, logError, logInfo, logWarn, redactAddress } from "./logging.js"; + +/** Default endpoint, used when `SUPERFLUID_SUBGRAPH_URL` is unset. */ +export const SUPERFLUID_CELO_SUBGRAPH_URL = "https://subgraph-endpoints.superfluid.dev/celo-mainnet/protocol-v1"; + +const SuperfluidStreamsResponseSchema = z.object({ + data: z.object({ + streams: z.array( + z.object({ + sender: z.object({ id: z.string().regex(/^0x[0-9a-fA-F]{40}$/) }), + currentFlowRate: z.string().regex(/^\d+$/), + updatedAtTimestamp: z.string().regex(/^\d+$/), + flowUpdatedEvents: z.array( + z.object({ + userData: z.string() + }) + ) + }) + ) + }) +}); + +export type SuperfluidIncomingStream = { + account: string; + flowRateWeiPerSecond: string; + /** For a closed stream this is its termination; for a live one, its last on-chain change. */ + lastUpdateAt: string; + /** AntSeed buyer decoded from the most recent FlowUpdatedEvent userdata. */ + buyerAddress?: string; +}; + +/** + * Every stream to the vault, closed ones included. The scheduled run needs the closed rows: a + * terminated stream reports `currentFlowRate = 0`, which is how the backend learns it stopped. + */ +export async function fetchSuperfluidIncomingStreams(cfg: ReturnType): Promise { + return fetchSuperfluidStreams(cfg, undefined, true); +} + +/** Active streams for one account; callers here use it to answer "is this account streaming now". */ +export async function fetchSuperfluidStreamsForAccount( + account: string, + cfg: ReturnType, + includeClosed = false +): Promise { + return fetchSuperfluidStreams(cfg, account, includeClosed); +} + +export async function fetchSuperfluidStreams( + cfg: ReturnType, + senderFilter?: string, + includeClosed = false +): Promise { + if (!cfg.CELO_VAULT_ADDRESS || !cfg.CELO_GD_SUPERTOKEN_ADDRESS) { + logWarn("superfluid.streams.skipped", { + reason: "missing_config", + hasVaultAddress: Boolean(cfg.CELO_VAULT_ADDRESS), + hasSuperTokenAddress: Boolean(cfg.CELO_GD_SUPERTOKEN_ADDRESS), + senderFilter: redactAddress(senderFilter) + }); + return []; + } + + const endpoint = cfg.SUPERFLUID_SUBGRAPH_URL ?? SUPERFLUID_CELO_SUBGRAPH_URL; + const receiver = cfg.CELO_VAULT_ADDRESS.toLowerCase(); + const token = cfg.CELO_GD_SUPERTOKEN_ADDRESS.toLowerCase(); + const pageSize = 500; + let skip = 0; + const streams: SuperfluidIncomingStream[] = []; + let pages = 0; + + logInfo("superfluid.streams.fetch.start", { + endpoint, + receiver: redactAddress(receiver), + token: redactAddress(token), + senderFilter: redactAddress(senderFilter) + }); + + try { + while (true) { + const filters = ["receiver: $receiver", "token: $token"]; + if (!includeClosed) filters.push(`currentFlowRate_gt: "0"`); + if (senderFilter) filters.push("sender: $sender"); + const whereClause = `{ ${filters.join(", ")} }`; + const queryParams = senderFilter + ? `$receiver: String!, $token: String!, $first: Int!, $skip: Int!, $sender: String!` + : `$receiver: String!, $token: String!, $first: Int!, $skip: Int!`; + const variables: Record = { + receiver, + token, + first: pageSize, + skip + }; + if (senderFilter) variables.sender = senderFilter.toLowerCase(); + + const response = await fetch(endpoint, { + method: "POST", + headers: { + "content-type": "application/json" + }, + body: JSON.stringify({ + query: ` + query ActiveIncomingStreams(${queryParams}) { + streams( + first: $first + skip: $skip + where: ${whereClause} + orderBy: updatedAtTimestamp + orderDirection: desc + ) { + sender { id } + currentFlowRate + updatedAtTimestamp + flowUpdatedEvents(orderBy: timestamp, orderDirection: desc, first: 1) { + userData + } + } + } + `, + variables + }) + }); + + if (!response.ok) { + throw new Error(`Superfluid subgraph HTTP ${response.status}`); + } + + const body = await response.json(); + const parsed = SuperfluidStreamsResponseSchema.safeParse(body); + if (!parsed.success) { + throw new Error("invalid Superfluid subgraph response shape"); + } + + const batch = parsed.data.data.streams.map((stream) => { + const flowRateWeiPerSecond = stream.currentFlowRate; + const updatedAtSeconds = Number(stream.updatedAtTimestamp); + // A zero `updatedAtTimestamp` must not become 1970: this is the baseline for an account's + // first stream credit, so a 1970 value makes that credit window decades wide. + const hasUsableUpdatedAt = Number.isFinite(updatedAtSeconds) && updatedAtSeconds > 0; + const lastUpdateAt = hasUsableUpdatedAt ? new Date(updatedAtSeconds * 1000).toISOString() : new Date().toISOString(); + const rawUserData = stream.flowUpdatedEvents[0]?.userData; + const buyerAddress = decodeBuyerFromUserData(rawUserData); + return { + account: stream.sender.id.toLowerCase(), + flowRateWeiPerSecond, + lastUpdateAt, + buyerAddress + }; + }); + + streams.push(...batch); + pages += 1; + + if (batch.length < pageSize) break; + skip += pageSize; + } + } catch (error) { + const message = errorMessage(error); + logError("superfluid.streams.fetch.failed", { + endpoint, + receiver: redactAddress(receiver), + token: redactAddress(token), + senderFilter: redactAddress(senderFilter), + message + }); + return []; + } + + logInfo("superfluid.streams.fetch.end", { + endpoint, + senderFilter: redactAddress(senderFilter), + pages, + count: streams.length + }); + + return streams; +} diff --git a/backend/src/worker.ts b/backend/src/worker.ts index 67fcbdd..8f0b452 100644 --- a/backend/src/worker.ts +++ b/backend/src/worker.ts @@ -1,17 +1,21 @@ import { z } from "zod"; import { AntSeedFundingVaultClient } from "./antseed-funding-vault.js"; -import { fetchCeloVaultEvents, fetchCeloVaultEventsForAccount, fetchCurrentGdPrice, fetchGoodIdRoot, decodeBuyerFromUserData } from "./celo-events.js"; +import { fetchCeloVaultEvents, fetchCeloVaultEventsForAccount, fetchCurrentGdPrice, fetchGoodIdRoot } from "./celo-events.js"; import { Env, configFromEnv } from "./env.js"; import { KVCreditStore } from "./kv-credit-store.js"; -import { GdCreditEntry, UserCreditProfile } from "./types.js"; -import { errorMessage, logError, logInfo, logWarn, redactAddress, redactHash } from "./logging.js"; -import { getAnalyticsWindow, runAnalyticsAggregation, KVAnalyticsStore } from "./analytics.js"; - -const ANALYTICS_REFRESH_COOLDOWN_MS = 60 * 60 * 1000; -const ANALYTICS_REFRESH_LAST_RUN_KEY = "analytics:refresh:last-run-at"; -const ANALYTICS_CRON_BACKFILL_DAYS = 40; -const ANALYTICS_CRON_BACKFILL_CURSOR_KEY = "analytics:cron:backfill-cursor"; - +import { errorMessage, logError, logInfo, redactAddress, redactHash } from "./logging.js"; +import { getAnalyticsWindow } from "./analytics.js"; +import { fundCredit } from "./funding.js"; +import { cors, json, notifySlackOnRequestError, parseJson, readRequestBody } from "./http.js"; +import { createStreamFundingId, runStreamCreditCycle, streamCreditAmount, streamElapsedSeconds } from "./stream-credits.js"; +import { + ANALYTICS_CRON_BACKFILL_DAYS, + ANALYTICS_REFRESH_COOLDOWN_MS, + ANALYTICS_REFRESH_LAST_RUN_KEY, + backFillAnalytics, + readAnalyticsRefreshTimestamp +} from "./analytics-cron.js"; +import { SUPERFLUID_CELO_SUBGRAPH_URL, fetchSuperfluidStreamsForAccount } from "./superfluid.js"; const CeloEventsRecordSchema = z .object({ txHash: z @@ -74,36 +78,10 @@ const ChannelOpSchema = z.object({ .regex(/^0x[0-9a-fA-F]+$/) .optional() }); -const SuperfluidStreamsResponseSchema = z.object({ - data: z.object({ - streams: z.array( - z.object({ - sender: z.object({ id: z.string().regex(/^0x[0-9a-fA-F]{40}$/) }), - currentFlowRate: z.string().regex(/^\d+$/), - updatedAtTimestamp: z.string().regex(/^\d+$/), - flowUpdatedEvents: z.array( - z.object({ - userData: z.string() - }) - ) - }) - ) - }) -}); const AnalyticsQuerySchema = z.object({ days: z.coerce.number().int().min(1).max(365).default(30) }); -const SUPERFLUID_CELO_SUBGRAPH_URL = "https://subgraph-endpoints.superfluid.dev/celo-mainnet/protocol-v1"; - -type SuperfluidIncomingStream = { - account: string; - flowRateWeiPerSecond: string; - /** For a closed stream this is its termination; for a live one, its last on-chain change. */ - lastUpdateAt: string; - /** AntSeed buyer decoded from the most recent FlowUpdatedEvent userdata. */ - buyerAddress?: string; -}; export default { async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { @@ -153,138 +131,7 @@ export default { bridgeEnabled: antseedFundingVault.enabled }); - const gdPrice = await fetchCurrentGdPrice(cfg); - // Includes closed streams: a terminated stream reports `currentFlowRate = 0`, and seeing that is - // the only way the backend learns it stopped. The termination event that would otherwise say so - // is pushed in by `POST /v1/celo/events/record` and may never arrive. - const streams = await fetchSuperfluidIncomingStreams(cfg); - - let inactiveAccounts = 0; - let syncedFlowRates = 0; - let syncedRootFlowRates = 0; - let skippedCooldown = 0; - let skippedMinAmount = 0; - let processed = 0; - let funded = 0; - let failed = 0; - const createdAt = new Date().toISOString(); - - // Sync every account's recorded rate from the subgraph before crediting. Doing it here rather - // than inside `recordGdCredit` means it happens on every run, not only when a credit clears the - // cooldown and the minimum -- at a 4000 G$ minimum those can be a fortnight apart. Summed per - // account, because one account can hold several revisions (a closed stream and its replacement) - // and a closed row must not clobber the rate of the live one. All-closed sums to 0, which is - // how the profile stops advertising a stream that no longer exists. - const flowRateByAccount = new Map(); - for (const stream of streams) { - const account = stream.account.toLowerCase(); - flowRateByAccount.set(account, (flowRateByAccount.get(account) ?? 0n) + BigInt(stream.flowRateWeiPerSecond)); - } - // One profile read per account, reused by the crediting pass below. `getSavedUser` rather than - // `getUser` because syncing is a correction to state that already exists -- an account that has - // never been credited has no profile, and the scheduled run must not create one as a side - // effect. Its first credit will create it, in `recordGdCredit`. - const profileByAccount = new Map(); - const flowRateByRoot = new Map(); - // The rate as it stood before this run's sync. A stream that has just closed reports 0, so its - // final window has to be priced at what was flowing beforehand. - const previousFlowRateByAccount = new Map(); - for (const [account, flowRate] of flowRateByAccount) { - const profile = await store.getSavedUser(account); - if (!profile) continue; - previousFlowRateByAccount.set(account, BigInt(profile.streamFlowRateWeiPerSecond)); - if (await store.recordStreamFlowRate(profile, flowRate)) syncedFlowRates += 1; - profileByAccount.set(account, profile); - flowRateByRoot.set(profile.rootAccount, (flowRateByRoot.get(profile.rootAccount) ?? 0n) + flowRate); - } - - // A GoodID root profile aggregates its identity's wallets. The totals get there by mirroring in - // `updateUser`, which works because they accumulate -- a flow rate is absolute, so it is summed - // here instead. A root that streams itself is part of its own sum, so this runs after the - // per-account pass and takes precedence over that account's own rate. - for (const [rootAccount, totalFlowRate] of flowRateByRoot) { - const rootProfile = profileByAccount.get(rootAccount) ?? (await store.getSavedUser(rootAccount)); - if (!rootProfile) continue; - if (await store.recordStreamFlowRate(rootProfile, totalFlowRate)) syncedRootFlowRates += 1; - } - inactiveAccounts = [...flowRateByAccount.values()].filter((flowRate) => flowRate === 0n).length; - - for (const stream of streams) { - const account = stream.account.toLowerCase(); - // Falls back to the synthesized default for an account streaming for the first time, which is - // exactly the case the sync above skips. - const profile = profileByAccount.get(account) ?? (await store.getUser(stream.account)); - const now = new Date(); - // Closed streams are credited too, for the window between the last credit and the termination - // -- without this the tail of every stream is silently forfeited. - const { elapsedSeconds, gdAmountWei } = streamCreditAmount({ - now, - lastStreamCreditAt: profile.lastStreamCreditAt, - stream, - previousFlowRateWeiPerSecond: previousFlowRateByAccount.get(account) ?? 0n - }); - if (gdAmountWei === 0n) continue; - if (elapsedSeconds < 60 * 60 * 24) { - skippedCooldown += 1; - continue; - } - if (gdAmountWei < cfg.MIN_STREAM_BONUS_WEI) { - skippedMinAmount += 1; - continue; - } - const rootAccount = await fetchGoodIdRoot(stream.account, cfg); - const buyerForOperatorCheck = stream.buyerAddress || stream.account; - const hasOperatorConsent = antseedFundingVault.enabled ? (await antseedFundingVault.isBuyerOperator(buyerForOperatorCheck)).isOperator : true; - const isValidForBonus = !!rootAccount && hasOperatorConsent; - - const depositId = createStreamFundingId(stream.account, new Date(createdAt)); - const entry = await store.recordGdCredit({ - id: depositId, - account: stream.account, - rootAccount, - source: "streamCron", - gdAmountWei: BigInt(gdAmountWei), - // The account's summed rate, not this row's: an account can hold several rows, and a closed - // one reports 0. Also what gives a first-time streamer -- skipped by the sync above for - // having no profile yet -- its rate on the profile this credit creates. - flowRate: flowRateByAccount.get(account) ?? 0n, - isVerified: isValidForBonus, // if root acccount was found it is whitelisted & operator set - gdPrice, - maxBonusCapUsd: cfg.MAX_BONUS_CAP_USD, - regularBonusBps: cfg.REGULAR_BONUS_BPS, - streamingBonusBps: cfg.STREAMING_BONUS_BPS, - buyerAddress: stream.buyerAddress - }); - logInfo("streamCron.credits.entry", { - entryId: entry.id, - account: redactAddress(entry.account), - rootAccount: redactAddress(entry.rootAccount), - buyer: redactAddress(entry.buyerAddress), - elapsedSeconds, - stream, - entry, - profile - }); - processed += 1; - ctx.waitUntil( - fundCredit(entry, store, antseedFundingVault).then((result) => { - if (result.fundingStatus === "funded") funded += 1; - if (result.fundingStatus === "failed") failed += 1; - }) - ); - } - logInfo("cron.summary", { - streamCount: streams.length, - processed, - skippedCooldown, - skippedMinAmount, - inactiveAccounts, - syncedFlowRates, - syncedRootFlowRates, - funded, - failed, - elapsedMs: Date.now() - startedAt - }); + await runStreamCreditCycle(cfg, store, antseedFundingVault, ctx, startedAt); try { const summaries = await backFillAnalytics(env, ANALYTICS_CRON_BACKFILL_DAYS); @@ -308,35 +155,6 @@ export default { } }; -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; -} async function route(request: Request, env: Env, _ctx: ExecutionContext): Promise { const url = new URL(request.url); const cfg = configFromEnv(env); @@ -803,422 +621,7 @@ async function route(request: Request, env: Env, _ctx: ExecutionContext): Promis return json({ error: "not found" }, 404); } -async function parseJson(request: Request): Promise { - const contentType = request.headers.get("content-type") ?? ""; - if (!contentType.includes("application/json")) throw new Error("content-type must be application/json"); - return request.json(); -} - -async function notifySlackOnRequestError(env: Env, details: { method: string; path: string; body: string | undefined; message: string }): Promise { - 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) - }); - } -} - -async function readRequestBody(request: Pick): Promise { - try { - const body = await request.text(); - if (!body) return undefined; - return truncate(body, 3000); - } catch (error) { - return `[unavailable: ${errorMessage(error)}]`; - } -} -function formatSlackBody(body: string | undefined): string { - if (body === undefined) return "[empty]"; - return body; -} - -function truncate(value: string, maxLength: number): string { - if (value.length <= maxLength) return value; - return `${value.slice(0, maxLength)}... [truncated]`; -} -function json(body: unknown, status = 200): Response { - return cors(Response.json(body, { status })); -} - -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 - }); -} -function createStreamFundingId(account: string, date: Date): string { - const day = date.toISOString().slice(0, 10); // YYYY-MM-DD - return `stream:${day}:${account.toLowerCase()}`; -} - -/** - * Seconds of stream to credit on this run: the window from the later of the account's last stream - * credit and the current stream revision's creation, up to `now`. - * - * The window is deliberately not capped: a long gap means the stream really did flow for that - * whole time (a stalled cron, a backfill), and the credit should reflect it. What must not happen - * is inventing a window out of a baseline we never had. When neither timestamp is usable the - * answer is 0, not "since the epoch". - */ -export function streamElapsedSeconds(now: Date, lastStreamCreditAt: string | undefined, streamLastUpdateAt?: string): number { - const lastCreditMs = lastStreamCreditAt ? Date.parse(lastStreamCreditAt) : 0; - const lastUpdateMs = streamLastUpdateAt ? Date.parse(streamLastUpdateAt) : 0; - // The later of the two wins. `lastStreamCreditAt` alone is not enough: it survives a stream being - // closed, so a user who closes a stream and opens a new one months later would be credited for - // the whole dormant gap. The stream's last on-chain update is a hard floor -- it moves forward on - // every create, rate change and close, so a stale clock can never be reached again. - // - // The trade-off: it also moves on a plain rate change, so when the resulting `StreamUpdated` - // event is not ingested (that endpoint is push-only, no retry) the window between the last credit - // and the rate change is skipped and never paid. Chosen deliberately -- it errs towards - // under-crediting rather than over-crediting. - const baselineMs = Math.max(Number.isFinite(lastCreditMs) ? lastCreditMs : 0, Number.isFinite(lastUpdateMs) ? lastUpdateMs : 0); - if (baselineMs <= 0) { - logWarn("stream.credits.unusable-baseline", { lastStreamCreditAt, streamLastUpdateAt }); - return 0; - } - return Math.max(0, Math.floor((now.getTime() - baselineMs) / 1000)); -} - -/** - * What a stream owes credit for, and over what window. - * - * Live stream: from the later of the last credit and the stream's last on-chain update, up to now, - * at the rate the subgraph reports. - * - * Closed stream: from the last credit up to the **termination**, at the rate that was flowing before - * it closed. The end has to be the termination rather than `now` -- the stream stopped carrying G$ - * then, so measuring to `now` would invent time it did not flow. - * - * NOTE: this branch has neither a floor (a closed stream's last update *is* its termination) nor a - * ceiling, so a stale `lastStreamCreditAt` priced at a remembered rate can over-credit. Tracked in - * the PR notes -- reading `streamPeriods` is the fix, not another guard. - */ -function streamCreditAmount(input: { - now: Date; - lastStreamCreditAt: string | undefined; - stream: SuperfluidIncomingStream; - /** The rate recorded before this run zeroed it; only consulted for a closed stream. */ - previousFlowRateWeiPerSecond: bigint; -}): { elapsedSeconds: number; gdAmountWei: bigint } { - const flowRateWeiPerSecond = BigInt(input.stream.flowRateWeiPerSecond); - if (flowRateWeiPerSecond > 0n) { - const elapsedSeconds = streamElapsedSeconds(input.now, input.lastStreamCreditAt, input.stream.lastUpdateAt); - return { elapsedSeconds, gdAmountWei: flowRateWeiPerSecond * BigInt(elapsedSeconds) }; - } - - const terminatedAtMs = Date.parse(input.stream.lastUpdateAt); - const lastCreditMs = input.lastStreamCreditAt ? Date.parse(input.lastStreamCreditAt) : 0; - if (!Number.isFinite(terminatedAtMs) || terminatedAtMs <= 0 || !Number.isFinite(lastCreditMs) || lastCreditMs <= 0) { - logWarn("stream.credits.unusable-final-window", { - lastStreamCreditAt: input.lastStreamCreditAt, - terminatedAt: input.stream.lastUpdateAt - }); - return { elapsedSeconds: 0, gdAmountWei: 0n }; - } - - const elapsedSeconds = Math.max(0, Math.floor((terminatedAtMs - lastCreditMs) / 1000)); - return { elapsedSeconds, gdAmountWei: input.previousFlowRateWeiPerSecond * BigInt(elapsedSeconds) }; -} - -async function readAnalyticsRefreshTimestamp(kv: Pick): Promise { - 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; -} - -async function readAnalyticsBackfillCursorDate(kv: Pick): Promise { - const raw = await kv.get(ANALYTICS_CRON_BACKFILL_CURSOR_KEY); - if (!raw || !/^\d{4}-\d{2}-\d{2}$/.test(raw)) return undefined; - return raw; -} - -function dateDaysAgo(date: Date, days: number): string { - const shifted = new Date(date.getTime()); - shifted.setUTCDate(shifted.getUTCDate() - days); - return shifted.toISOString().slice(0, 10); -} - -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); -} - -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 - } - }; - } -} - -/** - * Every stream to the vault, closed ones included. The scheduled run needs the closed rows: a - * terminated stream reports `currentFlowRate = 0`, which is how the backend learns it stopped. - */ -async function fetchSuperfluidIncomingStreams(cfg: ReturnType): Promise { - return fetchSuperfluidStreams(cfg, undefined, true); -} - -/** Active streams for one account; callers here use it to answer "is this account streaming now". */ -async function fetchSuperfluidStreamsForAccount( - account: string, - cfg: ReturnType, - includeClosed = false -): Promise { - return fetchSuperfluidStreams(cfg, account, includeClosed); -} - -async function fetchSuperfluidStreams( - cfg: ReturnType, - senderFilter?: string, - includeClosed = false -): Promise { - if (!cfg.CELO_VAULT_ADDRESS || !cfg.CELO_GD_SUPERTOKEN_ADDRESS) { - logWarn("superfluid.streams.skipped", { - reason: "missing_config", - hasVaultAddress: Boolean(cfg.CELO_VAULT_ADDRESS), - hasSuperTokenAddress: Boolean(cfg.CELO_GD_SUPERTOKEN_ADDRESS), - senderFilter: redactAddress(senderFilter) - }); - return []; - } - - const endpoint = cfg.SUPERFLUID_SUBGRAPH_URL ?? SUPERFLUID_CELO_SUBGRAPH_URL; - const receiver = cfg.CELO_VAULT_ADDRESS.toLowerCase(); - const token = cfg.CELO_GD_SUPERTOKEN_ADDRESS.toLowerCase(); - const pageSize = 500; - let skip = 0; - const streams: SuperfluidIncomingStream[] = []; - let pages = 0; - - logInfo("superfluid.streams.fetch.start", { - endpoint, - receiver: redactAddress(receiver), - token: redactAddress(token), - senderFilter: redactAddress(senderFilter) - }); - - try { - while (true) { - const filters = ["receiver: $receiver", "token: $token"]; - if (!includeClosed) filters.push(`currentFlowRate_gt: "0"`); - if (senderFilter) filters.push("sender: $sender"); - const whereClause = `{ ${filters.join(", ")} }`; - const queryParams = senderFilter - ? `$receiver: String!, $token: String!, $first: Int!, $skip: Int!, $sender: String!` - : `$receiver: String!, $token: String!, $first: Int!, $skip: Int!`; - const variables: Record = { - receiver, - token, - first: pageSize, - skip - }; - if (senderFilter) variables.sender = senderFilter.toLowerCase(); - - const response = await fetch(endpoint, { - method: "POST", - headers: { - "content-type": "application/json" - }, - body: JSON.stringify({ - query: ` - query ActiveIncomingStreams(${queryParams}) { - streams( - first: $first - skip: $skip - where: ${whereClause} - orderBy: updatedAtTimestamp - orderDirection: desc - ) { - sender { id } - currentFlowRate - updatedAtTimestamp - flowUpdatedEvents(orderBy: timestamp, orderDirection: desc, first: 1) { - userData - } - } - } - `, - variables - }) - }); - - if (!response.ok) { - throw new Error(`Superfluid subgraph HTTP ${response.status}`); - } - - const body = await response.json(); - const parsed = SuperfluidStreamsResponseSchema.safeParse(body); - if (!parsed.success) { - throw new Error("invalid Superfluid subgraph response shape"); - } - - const batch = parsed.data.data.streams.map((stream) => { - const flowRateWeiPerSecond = stream.currentFlowRate; - const updatedAtSeconds = Number(stream.updatedAtTimestamp); - // A zero `updatedAtTimestamp` must not become 1970: this is the baseline for an account's - // first stream credit, so a 1970 value makes that credit window decades wide. - const hasUsableUpdatedAt = Number.isFinite(updatedAtSeconds) && updatedAtSeconds > 0; - const lastUpdateAt = hasUsableUpdatedAt ? new Date(updatedAtSeconds * 1000).toISOString() : new Date().toISOString(); - const rawUserData = stream.flowUpdatedEvents[0]?.userData; - const buyerAddress = decodeBuyerFromUserData(rawUserData); - return { - account: stream.sender.id.toLowerCase(), - flowRateWeiPerSecond, - lastUpdateAt, - buyerAddress - }; - }); - - streams.push(...batch); - pages += 1; - - if (batch.length < pageSize) break; - skip += pageSize; - } - } catch (error) { - const message = errorMessage(error); - logError("superfluid.streams.fetch.failed", { - endpoint, - receiver: redactAddress(receiver), - token: redactAddress(token), - senderFilter: redactAddress(senderFilter), - message - }); - return []; - } - - logInfo("superfluid.streams.fetch.end", { - endpoint, - senderFilter: redactAddress(senderFilter), - pages, - count: streams.length - }); - - return streams; -} +export { fundCredit }; diff --git a/backend/test/worker.test.ts b/backend/test/worker.test.ts index b70ccc2..bc7feb8 100644 --- a/backend/test/worker.test.ts +++ b/backend/test/worker.test.ts @@ -1,7 +1,8 @@ import test from "node:test"; import assert from "node:assert/strict"; import { Interface, Wallet } from "ethers"; -import worker, { fundCredit, streamElapsedSeconds } from "../src/worker.js"; +import worker, { fundCredit } from "../src/worker.js"; +import { streamElapsedSeconds } from "../src/stream-credits.js"; import { AntSeedFundingVaultClient } from "../src/antseed-funding-vault.js"; import { Env } from "../src/env.js"; import { KVCreditStore } from "../src/kv-credit-store.js"; From 3ecaad952addce7c76467eba76597cd0e5a51775 Mon Sep 17 00:00:00 2001 From: blueogin Date: Mon, 12 Oct 2026 02:33:29 -0400 Subject: [PATCH 2/2] refactor: update stream credit calculations to use constant-rate periods for accuracy --- backend/src/stream-credits.ts | 160 ++++++++++++++++----------------- backend/src/superfluid.ts | 46 ++++++++++ backend/src/worker.ts | 49 +++++----- backend/test/worker.test.ts | 164 ++++++++++++++++++++++++++++++++-- 4 files changed, 306 insertions(+), 113 deletions(-) diff --git a/backend/src/stream-credits.ts b/backend/src/stream-credits.ts index d16ae6d..fdb01e8 100644 --- a/backend/src/stream-credits.ts +++ b/backend/src/stream-credits.ts @@ -6,7 +6,7 @@ import { KVCreditStore } from "./kv-credit-store.js"; import { UserCreditProfile } from "./types.js"; import { logInfo, logWarn, redactAddress } from "./logging.js"; import { fetchSuperfluidIncomingStreams } from "./superfluid.js"; -import { SuperfluidIncomingStream } from "./superfluid.js"; +import { StreamPeriod, SuperfluidIncomingStream } from "./superfluid.js"; /** * Stream credit entries are keyed per account per day, so the scheduled run and a user's own @@ -47,44 +47,47 @@ export function streamElapsedSeconds(now: Date, lastStreamCreditAt: string | und } /** - * What a stream owes credit for, and over what window. + * What a stream owes credit for between `lastStreamCreditAt` and `now`, read from Superfluid's own + * constant-rate periods rather than reconstructed from a rate and a window. * - * Live stream: from the later of the last credit and the stream's last on-chain update, up to now, - * at the rate the subgraph reports. + * Each period carries one rate for its whole span, so `flowRate * overlap` is exact. Summing the + * overlaps makes every awkward case fall out for free: * - * Closed stream: from the last credit up to the **termination**, at the rate that was flowing before - * it closed. The end has to be the termination rather than `now` -- the stream stopped carrying G$ - * then, so measuring to `now` would invent time it did not flow. + * - a rate change is two periods, each priced at its own rate, so nothing is skipped or averaged; + * - a closed period stops at its termination, so time after a stream ended is never credited; + * - the gap between a closed stream and its replacement has no period, so it cannot be billed; + * - a stale `lastStreamCreditAt` cannot inflate anything, because only real periods are summed. * - * NOTE: this branch has neither a floor (a closed stream's last update *is* its termination) nor a - * ceiling, so a stale `lastStreamCreditAt` priced at a remembered rate can over-credit. Tracked in - * the PR notes -- reading `streamPeriods` is the fix, not another guard. + * With no previous credit the whole history is owed, which is correct -- it is what the stream + * actually carried. */ -export function streamCreditAmount(input: { - now: Date; - lastStreamCreditAt: string | undefined; - stream: SuperfluidIncomingStream; - /** The rate recorded before this run zeroed it; only consulted for a closed stream. */ - previousFlowRateWeiPerSecond: bigint; -}): { elapsedSeconds: number; gdAmountWei: bigint } { - const flowRateWeiPerSecond = BigInt(input.stream.flowRateWeiPerSecond); - if (flowRateWeiPerSecond > 0n) { - const elapsedSeconds = streamElapsedSeconds(input.now, input.lastStreamCreditAt, input.stream.lastUpdateAt); - return { elapsedSeconds, gdAmountWei: flowRateWeiPerSecond * BigInt(elapsedSeconds) }; +export function streamCreditFromPeriods(now: Date, lastStreamCreditAt: string | undefined, periods: StreamPeriod[]): bigint { + const nowMs = now.getTime(); + const lastCreditMs = lastStreamCreditAt ? Date.parse(lastStreamCreditAt) : 0; + if (lastStreamCreditAt && !Number.isFinite(lastCreditMs)) { + logWarn("stream.credits.unusable-baseline", { lastStreamCreditAt }); + return 0n; } - const terminatedAtMs = Date.parse(input.stream.lastUpdateAt); - const lastCreditMs = input.lastStreamCreditAt ? Date.parse(input.lastStreamCreditAt) : 0; - if (!Number.isFinite(terminatedAtMs) || terminatedAtMs <= 0 || !Number.isFinite(lastCreditMs) || lastCreditMs <= 0) { - logWarn("stream.credits.unusable-final-window", { - lastStreamCreditAt: input.lastStreamCreditAt, - terminatedAt: input.stream.lastUpdateAt - }); - return { elapsedSeconds: 0, gdAmountWei: 0n }; + let gdAmountWei = 0n; + for (const period of periods) { + const from = Math.max(period.startedAtMs, lastCreditMs); + const to = Math.min(period.stoppedAtMs ?? nowMs, nowMs); + if (to <= from) continue; + gdAmountWei += period.flowRateWeiPerSecond * BigInt(Math.floor((to - from) / 1000)); } + return gdAmountWei; +} - const elapsedSeconds = Math.max(0, Math.floor((terminatedAtMs - lastCreditMs) / 1000)); - return { elapsedSeconds, gdAmountWei: input.previousFlowRateWeiPerSecond * BigInt(elapsedSeconds) }; +/** + * Seconds since the account was last credited, used only for the 24-hour cooldown. The amount owed + * comes from the periods above; this gates how often a credit may be issued at all. + */ +export function secondsSinceLastCredit(now: Date, lastStreamCreditAt: string | undefined): number { + if (!lastStreamCreditAt) return Number.MAX_SAFE_INTEGER; + const lastCreditMs = Date.parse(lastStreamCreditAt); + if (!Number.isFinite(lastCreditMs)) return 0; + return Math.max(0, Math.floor((now.getTime() - lastCreditMs) / 1000)); } /** @@ -103,12 +106,10 @@ export async function runStreamCreditCycle( startedAt: number ): Promise { const gdPrice = await fetchCurrentGdPrice(cfg); - // Includes closed streams: a terminated stream reports `currentFlowRate = 0`, and seeing that is - // the only way the backend learns it stopped. The termination event that would otherwise say so - // is pushed in by `POST /v1/celo/events/record` and may never arrive. + // Includes closed streams: their periods carry the tail of what was streamed before termination, + // and their `currentFlowRate: 0` is how the backend learns the stream stopped. const streams = await fetchSuperfluidIncomingStreams(cfg); - let inactiveAccounts = 0; let syncedFlowRates = 0; let syncedRootFlowRates = 0; let skippedCooldown = 0; @@ -118,61 +119,57 @@ export async function runStreamCreditCycle( let failed = 0; const createdAt = new Date().toISOString(); - // Sync every account's recorded rate from the subgraph before crediting. Doing it here rather - // than inside `recordGdCredit` means it happens on every run, not only when a credit clears the - // cooldown and the minimum -- at a 4000 G$ minimum those can be a fortnight apart. Summed per - // account, because one account can hold several revisions (a closed stream and its replacement) - // and a closed row must not clobber the rate of the live one. All-closed sums to 0, which is - // how the profile stops advertising a stream that no longer exists. - const flowRateByAccount = new Map(); + // Collapse the subgraph rows to one entry per account. An account can hold several stream + // revisions -- a closed one and its replacement -- and they are credited together, because a + // credit entry is keyed per account per day. + type AccountStreams = { flowRateWeiPerSecond: bigint; periods: StreamPeriod[]; buyerAddress?: string }; + const byAccount = new Map(); for (const stream of streams) { const account = stream.account.toLowerCase(); - flowRateByAccount.set(account, (flowRateByAccount.get(account) ?? 0n) + BigInt(stream.flowRateWeiPerSecond)); + const current = byAccount.get(account) ?? { flowRateWeiPerSecond: 0n, periods: [] }; + current.flowRateWeiPerSecond += BigInt(stream.flowRateWeiPerSecond); + current.periods.push(...stream.periods); + // Rows arrive newest-first, so the first buyer seen is the most recently declared one. + current.buyerAddress ??= stream.buyerAddress; + byAccount.set(account, current); } - // One profile read per account, reused by the crediting pass below. `getSavedUser` rather than - // `getUser` because syncing is a correction to state that already exists -- an account that has - // never been credited has no profile, and the scheduled run must not create one as a side - // effect. Its first credit will create it, in `recordGdCredit`. + + // Sync each account's recorded rate before crediting, so the profile reflects reality on every + // run rather than only when a credit clears the gates. Summed per account: a closed row reports + // 0 and must not clobber a live replacement. All-closed sums to 0, which is how a profile stops + // advertising a stream that no longer exists. + // + // `getSavedUser` rather than `getUser`: syncing corrects state that already exists, and must not + // enrol an account that has never been credited. Its first credit creates the profile. const profileByAccount = new Map(); const flowRateByRoot = new Map(); - // The rate as it stood before this run's sync. A stream that has just closed reports 0, so its - // final window has to be priced at what was flowing beforehand. - const previousFlowRateByAccount = new Map(); - for (const [account, flowRate] of flowRateByAccount) { + for (const [account, streamsForAccount] of byAccount) { const profile = await store.getSavedUser(account); if (!profile) continue; - previousFlowRateByAccount.set(account, BigInt(profile.streamFlowRateWeiPerSecond)); - if (await store.recordStreamFlowRate(profile, flowRate)) syncedFlowRates += 1; + if (await store.recordStreamFlowRate(profile, streamsForAccount.flowRateWeiPerSecond)) syncedFlowRates += 1; profileByAccount.set(account, profile); - flowRateByRoot.set(profile.rootAccount, (flowRateByRoot.get(profile.rootAccount) ?? 0n) + flowRate); + flowRateByRoot.set(profile.rootAccount, (flowRateByRoot.get(profile.rootAccount) ?? 0n) + streamsForAccount.flowRateWeiPerSecond); } - // A GoodID root profile aggregates its identity's wallets. The totals get there by mirroring in - // `updateUser`, which works because they accumulate -- a flow rate is absolute, so it is summed - // here instead. A root that streams itself is part of its own sum, so this runs after the - // per-account pass and takes precedence over that account's own rate. + // A GoodID root profile aggregates its identity's wallets. The totals reach it by mirroring in + // `updateUser`, which works because they accumulate; a flow rate is absolute, so it is summed + // here instead. A root that streams itself is part of its own sum, so this runs second and wins. for (const [rootAccount, totalFlowRate] of flowRateByRoot) { const rootProfile = profileByAccount.get(rootAccount) ?? (await store.getSavedUser(rootAccount)); if (!rootProfile) continue; if (await store.recordStreamFlowRate(rootProfile, totalFlowRate)) syncedRootFlowRates += 1; } - inactiveAccounts = [...flowRateByAccount.values()].filter((flowRate) => flowRate === 0n).length; + const inactiveAccounts = [...byAccount.values()].filter((entry) => entry.flowRateWeiPerSecond === 0n).length; - for (const stream of streams) { - const account = stream.account.toLowerCase(); + for (const [account, streamsForAccount] of byAccount) { // Falls back to the synthesized default for an account streaming for the first time, which is // exactly the case the sync above skips. - const profile = profileByAccount.get(account) ?? (await store.getUser(stream.account)); + const profile = profileByAccount.get(account) ?? (await store.getUser(account)); const now = new Date(); - // Closed streams are credited too, for the window between the last credit and the termination - // -- without this the tail of every stream is silently forfeited. - const { elapsedSeconds, gdAmountWei } = streamCreditAmount({ - now, - lastStreamCreditAt: profile.lastStreamCreditAt, - stream, - previousFlowRateWeiPerSecond: previousFlowRateByAccount.get(account) ?? 0n - }); + const gdAmountWei = streamCreditFromPeriods(now, profile.lastStreamCreditAt, streamsForAccount.periods); if (gdAmountWei === 0n) continue; + + const elapsedSeconds = secondsSinceLastCredit(now, profile.lastStreamCreditAt); if (elapsedSeconds < 60 * 60 * 24) { skippedCooldown += 1; continue; @@ -181,28 +178,25 @@ export async function runStreamCreditCycle( skippedMinAmount += 1; continue; } - const rootAccount = await fetchGoodIdRoot(stream.account, cfg); - const buyerForOperatorCheck = stream.buyerAddress || stream.account; + + const rootAccount = await fetchGoodIdRoot(account, cfg); + const buyerForOperatorCheck = streamsForAccount.buyerAddress || account; const hasOperatorConsent = antseedFundingVault.enabled ? (await antseedFundingVault.isBuyerOperator(buyerForOperatorCheck)).isOperator : true; const isValidForBonus = !!rootAccount && hasOperatorConsent; - const depositId = createStreamFundingId(stream.account, new Date(createdAt)); const entry = await store.recordGdCredit({ - id: depositId, - account: stream.account, + id: createStreamFundingId(account, new Date(createdAt)), + account, rootAccount, source: "streamCron", - gdAmountWei: BigInt(gdAmountWei), - // The account's summed rate, not this row's: an account can hold several rows, and a closed - // one reports 0. Also what gives a first-time streamer -- skipped by the sync above for - // having no profile yet -- its rate on the profile this credit creates. - flowRate: flowRateByAccount.get(account) ?? 0n, + gdAmountWei, + flowRate: streamsForAccount.flowRateWeiPerSecond, isVerified: isValidForBonus, // if root acccount was found it is whitelisted & operator set gdPrice, maxBonusCapUsd: cfg.MAX_BONUS_CAP_USD, regularBonusBps: cfg.REGULAR_BONUS_BPS, streamingBonusBps: cfg.STREAMING_BONUS_BPS, - buyerAddress: stream.buyerAddress + buyerAddress: streamsForAccount.buyerAddress }); logInfo("streamCron.credits.entry", { entryId: entry.id, @@ -210,7 +204,8 @@ export async function runStreamCreditCycle( rootAccount: redactAddress(entry.rootAccount), buyer: redactAddress(entry.buyerAddress), elapsedSeconds, - stream, + periodCount: streamsForAccount.periods.length, + gdAmountWei: gdAmountWei.toString(), entry, profile }); @@ -224,6 +219,7 @@ export async function runStreamCreditCycle( } logInfo("cron.summary", { streamCount: streams.length, + accountCount: byAccount.size, processed, skippedCooldown, skippedMinAmount, diff --git a/backend/src/superfluid.ts b/backend/src/superfluid.ts index f4c4a1f..d064138 100644 --- a/backend/src/superfluid.ts +++ b/backend/src/superfluid.ts @@ -3,6 +3,9 @@ import { decodeBuyerFromUserData } from "./celo-events.js"; import { configFromEnv } from "./env.js"; import { errorMessage, logError, logInfo, logWarn, redactAddress } from "./logging.js"; +/** Upper bound on constant-rate spans fetched per stream; a truncated list is logged, not silently dropped. */ +const PERIOD_PAGE_SIZE = 1000; + /** Default endpoint, used when `SUPERFLUID_SUBGRAPH_URL` is unset. */ export const SUPERFLUID_CELO_SUBGRAPH_URL = "https://subgraph-endpoints.superfluid.dev/celo-mainnet/protocol-v1"; @@ -13,6 +16,18 @@ const SuperfluidStreamsResponseSchema = z.object({ sender: z.object({ id: z.string().regex(/^0x[0-9a-fA-F]{40}$/) }), currentFlowRate: z.string().regex(/^\d+$/), updatedAtTimestamp: z.string().regex(/^\d+$/), + // Optional with an empty default: a subgraph that stops returning the field, or returns a + // shape we do not expect, must degrade to "no periods, no credit" rather than failing the + // whole response and silently blinding the scheduled run to every stream. + streamPeriods: z + .array( + z.object({ + startedAtTimestamp: z.string().regex(/^\d+$/), + stoppedAtTimestamp: z.string().regex(/^\d+$/).nullable(), + flowRate: z.string().regex(/^\d+$/) + }) + ) + .default([]), flowUpdatedEvents: z.array( z.object({ userData: z.string() @@ -23,11 +38,25 @@ const SuperfluidStreamsResponseSchema = z.object({ }) }); +/** + * A span during which the stream carried a constant rate. Superfluid closes one and opens another + * on every rate change, so `flowRate * overlap` is exact for any sub-interval of a period -- no + * rate has to be remembered or assumed to have held. + */ +export type StreamPeriod = { + startedAtMs: number; + /** null while the period is still running. */ + stoppedAtMs: number | null; + flowRateWeiPerSecond: bigint; +}; + export type SuperfluidIncomingStream = { account: string; flowRateWeiPerSecond: string; /** For a closed stream this is its termination; for a live one, its last on-chain change. */ lastUpdateAt: string; + /** Constant-rate spans, newest first. The exact record of what this stream carried. */ + periods: StreamPeriod[]; /** AntSeed buyer decoded from the most recent FlowUpdatedEvent userdata. */ buyerAddress?: string; }; @@ -114,6 +143,11 @@ export async function fetchSuperfluidStreams( sender { id } currentFlowRate updatedAtTimestamp + streamPeriods(orderBy: startedAtTimestamp, orderDirection: desc, first: ${PERIOD_PAGE_SIZE}) { + startedAtTimestamp + stoppedAtTimestamp + flowRate + } flowUpdatedEvents(orderBy: timestamp, orderDirection: desc, first: 1) { userData } @@ -141,12 +175,24 @@ export async function fetchSuperfluidStreams( // first stream credit, so a 1970 value makes that credit window decades wide. const hasUsableUpdatedAt = Number.isFinite(updatedAtSeconds) && updatedAtSeconds > 0; const lastUpdateAt = hasUsableUpdatedAt ? new Date(updatedAtSeconds * 1000).toISOString() : new Date().toISOString(); + if (stream.streamPeriods.length === PERIOD_PAGE_SIZE) { + logWarn("superfluid.streams.periods-truncated", { + sender: redactAddress(stream.sender.id), + pageSize: PERIOD_PAGE_SIZE + }); + } + const periods: StreamPeriod[] = stream.streamPeriods.map((period) => ({ + startedAtMs: Number(period.startedAtTimestamp) * 1000, + stoppedAtMs: period.stoppedAtTimestamp === null ? null : Number(period.stoppedAtTimestamp) * 1000, + flowRateWeiPerSecond: BigInt(period.flowRate) + })); const rawUserData = stream.flowUpdatedEvents[0]?.userData; const buyerAddress = decodeBuyerFromUserData(rawUserData); return { account: stream.sender.id.toLowerCase(), flowRateWeiPerSecond, lastUpdateAt, + periods, buyerAddress }; }); diff --git a/backend/src/worker.ts b/backend/src/worker.ts index 8f0b452..0a4042a 100644 --- a/backend/src/worker.ts +++ b/backend/src/worker.ts @@ -7,7 +7,7 @@ import { errorMessage, logError, logInfo, redactAddress, redactHash } from "./lo import { getAnalyticsWindow } from "./analytics.js"; import { fundCredit } from "./funding.js"; import { cors, json, notifySlackOnRequestError, parseJson, readRequestBody } from "./http.js"; -import { createStreamFundingId, runStreamCreditCycle, streamCreditAmount, streamElapsedSeconds } from "./stream-credits.js"; +import { createStreamFundingId, runStreamCreditCycle, secondsSinceLastCredit, streamCreditFromPeriods } from "./stream-credits.js"; import { ANALYTICS_CRON_BACKFILL_DAYS, ANALYTICS_REFRESH_COOLDOWN_MS, @@ -15,7 +15,7 @@ import { backFillAnalytics, readAnalyticsRefreshTimestamp } from "./analytics-cron.js"; -import { SUPERFLUID_CELO_SUBGRAPH_URL, fetchSuperfluidStreamsForAccount } from "./superfluid.js"; +import { SUPERFLUID_CELO_SUBGRAPH_URL, StreamPeriod, fetchSuperfluidStreamsForAccount } from "./superfluid.js"; const CeloEventsRecordSchema = z .object({ txHash: z @@ -289,7 +289,7 @@ async function route(request: Request, env: Env, _ctx: ExecutionContext): Promis const recorded = []; const gdPrice = await fetchCurrentGdPrice(cfg); /** Per-request cache: one receipt can carry several stream events for the same account. */ - const streamLastUpdateAtByAccount = new Map(); + const periodsByAccount = new Map(); for (const event of events) { const rootAccount = await fetchGoodIdRoot(event.account, cfg); @@ -351,14 +351,17 @@ async function route(request: Request, env: Env, _ctx: ExecutionContext): Promis const profile = await store.getUser(event.account); // Lowercased: ethers hands back checksummed addresses, the subgraph lowercase ones. const streamCacheKey = event.account.toLowerCase(); - if (!streamLastUpdateAtByAccount.has(streamCacheKey)) { - const accountStreams = await fetchSuperfluidStreamsForAccount(event.account, cfg); - streamLastUpdateAtByAccount.set(streamCacheKey, accountStreams[0]?.lastUpdateAt); + if (!periodsByAccount.has(streamCacheKey)) { + const accountStreams = await fetchSuperfluidStreamsForAccount(event.account, cfg, true); + periodsByAccount.set( + streamCacheKey, + accountStreams.flatMap((accountStream) => accountStream.periods) + ); } - const streamLastUpdateAt = streamLastUpdateAtByAccount.get(streamCacheKey); - const elapsedSeconds = streamElapsedSeconds(new Date(), profile.lastStreamCreditAt, streamLastUpdateAt); - const owedWei = BigInt(profile.streamFlowRateWeiPerSecond) * BigInt(elapsedSeconds); - const gdAmountWei = owedWei < event.totalFlowWei ? owedWei : event.totalFlowWei; + const periods = periodsByAccount.get(streamCacheKey) ?? []; + const now = new Date(); + const gdAmountWei = streamCreditFromPeriods(now, profile.lastStreamCreditAt, periods); + const elapsedSeconds = secondsSinceLastCredit(now, profile.lastStreamCreditAt); const buyerForOperatorCheck = event.buyer || event.account; const hasOperatorConsent = antseedFundingVault.enabled && gdAmountWei > 0n ? (await antseedFundingVault.isBuyerOperator(buyerForOperatorCheck)).isOperator : true; @@ -386,8 +389,7 @@ async function route(request: Request, env: Env, _ctx: ExecutionContext): Promis rootAccount: redactAddress(entry.rootAccount), buyer: redactAddress(entry.buyerAddress), elapsedSeconds, - streamLastUpdateAt, - previousFlowRateWeiPerSecond: profile.streamFlowRateWeiPerSecond, + periodCount: periods.length, creditedGdAmountWei: gdAmountWei.toString(), reportedTotalFlowWei: event.totalFlowWei.toString() }); @@ -444,15 +446,11 @@ async function route(request: Request, env: Env, _ctx: ExecutionContext): Promis const gdPrice = await fetchCurrentGdPrice(cfg); const now = new Date(); - const previousFlowRateWeiPerSecond = BigInt(profile.streamFlowRateWeiPerSecond); const accountFlowRateWeiPerSecond = streams.reduce((sum, stream) => sum + BigInt(stream.flowRateWeiPerSecond), 0n); - // The widest window any of this account's streams is owed, used only for the cooldown gate -- - // each stream's own amount is computed per stream below. - const elapsedSeconds = Math.max( - ...streams.map( - (stream) => streamCreditAmount({ now, lastStreamCreditAt: profile.lastStreamCreditAt, stream, previousFlowRateWeiPerSecond }).elapsedSeconds - ) - ); + // One credit per account: the entry id is per account per day, and the owed amount is the sum + // over every constant-rate period across all of this account's stream revisions. + const accountPeriods = streams.flatMap((stream) => stream.periods); + const elapsedSeconds = secondsSinceLastCredit(now, profile.lastStreamCreditAt); if (elapsedSeconds < 60 * 60 * 24) { // if last credit was less than 24h ago, don't issue new credits to prevent abuse and return how many seconds are left until next credit can be issued @@ -469,19 +467,19 @@ async function route(request: Request, env: Env, _ctx: ExecutionContext): Promis const recorded = []; let skippedMinAmount = 0; - for (const stream of streams) { - const buyerForOperatorCheck = stream.buyerAddress || account; + { + const buyerForOperatorCheck = streams[0]?.buyerAddress || account; const hasOperatorConsent = antseedFundingVault.enabled ? (await antseedFundingVault.isBuyerOperator(buyerForOperatorCheck)).isOperator : true; const isValidForBonus = !!rootAccount && hasOperatorConsent; const isVerified = isValidForBonus; - const { gdAmountWei } = streamCreditAmount({ now, lastStreamCreditAt: profile.lastStreamCreditAt, stream, previousFlowRateWeiPerSecond }); + const stream = streams[0]; + const gdAmountWei = streamCreditFromPeriods(now, profile.lastStreamCreditAt, accountPeriods); if (gdAmountWei <= cfg.MIN_STREAM_BONUS_WEI) { skippedMinAmount += 1; recorded.push({ message: `stream credit amount ${gdAmountWei} is below minimum ${cfg.MIN_STREAM_BONUS_WEI}` }); - continue; - } + } else { const depositId = createStreamFundingId(account, now); const entry = await store.recordGdCredit({ @@ -509,6 +507,7 @@ async function route(request: Request, env: Env, _ctx: ExecutionContext): Promis }); const res = await fundCredit(entry, store, antseedFundingVault); recorded.push(res); + } } if (recorded.length === 0) { return json({ diff --git a/backend/test/worker.test.ts b/backend/test/worker.test.ts index bc7feb8..535d362 100644 --- a/backend/test/worker.test.ts +++ b/backend/test/worker.test.ts @@ -2,7 +2,7 @@ import test from "node:test"; import assert from "node:assert/strict"; import { Interface, Wallet } from "ethers"; import worker, { fundCredit } from "../src/worker.js"; -import { streamElapsedSeconds } from "../src/stream-credits.js"; +import { streamCreditFromPeriods, streamElapsedSeconds } from "../src/stream-credits.js"; import { AntSeedFundingVaultClient } from "../src/antseed-funding-vault.js"; import { Env } from "../src/env.js"; import { KVCreditStore } from "../src/kv-credit-store.js"; @@ -1243,7 +1243,8 @@ test("/v1/celo/events/record credits the owed window, not the event's totalFlowW ANTSEED_KV: kv as never, CELO_RPC_URL: "https://celo.rpc.local", CELO_VAULT_ADDRESS: celoVault, - CELO_GOODID_ADDRESS: goodIdAddr + CELO_GOODID_ADDRESS: goodIdAddr, + CELO_GD_SUPERTOKEN_ADDRESS: "0x0000000000000000000000000000000000000fed" }); // totalFlowWei = 2 G$, flowRate ≈ 1 G$/month, monthlyGdAmountWei = 1 G$ @@ -1257,6 +1258,28 @@ test("/v1/celo/events/record credits the owed window, not the event's totalFlowW globalThis.fetch = (async (_url: string | URL | Request, init?: RequestInit) => { const body = JSON.parse(String(init?.body)); + if (body.query) { + // The stream the credit is measured from: one open period, running for ten days. + return Response.json({ + data: { + streams: [ + { + sender: { id: account }, + currentFlowRate: previousFlowRate.toString(), + updatedAtTimestamp: Math.floor(Date.now() / 1000).toString(), + streamPeriods: [ + { + startedAtTimestamp: (Math.floor(Date.now() / 1000) - 10 * 86400).toString(), + stoppedAtTimestamp: null, + flowRate: previousFlowRate.toString() + } + ], + flowUpdatedEvents: [{ userData: "0x" }] + } + ] + } + }); + } if (body.method === "eth_call") { // GoodID root lookup — non-zero root means verified (32-byte ABI-encoded address) return Response.json({ @@ -1567,6 +1590,7 @@ test("/v1/celo/events/record sets stream bonus to 0 when buyer revoked operator" CELO_RPC_URL: "https://celo.rpc.local", CELO_VAULT_ADDRESS: celoVault, CELO_GOODID_ADDRESS: goodIdAddr, + CELO_GD_SUPERTOKEN_ADDRESS: "0x0000000000000000000000000000000000000fed", ANTSEED_FUNDING_RPC_URL: "https://base.rpc.local", ANTSEED_FUNDING_VAULT_ADDRESS: "0x0000000000000000000000000000000000000b01", ANTSEED_FUNDING_OPERATOR_PRIVATE_KEY: Wallet.createRandom().privateKey @@ -1594,6 +1618,25 @@ test("/v1/celo/events/record sets stream bonus to 0 when buyer revoked operator" globalThis.fetch = (async (_url: string | URL | Request, init?: RequestInit) => { const body = JSON.parse(String(init?.body)); + if (body.query) { + // One open period covering the owed window, so the credit is non-zero and the operator + // check is actually exercised. + return Response.json({ + data: { + streams: [ + { + sender: { id: account }, + currentFlowRate: "3472222222222222", + updatedAtTimestamp: Math.floor(Date.now() / 1000).toString(), + streamPeriods: [ + { startedAtTimestamp: (Math.floor(Date.now() / 1000) - 40 * 86400).toString(), stoppedAtTimestamp: null, flowRate: "3472222222222222" } + ], + flowUpdatedEvents: [{ userData: "0x" }] + } + ] + } + }); + } if (body.method === "eth_call") { return Response.json({ jsonrpc: "2.0", @@ -1714,6 +1757,7 @@ test("POST /v1/accounts/:account/stream-credits rate-limits when credits issued currentFlowRate: "1000000000000000", createdAtTimestamp: "1735000000", updatedAtTimestamp: "1735000000", + streamPeriods: [{ startedAtTimestamp: "1735000000", stoppedAtTimestamp: null, flowRate: "1" }], flowUpdatedEvents: [{ userData: "0x" }] } ] @@ -1765,6 +1809,7 @@ test("POST /v1/accounts/:account/stream-credits skips streams below minimum G$ a currentFlowRate: "1", createdAtTimestamp: "1735000000", updatedAtTimestamp: "1735000000", + streamPeriods: [{ startedAtTimestamp: "1735000000", stoppedAtTimestamp: null, flowRate: "1" }], flowUpdatedEvents: [{ userData: "0x" }] } ] @@ -1883,6 +1928,7 @@ test("POST /v1/accounts/:account/stream-credits does not credit the epoch on a z currentFlowRate: "3472222222222222", // 9000 G$/month, the real account's rate createdAtTimestamp: "0", updatedAtTimestamp: "0", + streamPeriods: [], flowUpdatedEvents: [{ userData: "0x" }] } ] @@ -2054,6 +2100,7 @@ test("/v1/celo/events/record floors the window at the stream's last update even currentFlowRate: staleRate.toString(), createdAtTimestamp: createdAtSeconds.toString(), updatedAtTimestamp: createdAtSeconds.toString(), + streamPeriods: [{ startedAtTimestamp: createdAtSeconds.toString(), stoppedAtTimestamp: null, flowRate: staleRate.toString() }], flowUpdatedEvents: [{ userData: "0x" }] } ] @@ -2138,6 +2185,7 @@ test("scheduled run zeroes the recorded flow rate once a stream closes", { concu currentFlowRate: "0", createdAtTimestamp: (createdAtSeconds + revision * 1000).toString(), updatedAtTimestamp: (createdAtSeconds + revision * 2000).toString(), + streamPeriods: [], flowUpdatedEvents: [{ userData: "0x" }] })) } @@ -2205,6 +2253,7 @@ test("scheduled run keeps the live rate when an account also has a closed revisi currentFlowRate: liveRate.toString(), createdAtTimestamp: (nowSeconds - 3600).toString(), updatedAtTimestamp: (nowSeconds - 3600).toString(), + streamPeriods: [{ startedAtTimestamp: (nowSeconds - 3600).toString(), stoppedAtTimestamp: null, flowRate: liveRate.toString() }], flowUpdatedEvents: [{ userData: "0x" }] }, { @@ -2212,6 +2261,7 @@ test("scheduled run keeps the live rate when an account also has a closed revisi currentFlowRate: "0", createdAtTimestamp: (nowSeconds - 90 * 24 * 3600).toString(), updatedAtTimestamp: (nowSeconds - 60 * 24 * 3600).toString(), + streamPeriods: [], flowUpdatedEvents: [{ userData: "0x" }] } ] @@ -2257,6 +2307,7 @@ test("scheduled run does not create a profile for an uncredited streamer", { con currentFlowRate: "3433641975308641", createdAtTimestamp: (nowSeconds - 3600).toString(), updatedAtTimestamp: (nowSeconds - 3600).toString(), + streamPeriods: [{ startedAtTimestamp: (nowSeconds - 3600).toString(), stoppedAtTimestamp: null, flowRate: "3433641975308641" }], flowUpdatedEvents: [{ userData: "0x" }] } ] @@ -2328,6 +2379,7 @@ test("scheduled run sums the flow rate across a root's accounts", { concurrency: currentFlowRate: (rate as bigint).toString(), createdAtTimestamp: (nowSeconds - 3600).toString(), updatedAtTimestamp: (nowSeconds - 3600).toString(), + streamPeriods: [{ startedAtTimestamp: (nowSeconds - 3600).toString(), stoppedAtTimestamp: null, flowRate: (rate as bigint).toString() }], flowUpdatedEvents: [{ userData: "0x" }] })) } @@ -2382,12 +2434,30 @@ function seedStreamProfile(kv: MemoryKV, account: string, fields: { flowRate: bi ); } -function subgraphStream(account: string, flowRate: bigint, createdAtSeconds: number, updatedAtSeconds = createdAtSeconds) { +function streamPeriod(flowRate: bigint, startedAtSeconds: number, stoppedAtSeconds: number | null) { + return { + startedAtTimestamp: startedAtSeconds.toString(), + stoppedAtTimestamp: stoppedAtSeconds === null ? null : stoppedAtSeconds.toString(), + flowRate: flowRate.toString() + }; +} + +function subgraphStream( + account: string, + flowRate: bigint, + createdAtSeconds: number, + updatedAtSeconds = createdAtSeconds, + periods?: ReturnType[] +) { return { sender: { id: account }, currentFlowRate: flowRate.toString(), createdAtTimestamp: createdAtSeconds.toString(), updatedAtTimestamp: updatedAtSeconds.toString(), + // A live row is one open period at its rate. A closed row carries whatever it ran through + // before terminating, which a test must state explicitly -- `currentFlowRate` is 0 by then and + // says nothing about what flowed. + streamPeriods: periods ?? (flowRate > 0n ? [streamPeriod(flowRate, createdAtSeconds, null)] : []), flowUpdatedEvents: [{ userData: "0x" }] }; } @@ -2572,7 +2642,7 @@ test("scheduled run records the rate on a first-time streamer's new profile", { // window falls back to a possibly weeks-stale `lastStreamCreditAt`. With `/v1/celo/events/record` // unauthenticated, a small stream open for one cron tick and then closed was credited across the whole // stale window. The amount is capped at the event's on-chain `totalFlowWei`. -test("/v1/celo/events/record caps a termination credit at what actually flowed", { concurrency: false }, async () => { +test("/v1/celo/events/record credits a termination for exactly what the stream carried", { concurrency: false }, async () => { const account = "0x0000000000000000000000000000000000000ad3"; const buyer = "0x0000000000000000000000000000000000000aaa"; const txHash = `0x${"d".repeat(64)}`; @@ -2581,6 +2651,7 @@ test("/v1/celo/events/record caps a termination credit at what actually flowed", const actuallyFlowed = rate * 6n * 3600n; // open for one 6h cron interval const originalFetch = globalThis.fetch; const originalConsoleLog = console.log; + const closedAtSeconds = Math.floor(Date.now() / 1000); try { console.log = (() => {}) as typeof console.log; const kv = new MemoryKV(); @@ -2595,7 +2666,29 @@ test("/v1/celo/events/record caps a termination credit at what actually flowed", const closeLog = encodeVaultEventLog("StreamUpdated", [account, buyer, 0n, 0n, actuallyFlowed], celoVault, txHash, 0); globalThis.fetch = (async (_url: string | URL | Request, init?: RequestInit) => { const body = JSON.parse(String(init?.body ?? "{}")); - if (body.query?.includes("streams")) return Response.json({ data: { streams: [] } }); + if (body.query?.includes("streams")) { + // The closed stream with the single period it ran through. This is what bounds the credit + // now -- no ceiling is applied, because nothing is estimated. + return Response.json({ + data: { + streams: [ + { + sender: { id: account }, + currentFlowRate: "0", + updatedAtTimestamp: closedAtSeconds.toString(), + streamPeriods: [ + { + startedAtTimestamp: (closedAtSeconds - 6 * 3600).toString(), + stoppedAtTimestamp: closedAtSeconds.toString(), + flowRate: rate.toString() + } + ], + flowUpdatedEvents: [{ userData: "0x" }] + } + ] + } + }); + } if (body.method === "eth_call") { return Response.json({ jsonrpc: "2.0", id: body.id, result: "0x000000000000000000000000abababababababababababababababababababab" }); } @@ -2613,9 +2706,68 @@ test("/v1/celo/events/record caps a termination credit at what actually flowed", ); const body = (await res.json()) as { events: Array<{ gdAmountWei: string }> }; - assert.equal(BigInt(body.events[0].gdAmountWei), actuallyFlowed, "credit must not exceed the G$ the stream carried"); + assert.equal(BigInt(body.events[0].gdAmountWei), actuallyFlowed, "the credit is exactly the G$ the stream carried"); } finally { globalThis.fetch = originalFetch; console.log = originalConsoleLog; } }); + +// A closed period must stop contributing at its termination. Without this, time after a stream +// ended is credited as though it were still flowing -- the failure mode the whole periods switch +// exists to make impossible. +test("streamCreditFromPeriods stops a closed period at its termination", () => { + const now = new Date("2026-03-01T00:00:00.000Z"); + const rate = 1_000_000_000_000n; + const startedAtMs = Date.parse("2026-01-01T00:00:00.000Z"); + const stoppedAtMs = Date.parse("2026-01-11T00:00:00.000Z"); // ran ten days, then closed + + const credited = streamCreditFromPeriods(now, "2026-01-01T00:00:00.000Z", [ + { startedAtMs, stoppedAtMs, flowRateWeiPerSecond: rate } + ]); + + assert.equal(credited, rate * BigInt(10 * 86400), "only the ten days it actually ran"); + // Measuring to `now` instead would be six times larger. + assert.ok(rate * BigInt(60 * 86400) > credited * 5n); +}); + +test("streamCreditFromPeriods prices each rate change at its own rate", () => { + const now = new Date("2026-02-01T00:00:00.000Z"); + const firstRate = 1_000_000_000_000n; + const secondRate = 3_000_000_000_000n; + + const credited = streamCreditFromPeriods(now, "2026-01-01T00:00:00.000Z", [ + { startedAtMs: Date.parse("2026-01-01T00:00:00.000Z"), stoppedAtMs: Date.parse("2026-01-11T00:00:00.000Z"), flowRateWeiPerSecond: firstRate }, + { startedAtMs: Date.parse("2026-01-11T00:00:00.000Z"), stoppedAtMs: null, flowRateWeiPerSecond: secondRate } + ]); + + assert.equal(credited, firstRate * BigInt(10 * 86400) + secondRate * BigInt(21 * 86400)); +}); + +// The dormant gap between a closed stream and its replacement has no period, so there is nothing +// to bill -- no floor or ceiling required. +test("streamCreditFromPeriods ignores the gap between a closed stream and its replacement", () => { + const now = new Date("2026-03-01T00:00:00.000Z"); + const rate = 1_000_000_000_000n; + + const credited = streamCreditFromPeriods(now, "2026-01-01T00:00:00.000Z", [ + { startedAtMs: Date.parse("2026-01-01T00:00:00.000Z"), stoppedAtMs: Date.parse("2026-01-06T00:00:00.000Z"), flowRateWeiPerSecond: rate }, + { startedAtMs: Date.parse("2026-02-24T00:00:00.000Z"), stoppedAtMs: null, flowRateWeiPerSecond: rate } + ]); + + // Five days, then a 49-day gap carrying nothing, then five days. + assert.equal(credited, rate * BigInt(10 * 86400)); +}); + +// The last credit usually falls in the middle of a period, not on its edge. Without clamping the +// start, every run would re-credit the whole history from the stream's beginning. +test("streamCreditFromPeriods credits only the part of a period after the last credit", () => { + const now = new Date("2026-02-01T00:00:00.000Z"); + const rate = 1_000_000_000_000n; + + const credited = streamCreditFromPeriods(now, "2026-01-22T00:00:00.000Z", [ + { startedAtMs: Date.parse("2026-01-01T00:00:00.000Z"), stoppedAtMs: null, flowRateWeiPerSecond: rate } + ]); + + assert.equal(credited, rate * BigInt(10 * 86400), "ten days since the last credit, not 31 since the stream opened"); +});