diff --git a/README.md b/README.md index 94b9c752..80aa646e 100644 --- a/README.md +++ b/README.md @@ -258,6 +258,9 @@ const res = await fetch( const data = await res.json(); console.log(data); +// OData-style filters and projections are also supported: +// `http://localhost:3000/transfers/address/${ADDRESS}?$filter=ledger gt 1000 and contains(contractId,'CB64')&$select=contractId,amount&cursor=...` + // Expected response (same shape as /transfers/incoming — adds "direction" per row) // { // "total": 85, @@ -401,6 +404,8 @@ curl "http://localhost:3000/transfers/tx/abcdef1234567890..." | `STELLAR_NETWORK` | — | `testnet` or `mainnet`. Testnet auto-configures the default RPC URL. | | `SOROBAN_RPC_URL` | *(see below)* | Soroban RPC endpoint. Overrides any network default. Required when `STELLAR_NETWORK=mainnet`. | | `STELLAR_RPC_URL` | — | Backward-compat alias for `SOROBAN_RPC_URL`. Used when `SOROBAN_RPC_URL` is unset. | +| `HORIZON_URL` | — | Optional Horizon endpoint used as a fallback source when RPC is unhealthy. | +| `HORIZON_EVENTS_PATH` | `/events` | Horizon contract-events path used by the fallback source. | | `START_LEDGER` | *(tip)* | Ledger to start indexing from. Leave blank to resume from DB state or start near the tip. | | `POLL_INTERVAL_MS` | `6000` | Polling interval in ms (\~1 ledger ≈ 6 s) | | `CONTRACT_IDS` | *(all)* | Comma-separated token contract IDs to watch. Empty = watch all (very heavy on mainnet) | @@ -418,6 +423,10 @@ Wraith resolves the RPC endpoint in this order and fails fast at startup if noth 4. `STELLAR_NETWORK=mainnet` → **error**: requires explicit `SOROBAN_RPC_URL` 5. Nothing set → **error**: clear message explaining what to configure +### Indexer Source Fallback + +If `HORIZON_URL` is set, the indexer checks the RPC source first and switches to Horizon when the RPC health check fails. It switches back automatically once RPC becomes healthy again. + ### Mainnet RPC Providers | Provider | URL pattern | diff --git a/src/__tests__/indexerSources.test.ts b/src/__tests__/indexerSources.test.ts new file mode 100644 index 00000000..d440a139 --- /dev/null +++ b/src/__tests__/indexerSources.test.ts @@ -0,0 +1,55 @@ +jest.mock("../rpc", () => ({ + getLatestLedger: jest.fn(), + fetchEventsSafe: jest.fn(), +})); + +import { createSourceSwitcherWithConfig } from "../indexer/sources"; +import { getLatestLedger, fetchEventsSafe } from "../rpc"; + +const mockGetLatestLedger = getLatestLedger as jest.MockedFunction; +const mockFetchEventsSafe = fetchEventsSafe as jest.MockedFunction; + +describe("Indexer source switcher", () => { + beforeEach(() => { + mockGetLatestLedger.mockReset(); + mockFetchEventsSafe.mockReset(); + }); + + it("falls back to Horizon when RPC is unhealthy", async () => { + mockGetLatestLedger.mockRejectedValue(new Error("rpc down")); + + const fetchImpl = jest.fn() + .mockResolvedValueOnce({ + ok: true, + json: async () => ({ _embedded: { records: [{ sequence: 123 }] } }), + }) + .mockResolvedValueOnce({ + ok: true, + json: async () => ({ + records: [ + { + id: "evt-1", + ledger: 122, + ledgerCloseTime: "2025-01-01T00:00:00Z", + contractId: "C123", + txHash: "tx-1", + topic: [], + value: {}, + }, + ], + }), + }); + + const switcher = createSourceSwitcherWithConfig({ + horizonUrl: "https://horizon.example", + fetchImpl: fetchImpl as never, + }); + const result = await switcher.fetchEvents(120, 123, ["C123"], 50); + + expect(result.highestLedger).toBe(122); + expect(result.events).toHaveLength(1); + expect(result.events[0].contractId).toBe("C123"); + expect(fetchImpl).toHaveBeenCalledTimes(2); + expect(mockFetchEventsSafe).not.toHaveBeenCalled(); + }); +}); \ No newline at end of file diff --git a/src/__tests__/odata.test.ts b/src/__tests__/odata.test.ts new file mode 100644 index 00000000..ede06472 --- /dev/null +++ b/src/__tests__/odata.test.ts @@ -0,0 +1,29 @@ +import { parseODataFilter, parseODataSelect } from "../lib/odata"; + +describe("OData helper", () => { + const fields = { + contractId: { type: "string" as const }, + ledger: { type: "number" as const }, + ledgerClosedAt: { type: "date" as const }, + eventType: { type: "string" as const }, + }; + + it("parses a safe AND-only filter", () => { + expect(parseODataFilter("ledger gt 100 and contains(contractId,'C')", fields)).toEqual({ + AND: [ + { ledger: { gt: 100 } }, + { contractId: { contains: "C", mode: "insensitive" } }, + ], + }); + }); + + it("rejects unsafe or unsupported filter expressions", () => { + expect(() => parseODataFilter("ledger gt 100 or 1 eq 1", fields)).toThrow(/AND combinations/i); + expect(() => parseODataFilter("contains(ledger,'1')", fields)).toThrow(/string fields/i); + }); + + it("parses a projection list and rejects unknown fields", () => { + expect(parseODataSelect("contractId, ledger", ["contractId", "ledger"])) .toEqual(["contractId", "ledger"]); + expect(() => parseODataSelect("contractId, hacked", ["contractId"])) .toThrow(/Unsupported \$select field/i); + }); +}); \ No newline at end of file diff --git a/src/__tests__/routes/transfers.test.ts b/src/__tests__/routes/transfers.test.ts index 10778f09..9985e079 100644 --- a/src/__tests__/routes/transfers.test.ts +++ b/src/__tests__/routes/transfers.test.ts @@ -159,6 +159,33 @@ describe("Transfer route handlers", () => { ); }); + it("forwards OData filter, select, and cursor params", async () => { + mockQueryTransfers.mockResolvedValue({ + total: 1, + transfers: [makeTransfer({ amount: "10000000" })], + nextCursor: "cursor-1", + }); + + const res = await request(app) + .get(`/transfers/incoming/${ALICE}`) + .query({ + $filter: "ledger gt 1000 and contains(contractId,'C')", + $select: "contractId,amount", + cursor: "cursor-0", + }); + + expect(res.status).toBe(200); + expect(res.body.nextCursor).toBe("cursor-1"); + expect(mockQueryTransfers).toHaveBeenCalledWith( + expect.objectContaining({ + filter: "ledger gt 1000 and contains(contractId,'C')", + select: ["contractId", "amount"], + cursor: "cursor-0", + }) + ); + expect(res.body.transfers[0].displayAmount).toBe("1.0000000"); + }); + it("passes fromDate and toDate to queryTransfers", async () => { mockQueryTransfers.mockResolvedValue({ total: 2, transfers: SEED_TRANSFERS.slice(14, 16) }); @@ -343,14 +370,14 @@ describe("Transfer route handlers", () => { }); it("honours pagination params", async () => { - mockQueryAllTransfers.mockResolvedValue({ total: 20, transfers: [] }); + mockQueryAllTransfers.mockResolvedValue({ total: 20, transfers: [], nextCursor: "cursor-2" }); await request(app) .get(`/transfers/address/${ALICE}`) - .query({ limit: "10", offset: "5" }); + .query({ limit: "10", offset: "5", cursor: "cursor-1", $select: "contractId,direction" }); expect(mockQueryAllTransfers).toHaveBeenCalledWith( - expect.objectContaining({ limit: 10, offset: 5 }) + expect.objectContaining({ limit: 10, offset: 5, cursor: "cursor-1", select: ["contractId", "direction"] }) ); }); diff --git a/src/api.ts b/src/api.ts index 6fdc0df0..f85f727a 100644 --- a/src/api.ts +++ b/src/api.ts @@ -37,6 +37,11 @@ const withDisplay = (t: T) => ({ displayAmount: toDisplayAmount(t.amount), }); +function parseSelectQuery(value: unknown): string[] | undefined { + if (typeof value !== "string" || !value.trim()) return undefined; + return value.split(",").map((item) => item.trim()).filter(Boolean); +} + const VALID_EVENT_TYPES = new Set(["transfer", "mint", "burn", "clawback"]); // ── CSV utilities ───────────────────────────────────────────────────────────── @@ -216,7 +221,7 @@ export function createApp(): express.Application { async (req: Request, res: Response, next: NextFunction) => { try { const { address } = req.params; - const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset } = req.query; + const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset, cursor, $filter, $select } = req.query; const fromDateVal = parseDateParam(fromDate, res); if (fromDateVal === null) return; @@ -232,6 +237,9 @@ export function createApp(): express.Application { address, direction: "incoming", contractId: contractId as string | undefined, + filter: $filter as string | undefined, + select: parseSelectQuery($select), + cursor: cursor as string | undefined, fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined, toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined, fromDate: fromDateVal, @@ -241,7 +249,17 @@ export function createApp(): express.Application { offset: off, }); - res.json({ ...result, transfers: result.transfers.map(withDisplay), limit: lim, offset: off }); + res.json({ + ...result, + transfers: result.transfers.map((transfer) => { + if (transfer && typeof (transfer as { amount?: unknown }).amount === "string") { + return withDisplay(transfer as { amount: string }); + } + return transfer; + }), + limit: lim, + offset: off, + }); } catch (err) { next(err); } @@ -258,7 +276,7 @@ export function createApp(): express.Application { async (req: Request, res: Response, next: NextFunction) => { try { const { address } = req.params; - const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset } = req.query; + const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset, cursor, $filter, $select } = req.query; const fromDateVal = parseDateParam(fromDate, res); if (fromDateVal === null) return; @@ -274,6 +292,9 @@ export function createApp(): express.Application { address, direction: "outgoing", contractId: contractId as string | undefined, + filter: $filter as string | undefined, + select: parseSelectQuery($select), + cursor: cursor as string | undefined, fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined, toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined, fromDate: fromDateVal, @@ -283,7 +304,17 @@ export function createApp(): express.Application { offset: off, }); - res.json({ ...result, transfers: result.transfers.map(withDisplay), limit: lim, offset: off }); + res.json({ + ...result, + transfers: result.transfers.map((transfer) => { + if (transfer && typeof (transfer as { amount?: unknown }).amount === "string") { + return withDisplay(transfer as { amount: string }); + } + return transfer; + }), + limit: lim, + offset: off, + }); } catch (err) { next(err); } @@ -311,7 +342,7 @@ export function createApp(): express.Application { async (req: Request, res: Response, next: NextFunction) => { try { const { address } = req.params; - const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset } = req.query; + const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset, cursor, $filter, $select } = req.query; const fromDateVal = parseDateParam(fromDate, res); if (fromDateVal === null) return; @@ -326,6 +357,9 @@ export function createApp(): express.Application { const result = await queryAllTransfers({ address, contractId: contractId as string | undefined, + filter: $filter as string | undefined, + select: parseSelectQuery($select), + cursor: cursor as string | undefined, fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined, toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined, fromDate: fromDateVal, @@ -335,7 +369,17 @@ export function createApp(): express.Application { offset: off, }); - res.json({ ...result, transfers: result.transfers.map(withDisplay), limit: lim, offset: off }); + res.json({ + ...result, + transfers: result.transfers.map((transfer) => { + if (transfer && typeof (transfer as { amount?: unknown }).amount === "string") { + return withDisplay(transfer as { amount: string }); + } + return transfer; + }), + limit: lim, + offset: off, + }); } catch (err) { next(err); } @@ -521,7 +565,7 @@ export function createApp(): express.Application { "/nfts/transfers", async (req: Request, res: Response, next: NextFunction) => { try { - const { contract, token_id, address, fromLedger, toLedger, limit, offset } = req.query; + const { contract, token_id, address, fromLedger, toLedger, limit, offset, cursor, $filter, $select } = req.query; const lim = parseIntParam(limit, 50); const off = parseIntParam(offset, 0); @@ -529,6 +573,9 @@ export function createApp(): express.Application { contractId: contract as string | undefined, tokenId: token_id as string | undefined, address: address as string | undefined, + filter: $filter as string | undefined, + select: parseSelectQuery($select), + cursor: cursor as string | undefined, fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined, toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined, limit: lim, diff --git a/src/db.ts b/src/db.ts index 16435fcc..dfa1a924 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1,5 +1,6 @@ import { PrismaClient, Prisma } from "@prisma/client"; import type { NftTransferRecord, NftMetadataPayload } from "./ingester/nft"; +import { decodeCursor, encodeCursor, parseODataFilter, parseODataSelect, projectRecord } from "./lib/odata"; // ─── Singleton Prisma client ────────────────────────────────────────────────── // Re-use one connection pool across the process. @@ -29,6 +30,114 @@ export interface TransferRecord { eventId: string; } +type ListPage = { + rows: T[]; + nextCursor: string | null; +}; + +function buildListPage(rows: T[], limit: number): ListPage { + if (rows.length <= limit) { + return { rows, nextCursor: null }; + } + + const pageRows = rows.slice(0, limit); + return { + rows: pageRows, + nextCursor: encodeCursor(pageRows[pageRows.length - 1].id), + }; +} + +function selectRows>( + rows: T[], + select: string[] | undefined, + derived: Record unknown> = {} +): Array> { + return rows.map((row) => projectRecord(row, select, derived)); +} + +const TRANSFER_SELECTABLE_FIELDS = [ + "id", + "contractId", + "eventType", + "fromAddress", + "toAddress", + "amount", + "ledger", + "ledgerClosedAt", + "txHash", + "eventId", + "createdAt", + "displayAmount", + "direction", +]; + +const NFT_TRANSFER_SELECTABLE_FIELDS = [ + "id", + "contractId", + "tokenId", + "fromAddress", + "toAddress", + "ledger", + "ledgerClosedAt", + "txHash", + "eventId", + "createdAt", +]; + +const ACCOUNT_SUMMARY_SELECTABLE_FIELDS = [ + "id", + "address", + "contractId", + "totalSent", + "totalReceived", + "net", + "txCount", + "lastActivityAt", + "updatedAt", + "displayTotalSent", + "displayTotalReceived", + "displayNet", +]; + +const TRANSFER_FIELD_TYPES = { + id: { type: "number" as const }, + contractId: { type: "string" as const }, + eventType: { type: "string" as const }, + fromAddress: { type: "string" as const }, + toAddress: { type: "string" as const }, + amount: { type: "string" as const }, + ledger: { type: "number" as const }, + ledgerClosedAt: { type: "date" as const }, + txHash: { type: "string" as const }, + eventId: { type: "string" as const }, + createdAt: { type: "date" as const }, +}; + +const NFT_TRANSFER_FIELD_TYPES = { + id: { type: "number" as const }, + contractId: { type: "string" as const }, + tokenId: { type: "string" as const }, + fromAddress: { type: "string" as const }, + toAddress: { type: "string" as const }, + ledger: { type: "number" as const }, + ledgerClosedAt: { type: "date" as const }, + txHash: { type: "string" as const }, + eventId: { type: "string" as const }, + createdAt: { type: "date" as const }, +}; + +const ACCOUNT_SUMMARY_FIELD_TYPES = { + id: { type: "number" as const }, + address: { type: "string" as const }, + contractId: { type: "string" as const }, + totalSent: { type: "string" as const }, + totalReceived: { type: "string" as const }, + net: { type: "string" as const }, + txCount: { type: "number" as const }, + lastActivityAt: { type: "date" as const }, + updatedAt: { type: "date" as const }, +}; + // ─── Upsert helper ──────────────────────────────────────────────────────────── /** * Idempotently insert a batch of transfer events. @@ -97,6 +206,9 @@ export type TransferQueryParams = { address: string; direction: "incoming" | "outgoing"; contractId?: string; + filter?: string; + select?: string[]; + cursor?: string; fromLedger?: number; toLedger?: number; fromDate?: Date; @@ -111,6 +223,9 @@ export async function queryTransfers(params: TransferQueryParams) { address, direction, contractId, + filter, + select, + cursor, fromLedger, toLedger, fromDate, @@ -120,7 +235,7 @@ export async function queryTransfers(params: TransferQueryParams) { offset = 0, } = params; - const where: Prisma.TokenTransferWhereInput = { + const baseWhere: Prisma.TokenTransferWhereInput = { ...(direction === "incoming" ? { toAddress: address } : { fromAddress: address }), ...(contractId ? { contractId } : {}), ...(eventTypes?.length ? { eventType: { in: eventTypes } } : {}), @@ -142,17 +257,51 @@ export async function queryTransfers(params: TransferQueryParams) { : {}), }; + const odataWhere = parseODataFilter(filter, TRANSFER_FIELD_TYPES); + const where: Prisma.TokenTransferWhereInput = odataWhere + ? { AND: [baseWhere, odataWhere as Prisma.TokenTransferWhereInput] } + : baseWhere; + + const requestedSelect = parseODataSelect(select?.join(","), TRANSFER_SELECTABLE_FIELDS); + const prismaSelect = requestedSelect + ? { + id: true, + contractId: requestedSelect.includes("contractId"), + eventType: requestedSelect.includes("eventType"), + fromAddress: requestedSelect.includes("fromAddress"), + toAddress: requestedSelect.includes("toAddress"), + amount: requestedSelect.includes("amount") || requestedSelect.includes("displayAmount"), + ledger: requestedSelect.includes("ledger"), + ledgerClosedAt: requestedSelect.includes("ledgerClosedAt"), + txHash: requestedSelect.includes("txHash"), + eventId: requestedSelect.includes("eventId"), + createdAt: requestedSelect.includes("createdAt"), + } + : undefined; + + const cap = Math.min(limit, 200); + const cursorId = decodeCursor(cursor); + const [total, transfers] = await prisma.$transaction([ prisma.tokenTransfer.count({ where }), prisma.tokenTransfer.findMany({ where, orderBy: [{ ledger: "desc" }, { id: "desc" }], - take: Math.min(limit, 200), // hard cap — no one needs 10k rows per request - skip: offset, + take: cap + 1, + ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : { skip: offset }), + ...(prismaSelect ? { select: prismaSelect } : {}), }), ]); - return { total, transfers }; + const page = buildListPage(transfers as Array<{ id: number }>, cap); + + return { + total, + transfers: selectRows(page.rows as Array>, requestedSelect, { + displayAmount: (row) => toDisplayAmount(String((row as { amount?: string }).amount)), + }), + nextCursor: page.nextCursor, + }; } export async function queryByTxHash(txHash: string) { @@ -243,6 +392,9 @@ export type NftTransferQueryParams = { contractId?: string; tokenId?: string; address?: string; + filter?: string; + select?: string[]; + cursor?: string; fromLedger?: number; toLedger?: number; limit?: number; @@ -254,13 +406,16 @@ export async function queryNftTransfers(params: NftTransferQueryParams) { contractId, tokenId, address, + filter, + select, + cursor, fromLedger, toLedger, limit = 50, offset = 0, } = params; - const where: Prisma.NftTransferWhereInput = { + const baseWhere: Prisma.NftTransferWhereInput = { ...(contractId ? { contractId } : {}), ...(tokenId ? { tokenId } : {}), ...(address ? { OR: [{ fromAddress: address }, { toAddress: address }] } : {}), @@ -274,18 +429,47 @@ export async function queryNftTransfers(params: NftTransferQueryParams) { : {}), }; + const odataWhere = parseODataFilter(filter, NFT_TRANSFER_FIELD_TYPES); + const where: Prisma.NftTransferWhereInput = odataWhere + ? { AND: [baseWhere, odataWhere as Prisma.NftTransferWhereInput] } + : baseWhere; + + const requestedSelect = parseODataSelect(select?.join(","), NFT_TRANSFER_SELECTABLE_FIELDS); + const prismaSelect = requestedSelect + ? { + id: true, + contractId: requestedSelect.includes("contractId"), + tokenId: requestedSelect.includes("tokenId"), + fromAddress: requestedSelect.includes("fromAddress"), + toAddress: requestedSelect.includes("toAddress"), + ledger: requestedSelect.includes("ledger"), + ledgerClosedAt: requestedSelect.includes("ledgerClosedAt"), + txHash: requestedSelect.includes("txHash"), + eventId: requestedSelect.includes("eventId"), + createdAt: requestedSelect.includes("createdAt"), + } + : undefined; + const cap = Math.min(limit, 200); + const cursorId = decodeCursor(cursor); const [total, transfers] = await prisma.$transaction([ prisma.nftTransfer.count({ where }), prisma.nftTransfer.findMany({ where, orderBy: [{ ledger: "desc" }, { id: "desc" }], - take: cap, - skip: offset, + take: cap + 1, + ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : { skip: offset }), + ...(prismaSelect ? { select: prismaSelect } : {}), }), ]); - return { total, transfers }; + const page = buildListPage(transfers as Array<{ id: number }>, cap); + + return { + total, + transfers: selectRows(page.rows as Array>, requestedSelect), + nextCursor: page.nextCursor, + }; } /** @@ -388,10 +572,77 @@ export async function getAccountSummary(address: string, contractId?: string) { }); } +export type AccountSummaryQueryParams = { + address: string; + contractId?: string; + filter?: string; + select?: string[]; + cursor?: string; + limit?: number; + offset?: number; +}; + +export async function queryAccountSummaries(params: AccountSummaryQueryParams) { + const { address, contractId, filter, select, cursor, limit = 50, offset = 0 } = params; + + const baseWhere: Prisma.AccountSummaryWhereInput = { + address, + ...(contractId ? { contractId } : {}), + }; + + const odataWhere = parseODataFilter(filter, ACCOUNT_SUMMARY_FIELD_TYPES); + const where: Prisma.AccountSummaryWhereInput = odataWhere + ? { AND: [baseWhere, odataWhere as Prisma.AccountSummaryWhereInput] } + : baseWhere; + + const requestedSelect = parseODataSelect(select?.join(","), ACCOUNT_SUMMARY_SELECTABLE_FIELDS); + const prismaSelect = requestedSelect + ? { + id: true, + address: requestedSelect.includes("address"), + contractId: requestedSelect.includes("contractId"), + totalSent: requestedSelect.includes("totalSent"), + totalReceived: requestedSelect.includes("totalReceived"), + net: requestedSelect.includes("net"), + txCount: requestedSelect.includes("txCount"), + lastActivityAt: requestedSelect.includes("lastActivityAt"), + updatedAt: requestedSelect.includes("updatedAt"), + } + : undefined; + + const cap = Math.min(limit, 200); + const cursorId = decodeCursor(cursor); + const [total, rows] = await prisma.$transaction([ + prisma.accountSummary.count({ where }), + prisma.accountSummary.findMany({ + where, + orderBy: [{ lastActivityAt: "desc" }, { id: "desc" }], + take: cap + 1, + ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : { skip: offset }), + ...(prismaSelect ? { select: prismaSelect } : {}), + }), + ]); + + const page = buildListPage(rows as Array<{ id: number }>, cap); + + return { + total, + transfers: selectRows(page.rows as Array>, requestedSelect, { + displayTotalSent: (row) => row.totalSent, + displayTotalReceived: (row) => row.totalReceived, + displayNet: (row) => row.net, + }), + nextCursor: page.nextCursor, + }; +} + // ─── Combined address query ─────────────────────────────────────────────────── export type AllTransfersQueryParams = { address: string; contractId?: string; + filter?: string; + select?: string[]; + cursor?: string; fromLedger?: number; toLedger?: number; fromDate?: Date; @@ -405,6 +656,9 @@ export async function queryAllTransfers(params: AllTransfersQueryParams) { const { address, contractId, + filter, + select, + cursor, fromLedger, toLedger, fromDate, @@ -414,7 +668,7 @@ export async function queryAllTransfers(params: AllTransfersQueryParams) { offset = 0, } = params; - const where: Prisma.TokenTransferWhereInput = { + const baseWhere: Prisma.TokenTransferWhereInput = { OR: [{ toAddress: address }, { fromAddress: address }], ...(contractId ? { contractId } : {}), ...(eventTypes?.length ? { eventType: { in: eventTypes } } : {}), @@ -436,22 +690,47 @@ export async function queryAllTransfers(params: AllTransfersQueryParams) { : {}), }; + const odataWhere = parseODataFilter(filter, TRANSFER_FIELD_TYPES); + const where: Prisma.TokenTransferWhereInput = odataWhere + ? { AND: [baseWhere, odataWhere as Prisma.TokenTransferWhereInput] } + : baseWhere; + const cap = Math.min(limit, 200); + const cursorId = decodeCursor(cursor); + const requestedSelect = parseODataSelect(select?.join(","), TRANSFER_SELECTABLE_FIELDS); + const prismaSelect = requestedSelect + ? { + id: true, + contractId: requestedSelect.includes("contractId"), + eventType: requestedSelect.includes("eventType"), + fromAddress: requestedSelect.includes("fromAddress"), + toAddress: requestedSelect.includes("toAddress"), + amount: requestedSelect.includes("amount") || requestedSelect.includes("displayAmount"), + ledger: requestedSelect.includes("ledger"), + ledgerClosedAt: requestedSelect.includes("ledgerClosedAt"), + txHash: requestedSelect.includes("txHash"), + eventId: requestedSelect.includes("eventId"), + createdAt: requestedSelect.includes("createdAt"), + } + : undefined; const [total, rows] = await prisma.$transaction([ prisma.tokenTransfer.count({ where }), prisma.tokenTransfer.findMany({ where, orderBy: [{ ledger: "desc" }, { id: "desc" }], - take: cap, - skip: offset, + take: cap + 1, + ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : { skip: offset }), + ...(prismaSelect ? { select: prismaSelect } : {}), }), ]); - const transfers = rows.map((r) => ({ - ...r, - direction: r.toAddress === address ? "incoming" : "outgoing", - })); + const page = buildListPage(rows as Array<{ id: number }>, cap); + + const transfers = selectRows(page.rows as Array>, requestedSelect ? [...requestedSelect, "direction"] : undefined, { + displayAmount: (row) => toDisplayAmount(String((row as { amount?: string }).amount)), + direction: (row) => ((row as { toAddress?: string | null }).toAddress === address ? "incoming" : "outgoing"), + }); - return { total, transfers }; + return { total, transfers, nextCursor: page.nextCursor }; } diff --git a/src/indexer.ts b/src/indexer.ts index 20ac01f2..cf67eaf2 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -1,5 +1,5 @@ import "dotenv/config"; -import { fetchEventsSafe, getLatestLedger, withRetry, validateNetworkConfig } from "./rpc"; +import { validateNetworkConfig, withRetry } from "./rpc"; import { parseEvents } from "./decoder"; import { upsertTransfers, @@ -13,6 +13,7 @@ import { } from "./db"; import { emitTransfer } from "./events"; import { isNftTransferEvent, parseNftEvents, fetchNftMetadata } from "./ingester/nft"; +import { createSourceSwitcherWithConfig } from "./indexer/sources"; // ─── NFT Contract IDs ───────────────────────────────────────────────────────── /** @@ -73,6 +74,14 @@ const SAC_CONTRACT_IDS = resolveSacContractIds(); const NFT_CONTRACT_IDS = resolveNftContractIds(); // Combined watch list — deduplicated so we don't request the same contract twice const ALL_CONTRACT_IDS = [...new Set([...SAC_CONTRACT_IDS, ...NFT_CONTRACT_IDS])]; +const sourceSwitcher = createSourceSwitcherWithConfig({ + horizonUrl: process.env.HORIZON_URL, + horizonEventsPath: process.env.HORIZON_EVENTS_PATH, + fetchImpl: (globalThis as { fetch?: (input: string, init?: unknown) => Promise }).fetch as unknown as ( + input: string, + init?: { headers?: Record } + ) => Promise<{ ok: boolean; status: number; json(): Promise }>, +}); // Stellar testnet RPC retains ~7 days ≈ 120 000 ledgers (at ~5s per ledger). // We cap the back-fill look-back so we never request a ledger that's already pruned. @@ -111,9 +120,7 @@ async function pollOnce( `[indexer] Polling ledgers ${fromLedger} → ${latestLedger} (lag: ${latestLedger - fromLedger})` ); - // fetchEventsSafe bisects on XDR decode errors so newer protocol ledgers - // don't crash the whole indexer — they're skipped with a warning instead. - const { events, highestLedger } = await fetchEventsSafe( + const { events, highestLedger } = await sourceSwitcher.fetchEvents( fromLedger, latestLedger, ALL_CONTRACT_IDS, BATCH_SIZE ); @@ -195,7 +202,7 @@ export async function startIndexer(): Promise { startedAt = Date.now(); // ── Determine start ledger ────────────────────────────────────────────────── - const latestLedger = await withRetry(getLatestLedger); + const latestLedger = await withRetry(() => sourceSwitcher.getLatestLedger()); const minSafeLedger = latestLedger - RPC_MAX_LOOKBACK_LEDGERS; let currentLedger: number; @@ -218,7 +225,7 @@ export async function startIndexer(): Promise { // ── Polling loop ──────────────────────────────────────────────────────────── while (true) { try { - const tip = await withRetry(getLatestLedger); + const tip = await withRetry(() => sourceSwitcher.getLatestLedger()); const target = tip - TIP_LAG; if (currentLedger >= target) { diff --git a/src/indexer/sources/horizon.ts b/src/indexer/sources/horizon.ts new file mode 100644 index 00000000..6dd5ef08 --- /dev/null +++ b/src/indexer/sources/horizon.ts @@ -0,0 +1,121 @@ +import type { RawEvent } from "../../rpc"; +import type { EventSource } from "./rpc"; + +export type FetchLike = (input: string, init?: { headers?: Record }) => Promise<{ + ok: boolean; + status: number; + json(): Promise; +}>; + +export type HorizonSourceConfig = { + baseUrl: string; + eventPath: string; + fetchImpl: FetchLike; +}; + +type HorizonEventPayload = { + id?: string; + paging_token?: string; + ledger?: number; + ledger_sequence?: number; + ledgerClosedAt?: string; + ledger_close_time?: string; + contractId?: string; + contract_id?: string; + txHash?: string; + tx_hash?: string; + topic?: unknown[]; + topics?: unknown[]; + value?: unknown; +}; + +function resolveBaseUrl(): string { + throw new Error("[indexer] Horizon source configuration missing."); +} + +function buildUrl(baseUrl: string, path: string, params: Record): string { + const query = Object.entries(params) + .map(([key, value]) => `${encodeURIComponent(key)}=${encodeURIComponent(value)}`) + .join("&"); + return `${baseUrl}${path}${query ? `?${query}` : ""}`; +} + +function normalizeEvent(raw: HorizonEventPayload): RawEvent { + return { + id: raw.id ?? raw.paging_token ?? "", + type: "contract", + ledger: raw.ledger ?? raw.ledger_sequence ?? 0, + ledgerClosedAt: raw.ledgerClosedAt ?? raw.ledger_close_time ?? new Date().toISOString(), + contractId: raw.contractId ?? raw.contract_id ?? "", + txHash: raw.txHash ?? raw.tx_hash ?? "", + topic: (raw.topic ?? raw.topics ?? []) as RawEvent["topic"], + value: raw.value as RawEvent["value"], + }; +} + +async function fetchJson(url: string, fetchImpl: FetchLike): Promise { + if (!fetchImpl) { + throw new Error("[indexer] Global fetch is unavailable."); + } + + const response = await fetchImpl(url, { + headers: { Accept: "application/json" }, + }); + + if (!response.ok) { + throw new Error(`[indexer] Horizon request failed with ${response.status}`); + } + + return response.json() as Promise; +} + +export function createHorizonSource(config: HorizonSourceConfig): EventSource { + const { baseUrl, eventPath, fetchImpl } = config; + + return { + name: "horizon", + async isHealthy() { + try { + await this.getLatestLedger(); + return true; + } catch { + return false; + } + }, + async getLatestLedger() { + const url = buildUrl(baseUrl, "/ledgers", { order: "desc", limit: "1" }); + + const payload = await fetchJson<{ _embedded?: { records?: Array<{ sequence?: number }> } }>(url, fetchImpl); + const latest = payload._embedded?.records?.[0]?.sequence; + if (typeof latest !== "number") { + throw new Error("[indexer] Horizon ledger tip unavailable."); + } + return latest; + }, + async fetchEvents(startLedger, endLedger, contractIds, limit = 10_000) { + const url = buildUrl(baseUrl, eventPath, { + start_ledger: String(startLedger), + end_ledger: String(endLedger), + limit: String(limit), + }); + if (contractIds.length > 0) { + // Append rather than rebuild to keep the helper simple. + const contracted = `${url}${url.includes("?") ? "&" : "?"}contract_ids=${encodeURIComponent(contractIds.join(","))}`; + return fetchJson<{ records?: HorizonEventPayload[]; _embedded?: { records?: HorizonEventPayload[] } }>(contracted, fetchImpl).then((payload) => { + const records = payload.records ?? payload._embedded?.records ?? []; + const events = records.map(normalizeEvent).filter((event) => event.id && event.contractId); + const highestLedger = events.reduce((max, event) => Math.max(max, event.ledger), startLedger); + + return { events, highestLedger }; + }); + } + + const payload = await fetchJson<{ records?: HorizonEventPayload[]; _embedded?: { records?: HorizonEventPayload[] } }>(url, fetchImpl); + const records = payload.records ?? payload._embedded?.records ?? []; + const events = records.map(normalizeEvent).filter((event) => event.id && event.contractId); + const highestLedger = events.reduce((max, event) => Math.max(max, event.ledger), startLedger); + + return { events, highestLedger }; + }, + }; +} \ No newline at end of file diff --git a/src/indexer/sources/index.ts b/src/indexer/sources/index.ts new file mode 100644 index 00000000..75c00357 --- /dev/null +++ b/src/indexer/sources/index.ts @@ -0,0 +1,84 @@ +import type { RawEvent } from "../../rpc"; +import { createHorizonSource, type FetchLike, type HorizonSourceConfig } from "./horizon"; +import { createRpcSource, type EventSource } from "./rpc"; + +export type SourceSwitcherConfig = { + horizonUrl?: string; + horizonEventsPath?: string; + fetchImpl: FetchLike; +}; + +export interface SourceSwitcher { + getLatestLedger(): Promise; + fetchEvents( + startLedger: number, + endLedger: number, + contractIds: string[], + limit?: number + ): Promise<{ events: RawEvent[]; highestLedger: number }>; + getActiveSourceName(): Promise; +} + +function isTruthySource(source: EventSource | null): source is EventSource { + return source !== null; +} + +export function createSourceSwitcherWithConfig(config: SourceSwitcherConfig): SourceSwitcher { + const sources = [ + createRpcSource(), + config.horizonUrl + ? createHorizonSource({ + baseUrl: config.horizonUrl.replace(/\/$/, ""), + eventPath: config.horizonEventsPath ?? "/events", + fetchImpl: config.fetchImpl, + } satisfies HorizonSourceConfig) + : null, + ].filter(isTruthySource); + if (sources.length === 0) { + throw new Error("[indexer] No event sources configured."); + } + + let preferred = sources[0]; + + const pickHealthySource = async (): Promise => { + if (await preferred.isHealthy()) return preferred; + + for (const source of sources) { + if (await source.isHealthy()) { + preferred = source; + return source; + } + } + + throw new Error("[indexer] No healthy indexing source available."); + }; + + return { + async getLatestLedger() { + return (await pickHealthySource()).getLatestLedger(); + }, + async fetchEvents(startLedger, endLedger, contractIds, limit) { + const source = await pickHealthySource(); + try { + return await source.fetchEvents(startLedger, endLedger, contractIds, limit); + } catch (error) { + for (const fallback of sources) { + if (fallback.name === source.name) continue; + if (!(await fallback.isHealthy())) continue; + + preferred = fallback; + return fallback.fetchEvents(startLedger, endLedger, contractIds, limit); + } + + throw error; + } + }, + async getActiveSourceName() { + return (await pickHealthySource()).name; + }, + }; +} + +export function createSourceSwitcher(config: SourceSwitcherConfig): SourceSwitcher { + return createSourceSwitcherWithConfig(config); +} \ No newline at end of file diff --git a/src/indexer/sources/rpc.ts b/src/indexer/sources/rpc.ts new file mode 100644 index 00000000..3900f949 --- /dev/null +++ b/src/indexer/sources/rpc.ts @@ -0,0 +1,31 @@ +import { fetchEventsSafe, getLatestLedger, type RawEvent } from "../../rpc"; + +export interface EventSource { + name: string; + isHealthy(): Promise; + getLatestLedger(): Promise; + fetchEvents( + startLedger: number, + endLedger: number, + contractIds: string[], + limit?: number + ): Promise<{ events: RawEvent[]; highestLedger: number }>; +} + +export function createRpcSource(): EventSource { + return { + name: "rpc", + async isHealthy() { + try { + await getLatestLedger(); + return true; + } catch { + return false; + } + }, + getLatestLedger, + fetchEvents(startLedger, endLedger, contractIds, limit) { + return fetchEventsSafe(startLedger, endLedger, contractIds, limit); + }, + }; +} \ No newline at end of file diff --git a/src/lib/odata.ts b/src/lib/odata.ts new file mode 100644 index 00000000..aa439863 --- /dev/null +++ b/src/lib/odata.ts @@ -0,0 +1,235 @@ +export type ODataFieldType = "string" | "number" | "date"; + +type FieldDefinition = { + type: ODataFieldType; +}; + +type ComparisonOperator = "eq" | "gt" | "lt"; + +type ParsedClause = + | { kind: "comparison"; field: string; operator: ComparisonOperator; value: unknown } + | { kind: "contains"; field: string; value: string }; + +function splitAndClauses(filter: string): string[] { + const clauses: string[] = []; + let current = ""; + let depth = 0; + let inString = false; + + for (let index = 0; index < filter.length; index++) { + const char = filter[index]; + const next = filter[index + 1]; + + if (char === "'") { + current += char; + if (inString && next === "'") { + current += next; + index++; + continue; + } + inString = !inString; + continue; + } + + if (!inString) { + if (char === "(") depth++; + if (char === ")") depth = Math.max(0, depth - 1); + + if (depth === 0 && filter.slice(index, index + 4).toLowerCase() === " and") { + clauses.push(current.trim()); + current = ""; + index += 3; + continue; + } + } + + current += char; + } + + if (current.trim()) clauses.push(current.trim()); + return clauses; +} + +function parseStringLiteral(raw: string): string { + if (!raw.startsWith("'") || !raw.endsWith("'")) { + throw new Error("String values must be wrapped in single quotes."); + } + + return raw.slice(1, -1).replace(/''/g, "'"); +} + +function parseClause(rawClause: string): ParsedClause { + const containsMatch = rawClause.match(/^contains\(\s*([A-Za-z_][A-Za-z0-9_\.]*)\s*,\s*('(?:''|[^'])*')\s*\)$/i); + if (containsMatch) { + return { + kind: "contains", + field: containsMatch[1], + value: parseStringLiteral(containsMatch[2]), + }; + } + + const comparisonMatch = rawClause.match(/^([A-Za-z_][A-Za-z0-9_\.]*)\s+(eq|gt|lt)\s+(.+)$/i); + if (!comparisonMatch) { + throw new Error(`Unsupported $filter clause: ${rawClause}`); + } + + const [, field, operator, rawValue] = comparisonMatch; + return { + kind: "comparison", + field, + operator: operator.toLowerCase() as ComparisonOperator, + value: rawValue.trim(), + }; +} + +function parseComparisonValue(field: string, rawValue: unknown, definition: FieldDefinition): unknown { + if (definition.type === "string") { + if (typeof rawValue !== "string") throw new Error(`Invalid value for ${field}.`); + return parseStringLiteral(rawValue.trim()); + } + + if (definition.type === "number") { + const normalized = String(rawValue).trim(); + if (/^-?\d+$/.test(normalized)) return Number(normalized); + if (normalized.startsWith("'") && normalized.endsWith("'")) { + const unquoted = parseStringLiteral(normalized); + if (/^-?\d+$/.test(unquoted)) return Number(unquoted); + } + throw new Error(`Invalid numeric value for ${field}.`); + } + + const normalized = String(rawValue).trim(); + const candidate = normalized.startsWith("'") && normalized.endsWith("'") + ? parseStringLiteral(normalized) + : normalized; + const date = new Date(candidate); + if (Number.isNaN(date.getTime())) { + throw new Error(`Invalid date value for ${field}.`); + } + return date; +} + +export function parseODataFilter( + filter: string | undefined, + fields: Record +): Record | undefined { + const normalized = filter?.trim(); + if (!normalized) return undefined; + + if (/\bor\b/i.test(normalized)) { + throw new Error("$filter only supports AND combinations."); + } + + const clauses = splitAndClauses(normalized); + if (clauses.length === 0) return undefined; + + const filters = clauses.map((clause) => { + const parsed = parseClause(clause); + const definition = fields[parsed.field]; + if (!definition) { + throw new Error(`Unsupported $filter field: ${parsed.field}`); + } + + if (parsed.kind === "contains") { + if (definition.type !== "string") { + throw new Error(`contains() is only supported for string fields: ${parsed.field}`); + } + + return { + [parsed.field]: { + contains: parsed.value, + mode: "insensitive", + }, + }; + } + + const value = parseComparisonValue(parsed.field, parsed.value, definition); + + if (parsed.operator === "eq") { + return { [parsed.field]: value }; + } + + return { + [parsed.field]: { + [parsed.operator]: value, + }, + }; + }); + + return filters.length === 1 ? filters[0] : { AND: filters }; +} + +export function parseODataSelect(select: string | undefined, allowedFields: string[]): string[] | undefined { + const normalized = select?.trim(); + if (!normalized) return undefined; + + const requested = normalized + .split(",") + .map((field) => field.trim()) + .filter(Boolean); + + if (requested.length === 0) return undefined; + + const allowed = new Set(allowedFields); + const result: string[] = []; + for (const field of requested) { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(field)) { + throw new Error(`Unsupported $select field: ${field}`); + } + if (!allowed.has(field)) { + throw new Error(`Unsupported $select field: ${field}`); + } + if (!result.includes(field)) result.push(field); + } + + return result; +} + +export function encodeCursor(id: number): string { + return Buffer.from(JSON.stringify({ id }), "utf8").toString("base64url"); +} + +export function decodeCursor(cursor: string | undefined): number | undefined { + const normalized = cursor?.trim(); + if (!normalized) return undefined; + + try { + const decoded = JSON.parse(Buffer.from(normalized, "base64url").toString("utf8")); + if (typeof decoded.id === "number" && Number.isInteger(decoded.id)) { + return decoded.id; + } + } catch { + // Treat malformed cursors as absent so the route can fall back cleanly. + } + + return undefined; +} + +export function projectRecord>( + record: T, + select: string[] | undefined, + derived: Record unknown> = {} +): Record { + const projected: Record = {}; + const selectedFields = select?.length ? select : Object.keys(record); + + for (const field of selectedFields) { + if (field in record) { + projected[field] = record[field]; + continue; + } + + const compute = derived[field]; + if (compute) { + projected[field] = compute(record); + } + } + + if (!select?.length) { + for (const [field, compute] of Object.entries(derived)) { + projected[field] = compute(record); + } + } + + return projected; +} \ No newline at end of file