feat: property tests, e2e ingest→query test, parallel ingest, host-fn log

package.json

@@ -27,6 +27,9 @@
       "**/__tests__/**/*.test.ts",
       "**/tests/**/*.test.ts"
     ],
+    "transform": {
+      "^.+\\.tsx?$": ["ts-jest", { "tsconfig": "tsconfig.test.json" }]
+    },
     "moduleFileExtensions": [
       "ts",
       "js",
@@ -59,6 +62,7 @@
     "supertest": "^7.2.2",
     "ts-jest": "^29.4.9",
     "ts-node-dev": "^2.0.0",
+    "fast-check": "^3.22.0",
     "typedoc": "^0.28.19",
     "typescript": "^5.4.2"
   }

prisma/schema.prisma

@@ -38,6 +38,46 @@ model TokenTransfer {
   @@schema("wraith")
 }
 
+// ─── Host-function invocation log ────────────────────────────────────────────
+// One row per contract event — includes token events and all other contracts.
+// Allows downstream consumers to interpret events from arbitrary contracts.
+model HostFnLog {
+  id           Int      @id @default(autoincrement())
+
+  // The contract that emitted the event (C...)
+  contractId   String
+
+  // topics[0] decoded as a symbol string (e.g. "transfer", "swap", "deposit")
+  functionName String
+
+  // topics[1..n] serialised via scValToNative — BigInt values become strings
+  args         Json
+
+  // Event value serialised via scValToNative; null if the value is scvVoid
+  result       Json?
+
+  // Gas consumed by the invocation — populated externally when tx metadata is
+  // available; null otherwise
+  gasUsed      BigInt?
+
+  // Ledger sequence number where the event was emitted
+  ledger       Int
+
+  // UTC close time of the ledger
+  ledgerClosedAt DateTime
+
+  // Transaction hash (SHA-256 hex, no 0x prefix)
+  txHash       String
+
+  // Stellar RPC paging token — unique per event, used for deduplication
+  eventId      String   @unique
+
+  createdAt    DateTime @default(now())
+
+  @@index([contractId])
+  @@index([contractId, functionName])
+  @@index([ledger])
+  @@index([txHash])
 // ─── NFT Transfers ────────────────────────────────────────────────────────────
 model NftTransfer {
   id             Int      @id @default(autoincrement())

src/api.ts

@@ -1,6 +1,8 @@
 import express, { Request, Response, NextFunction } from "express";
 import cors from "cors";
 import rateLimit from "express-rate-limit";
+import { queryTransfers, queryAllTransfers, queryByTxHash, querySummary, getLastIndexedLedger, prisma } from "./db";
+import { queryHostFnLogs } from "./indexer/host-fn-log";
 import { queryTransfers, queryAllTransfers, queryByTxHash, querySummary, queryNftTransfers, getNftOwner, getNftMetadata, getLastIndexedLedger, prisma } from "./db";
 import { getLatestLedger } from "./rpc";
 import { getIndexerStats } from "./indexer";
@@ -553,6 +555,40 @@ export function createApp(): express.Application {
     }
   );
 
+  // ── GET /host-fn/:contractId ─────────────────────────────────────────────────
+  /**
+   * Query raw host-function invocation logs for a contract.
+   *
+   * Every contract event indexed by Wraith is stored here — not just SEP-41
+   * token events — so downstream consumers can interpret arbitrary contracts.
+   *
+   * Query params:
+   *   functionName  — filter by function name (e.g. "swap")
+   *   limit         — max rows (default 50, hard cap 200)
+   *   offset        — pagination offset (default 0)
+   */
+  app.get(
+    "/host-fn/:contractId",
+    async (req: Request, res: Response, next: NextFunction) => {
+      try {
+        const { contractId } = req.params;
+        const functionName = req.query.functionName as string | undefined;
+        const limit  = parseIntParam(req.query.limit,  50);
+        const offset = parseIntParam(req.query.offset, 0);
+
+        const { total, logs } = await queryHostFnLogs({
+          contractId,
+          functionName,
+          limit,
+          offset,
+        });
+
+        res.json({
+          contractId,
+          total,
+          limit: Math.min(limit, 200),
+          offset,
+          logs,
   // ── GET /nfts/transfers ──────────────────────────────────────────────────────
   /**
    * Query CAP-46 NFT transfer events.
@@ -638,6 +674,7 @@ export function createApp(): express.Application {
       } catch (err) {
         next(err);
       }
+    },
     }
   );
 

src/indexer.ts

@@ -12,6 +12,8 @@ import {
   pruneOldTransfers,
 } from "./db";
 import { emitTransfer } from "./events";
+import { parseHostFnEvent, upsertHostFnLogs, type HostFnRecord } from "./indexer/host-fn-log";
+import { pollParallel } from "./indexer/parallel";
 import { isNftTransferEvent, parseNftEvents, fetchNftMetadata } from "./ingester/nft";
 import { createSourceSwitcherWithConfig } from "./indexer/sources";
 
@@ -68,6 +70,10 @@ export function resolveSacContractIds(): string[] {
 }
 
 // ─── Config ───────────────────────────────────────────────────────────────────
+const POLL_INTERVAL_MS  = parseInt(process.env.POLL_INTERVAL_MS    ?? "6000",  10);
+const BATCH_SIZE        = parseInt(process.env.EVENTS_BATCH_SIZE   ?? "10000", 10);
+const INGEST_WORKERS    = parseInt(process.env.INGEST_WORKERS      ?? "1",     10);
+const SAC_CONTRACT_IDS  = resolveSacContractIds();
 const POLL_INTERVAL_MS = parseInt(process.env.POLL_INTERVAL_MS ?? "6000", 10);
 const BATCH_SIZE = parseInt(process.env.EVENTS_BATCH_SIZE ?? "10000", 10);
 const SAC_CONTRACT_IDS = resolveSacContractIds();
@@ -129,6 +135,10 @@ async function pollOnce(
     return highestLedger;
   }
 
+  // Parse token transfer events
+  const records = parseEvents(events);
+
+  // Persist token transfers
   // Split events by type: NFT (4 topics) vs fungible (3 topics)
   const fungibleEvents = events.filter((e) => !isNftTransferEvent(e));
   const nftRawEvents   = events.filter((e) => isNftTransferEvent(e));
@@ -150,6 +160,14 @@ async function pollOnce(
     records.forEach(emitTransfer);
   }
 
+  // Log every event as a raw host-fn invocation for downstream consumers (#84)
+  const hostFnRecords = events
+    .map(raw => { try { return parseHostFnEvent(raw); } catch { return null; } })
+    .filter((r): r is HostFnRecord => r !== null);
+  if (hostFnRecords.length > 0) {
+    await upsertHostFnLogs(hostFnRecords).catch(err =>
+      console.error("[indexer] host-fn log error:", err),
+    );
   // ── NFT path ─────────────────────────────────────────────────────────────────
   const nftParsed   = parseNftEvents(nftRawEvents);
   const nftRecords  = nftParsed.map((p) => p.record);
@@ -234,7 +252,20 @@ export async function startIndexer(): Promise<void> {
         continue;
       }
 
-      currentLedger = await pollOnce(currentLedger, target);
+      if (INGEST_WORKERS > 1 && SAC_CONTRACT_IDS.length > 1) {
+        // Parallel path: shard contracts across N workers for higher throughput (#83)
+        const { totalInserted, highestLedger } = await pollParallel(
+          SAC_CONTRACT_IDS,
+          currentLedger,
+          target,
+          BATCH_SIZE,
+          INGEST_WORKERS,
+        );
+        totalIndexed += totalInserted;
+        currentLedger = highestLedger;
+      } else {
+        currentLedger = await pollOnce(currentLedger, target);
+      }
 
       // Periodic data retention cleanup
       pollCycleCount++;

src/indexer/host-fn-log.ts

@@ -0,0 +1,170 @@
+/**
+ * InvokeHostFn decoder and log store (#84).
+ *
+ * Every contract event flowing through the indexer is also recorded as a
+ * raw host-function invocation so downstream consumers can interpret events
+ * from arbitrary contracts — not just SAC token transfers.
+ *
+ * Storage layout:
+ *   functionName  — topics[0] decoded as a symbol string
+ *   args          — topics[1..n] serialised to JSON via scValToNative
+ *   result        — value field serialised to JSON via scValToNative
+ *   gasUsed       — nullable; populated externally if transaction metadata
+ *                   is available (requires a separate getTransaction call)
+ */
+
+import * as StellarSdk from "@stellar/stellar-sdk";
+import { Prisma } from "@prisma/client";
+import { prisma } from "../db";
+import type { RawEvent } from "../rpc";
+
+// ─── Types ────────────────────────────────────────────────────────────────────
+
+export interface HostFnRecord {
+  contractId: string;
+  functionName: string;
+  args: unknown;
+  result: unknown;
+  gasUsed: bigint | null;
+  ledger: number;
+  ledgerClosedAt: Date;
+  txHash: string;
+  eventId: string;
+}
+
+// ─── Helpers ──────────────────────────────────────────────────────────────────
+
+/**
+ * Convert any value returned by scValToNative to something JSON.stringify can
+ * handle.  BigInt values (i128/u128/i64/u64) are turned into decimal strings.
+ */
+function toJsonSafe(val: unknown): unknown {
+  if (typeof val === "bigint") return val.toString();
+  if (Array.isArray(val)) return val.map(toJsonSafe);
+  if (val !== null && typeof val === "object") {
+    return Object.fromEntries(
+      Object.entries(val as Record<string, unknown>).map(([k, v]) => [k, toJsonSafe(v)]),
+    );
+  }
+  return val;
+}
+
+function scValToSafeJson(scVal: StellarSdk.xdr.ScVal): unknown {
+  try {
+    return toJsonSafe(StellarSdk.scValToNative(scVal));
+  } catch {
+    // Fall back to base64 XDR so no data is lost on unknown future ScVal types
+    return { xdr: scVal.toXDR("base64") };
+  }
+}
+
+// ─── Parser ───────────────────────────────────────────────────────────────────
+
+/**
+ * Convert any raw contract event into a HostFnRecord.
+ * Returns null if the event cannot be decoded (e.g. empty topic list).
+ */
+export function parseHostFnEvent(raw: RawEvent): HostFnRecord | null {
+  const { topic, value, contractId, ledger, ledgerClosedAt, txHash, id: eventId } = raw;
+
+  if (!topic || topic.length === 0) return null;
+
+  let functionName: string;
+  try {
+    const native = StellarSdk.scValToNative(topic[0]);
+    if (typeof native !== "string") return null;
+    functionName = native;
+  } catch {
+    return null;
+  }
+
+  const args   = topic.slice(1).map(scValToSafeJson);
+  const result = scValToSafeJson(value);
+
+  return {
+    contractId,
+    functionName,
+    args,
+    result,
+    gasUsed: null,
+    ledger,
+    ledgerClosedAt: new Date(ledgerClosedAt),
+    txHash,
+    eventId,
+  };
+}
+
+// ─── DB helpers ───────────────────────────────────────────────────────────────
+
+/**
+ * Idempotently persist a batch of host-fn log records.
+ * Conflicts on `eventId` are silently ignored — safe to replay ledger ranges.
+ */
+export async function upsertHostFnLogs(records: HostFnRecord[]): Promise<number> {
+  if (records.length === 0) return 0;
+
+  const result = await prisma.hostFnLog.createMany({
+    data: records.map(r => ({
+      contractId:    r.contractId,
+      functionName:  r.functionName,
+      args:          r.args as Prisma.InputJsonValue,
+      result:        r.result != null ? (r.result as Prisma.InputJsonValue) : Prisma.JsonNull,
+      gasUsed:       r.gasUsed,
+      ledger:        r.ledger,
+      ledgerClosedAt: r.ledgerClosedAt,
+      txHash:        r.txHash,
+      eventId:       r.eventId,
+    })),
+    skipDuplicates: true,
+  });
+
+  return result.count;
+}
+
+export type HostFnQueryParams = {
+  contractId: string;
+  functionName?: string;
+  limit?: number;
+  offset?: number;
+};
+
+/**
+ * Query host-function invocations for a given contract.
+ * Results are ordered newest-ledger-first.
+ */
+export async function queryHostFnLogs(
+  params: HostFnQueryParams,
+): Promise<{ total: number; logs: HostFnRecord[] }> {
+  const { contractId, functionName, limit = 50, offset = 0 } = params;
+  const cap = Math.min(limit, 200);
+
+  const where = {
+    contractId,
+    ...(functionName ? { functionName } : {}),
+  };
+
+  const [total, rows] = await prisma.$transaction([
+    prisma.hostFnLog.count({ where }),
+    prisma.hostFnLog.findMany({
+      where,
+      orderBy: [{ ledger: "desc" }, { id: "desc" }],
+      take: cap,
+      skip: offset,
+    }),
+  ]);
+
+  return {
+    total,
+    logs: rows.map(r => ({
+      contractId:    r.contractId,
+      functionName:  r.functionName,
+      args:          r.args,
+      result:        r.result,
+      gasUsed:       r.gasUsed,
+      ledger:        r.ledger,
+      ledgerClosedAt: r.ledgerClosedAt,
+      txHash:        r.txHash,
+      eventId:       r.eventId,
+    })),
+  };
+}