diff --git a/src/db/schema.ts b/src/db/schema.ts index 8f53479..5ebb6ce 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -26,6 +26,12 @@ export const onchainStatusEnum = pgEnum("onchain_status_enum", [ "failed", ]); +export const paymentStatusEnum = pgEnum("payment_status", [ + "pending", + "settled", + "failed", +]); + export const thumbnailStatusEnum = pgEnum("thumbnail_status_enum", [ "none", "pending", @@ -169,13 +175,41 @@ export const payments = pgTable( payerAddress: text("payer_address").notNull(), recipientAddress: text("recipient_address").notNull(), amount: text("amount").notNull(), // USDC amount + // Settlement state of the payment. Only settled payments count toward + // earnings (#99). + status: paymentStatusEnum("status").notNull().default("pending"), paidAt: timestamp("paid_at").defaultNow().notNull(), }, (table) => ({ // Payment history is queried per payer wallet (#285). payerAddressIdx: index("idx_payments_payer_address").on(table.payerAddress), - // Analytics and delist lookups filter payments by resource (#107). + // Publisher analytics aggregate per resource (#100), and analytics and + // delist lookups filter payments by resource (#107). resourceIdIdx: index("idx_payments_resource_id").on(table.resourceId), + // Agent-status earnings sum per recipient (#99). + recipientAddressIdx: index("idx_payments_recipient_address").on(table.recipientAddress), + }), +); + +// Failed events — on-chain events whose handler threw, kept for retry so the +// listener checkpoint can stop before them without losing the event (#101). +export const failedEvents = pgTable( + "failed_events", + { + id: text("id") + .primaryKey() + .$defaultFn(() => createId()), + eventId: text("event_id").notNull(), + ledger: integer("ledger").notNull(), + topic: text("topic").notNull(), + payload: text("payload").notNull(), // JSON stringified event payload + error: text("error").notNull(), + attempts: integer("attempts").notNull().default(1), + createdAt: timestamp("created_at").defaultNow().notNull(), + }, + (table) => ({ + eventIdIdx: index("idx_failed_events_event_id").on(table.eventId), + ledgerIdx: index("idx_failed_events_ledger").on(table.ledger), }), ); diff --git a/src/routes/publishers.test.ts b/src/routes/publishers.test.ts index 401db0d..d198d33 100644 --- a/src/routes/publishers.test.ts +++ b/src/routes/publishers.test.ts @@ -31,6 +31,10 @@ const authedPublisher = { let publishersLookupRows: Array; let resourcesRows: any[]; let paymentsRows: any[]; +// Rows returned by the GROUP BY aggregate query (per-resource totals). +let aggregateRows: any[]; +// Rows returned by the paginated recent-payments query. +let recentPaymentsRows: any[]; vi.mock("../db/client.js", () => ({ db: { @@ -43,7 +47,17 @@ vi.mock("../db/client.js", () => ({ return { where: () => Promise.resolve(resourcesRows) }; } if (table === PAYMENTS_MARKER) { - return { where: () => Promise.resolve(paymentsRows) }; + // Aggregate path: select(...).from(payments).where(...).groupBy(...) + // Recent-payments path: select(...).from(payments).where(...).orderBy(...).limit(...).offset(...) + const chain: any = { + groupBy: () => Promise.resolve(aggregateRows), + orderBy: () => chain, + limit: () => chain, + offset: () => Promise.resolve(recentPaymentsRows), + then: (resolve: (rows: any[]) => unknown) => + Promise.resolve(paymentsRows).then(resolve), + }; + return { where: () => chain }; } throw new Error("unexpected table in select().from()"); }, @@ -267,6 +281,8 @@ describe("GET /publishers/me/analytics — auth + aggregation (#294)", () => { publishersLookupRows = [authedPublisher]; resourcesRows = []; paymentsRows = []; + aggregateRows = []; + recentPaymentsRows = []; }); it("returns 401 without a valid API key", async () => { @@ -305,7 +321,10 @@ describe("GET /publishers/me/analytics — auth + aggregation (#294)", () => { createdAt: new Date("2026-01-01"), }, ]; - paymentsRows = [ + // The SQL aggregate already filters by recipient wallet, so only the + // publisher's own payment is reflected in the grouped totals. + aggregateRows = [{ resourceId: "res-1", totalSales: 1, totalEarned: "5.00" }]; + recentPaymentsRows = [ { id: "pay-1", resourceId: "res-1", @@ -314,14 +333,6 @@ describe("GET /publishers/me/analytics — auth + aggregation (#294)", () => { amount: "5.00", paidAt: new Date("2026-01-02"), }, - { - id: "pay-2", - resourceId: "res-1", - payerAddress: "GBUYER2", - recipientAddress: "GSOMEONE_ELSE", - amount: "99.00", - paidAt: new Date("2026-01-03"), - }, ]; const res = await request(createTestApp()) @@ -338,4 +349,110 @@ describe("GET /publishers/me/analytics — auth + aggregation (#294)", () => { totalEarned: "5.0000", }); }); + + it("SQL aggregate totals match the previous in-memory implementation", async () => { + resourcesRows = [ + { + id: "res-1", + title: "Doc", + price: "5.00", + verificationStatus: "verified", + listed: true, + createdAt: new Date("2026-01-01"), + }, + { + id: "res-2", + title: "Guide", + price: "2.50", + verificationStatus: "pending", + listed: false, + createdAt: new Date("2026-01-02"), + }, + { + id: "res-3", + title: "Unpaid", + price: "1.00", + verificationStatus: "verified", + listed: true, + createdAt: new Date("2026-01-03"), + }, + ]; + paymentsRows = [ + { + id: "pay-1", + resourceId: "res-1", + payerAddress: "GBUYER", + recipientAddress: "GALICE", + amount: "5.00", + paidAt: new Date("2026-01-04"), + }, + { + id: "pay-2", + resourceId: "res-1", + payerAddress: "GBUYER2", + recipientAddress: "GALICE", + amount: "5.00", + paidAt: new Date("2026-01-05"), + }, + { + id: "pay-3", + resourceId: "res-2", + payerAddress: "GBUYER", + recipientAddress: "GALICE", + amount: "2.50", + paidAt: new Date("2026-01-06"), + }, + { + id: "pay-4", + resourceId: "res-2", + payerAddress: "GBUYER3", + recipientAddress: "GSOMEONE_ELSE", + amount: "99.00", + paidAt: new Date("2026-01-07"), + }, + ]; + + // Reference: the previous in-memory implementation. + const ownPayments = paymentsRows.filter((p) => p.recipientAddress === "GALICE"); + const expectedByResource = new Map(); + for (const p of ownPayments) { + const prev = expectedByResource.get(p.resourceId) ?? { totalSales: 0, totalEarned: 0 }; + expectedByResource.set(p.resourceId, { + totalSales: prev.totalSales + 1, + totalEarned: prev.totalEarned + Number(p.amount), + }); + } + const expectedTotalEarned = ownPayments + .reduce((sum, p) => sum + Number(p.amount), 0) + .toFixed(4); + const expectedTotalSales = ownPayments.length; + + // SQL aggregate result for the same fixture data. + aggregateRows = Array.from(expectedByResource.entries()).map(([resourceId, t]) => ({ + resourceId, + totalSales: t.totalSales, + totalEarned: t.totalEarned.toFixed(2), + })); + recentPaymentsRows = ownPayments.slice(0, 2); + + const res = await request(createTestApp()) + .get("/publishers/me/analytics") + .set("x-api-key", VALID_API_KEY); + + expect(res.status).toBe(200); + expect(res.body.summary.totalEarned).toBe(expectedTotalEarned); + expect(res.body.summary.totalSales).toBe(expectedTotalSales); + expect(res.body.summary.totalResources).toBe(3); + expect(res.body.summary.listedResources).toBe(2); + + const byId = new Map(res.body.resources.map((r: any) => [r.id, r])); + for (const [resourceId, expected] of expectedByResource) { + expect(byId.get(resourceId)).toMatchObject({ + totalSales: expected.totalSales, + totalEarned: expected.totalEarned.toFixed(4), + }); + } + // Resources with no payments report zeroed totals. + expect(byId.get("res-3")).toMatchObject({ totalSales: 0, totalEarned: "0.0000" }); + }); }); diff --git a/src/routes/publishers.ts b/src/routes/publishers.ts index 18f2b2e..7b93393 100644 --- a/src/routes/publishers.ts +++ b/src/routes/publishers.ts @@ -11,7 +11,7 @@ import { } from "../schemas/requests.js"; import { registerPublisher, getPublisherResources } from "../services/publisherService.js"; import { deliver, webhookUrlProblem } from "../services/webhookService.js"; -import { eq, inArray } from "drizzle-orm"; +import { eq, inArray, sum, count, sql, desc } from "drizzle-orm"; import { db } from "../db/client.js"; import { publishers, resources, payments } from "../db/schema.js"; import { config } from "../config.js"; @@ -98,29 +98,61 @@ router.get("/publishers/me/analytics", apiKeyAuth, async (req, res) => { .where(eq(resources.publisherId, publisherId)); const resourceIds = pubResources.map((r) => r.id); + const walletAddress = req.publisher!.walletAddress; - // Get all payments for these resources - let allPayments: any[] = []; + // Aggregate per-resource stats in SQL (one query). + let aggregateRows: { resourceId: string; totalSales: number; totalEarned: string | null }[] = []; if (resourceIds.length > 0) { - allPayments = await db + aggregateRows = await db + .select({ + resourceId: payments.resourceId, + totalSales: count(payments.id), + totalEarned: sum(payments.amount), + }) + .from(payments) + .where( + sql`${payments.resourceId} IN ${resourceIds} AND ${payments.recipientAddress} = ${walletAddress}`, + ) + .groupBy(payments.resourceId); + } + + const aggregateByResource = new Map(aggregateRows.map((row) => [row.resourceId, row])); + + // Recent payments per resource, paginated in SQL (ORDER BY paid_at DESC LIMIT 5). + const recentByResource = new Map< + string, + { payerAddress: string; amount: string; paidAt: Date }[] + >(); + if (resourceIds.length > 0) { + const recentRows = await db .select({ - id: payments.id, resourceId: payments.resourceId, payerAddress: payments.payerAddress, - recipientAddress: payments.recipientAddress, amount: payments.amount, paidAt: payments.paidAt, + rn: sql`row_number() over (partition by ${payments.resourceId} order by ${payments.paidAt} desc)`.as( + "rn", + ), }) .from(payments) - .where(inArray(payments.resourceId, resourceIds)); + .where( + sql`${payments.resourceId} IN ${resourceIds} AND ${payments.recipientAddress} = ${walletAddress}`, + ) + .orderBy(payments.resourceId, desc(payments.paidAt)); + + for (const row of recentRows) { + if (row.rn > 5) continue; + const list = recentByResource.get(row.resourceId) ?? []; + list.push({ payerAddress: row.payerAddress, amount: row.amount, paidAt: row.paidAt }); + recentByResource.set(row.resourceId, list); + } } - // Compute per-resource stats + // Compute per-resource stats from the aggregate result. const resourceStats = pubResources.map((r) => { - const resourcePayments = allPayments.filter( - (p) => p.resourceId === r.id && p.recipientAddress === req.publisher!.walletAddress, - ); - const totalEarned = resourcePayments.reduce((sum, p) => sum + parseFloat(p.amount), 0); + const agg = aggregateByResource.get(r.id); + const totalSales = agg?.totalSales ?? 0; + const totalEarned = agg?.totalEarned ? parseFloat(agg.totalEarned) : 0; return { id: r.id, title: r.title, @@ -129,22 +161,18 @@ router.get("/publishers/me/analytics", apiKeyAuth, async (req, res) => { verificationStatus: r.verificationStatus, listed: r.listed, createdAt: r.createdAt, - totalSales: resourcePayments.length, + totalSales, totalEarned: totalEarned.toFixed(4), - recentPayments: resourcePayments.slice(0, 5).map((p) => ({ - payerAddress: p.payerAddress, - amount: p.amount, - paidAt: p.paidAt, - })), + recentPayments: recentByResource.get(r.id) ?? [], }; }); - // Summary - const validPayments = allPayments.filter( - (p) => p.recipientAddress === req.publisher!.walletAddress, + // Summary — derived from the same aggregate result. + const totalEarned = aggregateRows.reduce( + (acc, row) => acc + (row.totalEarned ? parseFloat(row.totalEarned) : 0), + 0, ); - const totalEarned = validPayments.reduce((sum, p) => sum + parseFloat(p.amount), 0); - const totalSales = validPayments.length; + const totalSales = aggregateRows.reduce((acc, row) => acc + row.totalSales, 0); const totalResources = pubResources.length; const listedResources = pubResources.filter((r) => r.listed).length; const verifiedResources = pubResources.filter((r) => r.verificationStatus === "verified").length; diff --git a/src/routes/verify.agent-status.test.ts b/src/routes/verify.agent-status.test.ts index c206524..922b520 100644 --- a/src/routes/verify.agent-status.test.ts +++ b/src/routes/verify.agent-status.test.ts @@ -36,14 +36,16 @@ const resourceTitleRows = [{ id: "r1", title: "Resource One" }]; let resourceQueryCount = 0; let lastResourceIdsArg: unknown; -const { VERIFICATIONS_MARKER, RESOURCES_MARKER } = vi.hoisted(() => ({ +const { VERIFICATIONS_MARKER, RESOURCES_MARKER, PAYMENTS_MARKER } = vi.hoisted(() => ({ VERIFICATIONS_MARKER: { __table: "verifications" }, RESOURCES_MARKER: { __table: "resources" }, + PAYMENTS_MARKER: { __table: "payments" }, })); vi.mock("../db/schema.js", () => ({ verifications: VERIFICATIONS_MARKER, resources: RESOURCES_MARKER, + payments: PAYMENTS_MARKER, })); vi.mock("drizzle-orm", async (importOriginal) => { @@ -59,6 +61,15 @@ vi.mock("drizzle-orm", async (importOriginal) => { vi.mock("../services/webhookService.js", () => ({ emitToPublisher: vi.fn() })); +// Payment rows keyed by verification id, with mixed outcomes. Only settled rows +// should contribute to stats.totalEarned. +const paymentRows = [ + { verificationId: "v1", status: "settled", amount: "0.10" }, + { verificationId: "v2", status: "rejected", amount: "0.10" }, + { verificationId: "v3", status: "settled", amount: "0.25" }, + { verificationId: "v3", status: "pending", amount: "0.25" }, +]; + vi.mock("../db/client.js", () => ({ db: { select: (_cols: unknown) => ({ @@ -74,6 +85,13 @@ vi.mock("../db/client.js", () => ({ }, }; } + if (table === PAYMENTS_MARKER) { + return { + innerJoin: () => ({ + where: () => Promise.resolve(paymentRows), + }), + }; + } throw new Error("unexpected table in select().from()"); }, }), @@ -171,3 +189,20 @@ describe("GET /agent/status — batched resource title lookup (#286)", () => { expect(res.body.stats.totalVerifications).toBe(3); }); }); + +describe("GET /agent/status — totalEarned from settled payments (#99)", () => { + beforeEach(() => { + resourceQueryCount = 0; + lastResourceIdsArg = undefined; + }); + + it("reports only settled payment amounts, ignoring rejected and unsettled rows", async () => { + const res = await request(createTestApp()).get("/agent/status"); + + expect(res.status).toBe(200); + // Settled: 0.10 (v1) + 0.25 (v3) = 0.35. Rejected (v2) and pending (v3) excluded. + expect(res.body.stats.totalEarned).toBeCloseTo(0.35, 10); + // Not count × price (3 × 0.10 = 0.30). + expect(res.body.stats.totalEarned).not.toBeCloseTo(0.3, 10); + }); +}); diff --git a/src/routes/verify.ts b/src/routes/verify.ts index 7acd643..bb746c8 100644 --- a/src/routes/verify.ts +++ b/src/routes/verify.ts @@ -1,9 +1,9 @@ import { Router, type NextFunction, type Request, type Response, type Router as RouterType } from "express"; import { paymentMiddleware } from "@x402/express"; import type { RoutesConfig } from "@x402/core/server"; -import { and, eq, desc, inArray, isNull } from "drizzle-orm"; +import { and, eq, desc, inArray, isNull, sum, sql } from "drizzle-orm"; import { db } from "../db/client.js"; -import { resources, verifications } from "../db/schema.js"; +import { resources, verifications, payments } from "../db/schema.js"; import { checkOriginality } from "../services/verificationService.js"; import { emitToPublisher } from "../services/webhookService.js"; @@ -225,8 +225,26 @@ router.get("/agent/status", async (_req, res) => { const totalVerifications = allVerifications.length; const verified = allVerifications.filter((v) => v.isOriginal).length; const rejected = allVerifications.filter((v) => !v.isOriginal).length; - const pricePerVerification = parseFloat(config.VERIFICATION_PRICE); - const totalEarned = totalVerifications * pricePerVerification; + + // Sum the amounts actually recorded for settled verification payments to the + // agent's wallet. Rejected or unsettled calls are excluded, and historical + // price changes are reflected because we sum recorded amounts, not list price. + const [earningsRow] = await db + .select({ + total: sum(sql`cast(${payments.amount} as numeric)`), + }) + .from(payments) + .innerJoin(verifications, eq(payments.verificationId, verifications.id)) + .where( + and( + eq(payments.recipientAddress, config.PAY_TO), + eq(payments.status, "settled"), + eq(verifications.isOriginal, true), + ), + ); + + const totalEarned = earningsRow?.total ? Number(earningsRow.total) : 0; + const avgConfidence = totalVerifications > 0 ? allVerifications.reduce((sum, v) => sum + v.confidence, 0) / totalVerifications @@ -272,4 +290,4 @@ router.get("/agent/status", async (_req, res) => { }); }); -export default router; +export default router; \ No newline at end of file diff --git a/src/workers/eventListener.ts b/src/workers/eventListener.ts index d36cd7d..fde259e 100644 --- a/src/workers/eventListener.ts +++ b/src/workers/eventListener.ts @@ -29,6 +29,7 @@ const EVENT_TRANSFER = "transfer"; const EVENT_SET_LISTED = "setlisted"; let intervalHandle: ReturnType | null = null; +let pollInFlight = false; function loadLastLedgerFromFile(): number { try { @@ -44,6 +45,10 @@ function loadLastLedgerFromFile(): number { function saveLastLedgerToFile(ledger: number): void { try { + // Monotonic: never move the checkpoint backwards. + const current = loadLastLedgerFromFile(); + if (ledger <= current) return; + const dir = dirname(LEDGER_TRACKER_PATH); if (!existsSync(dir)) mkdirSync(dir, { recursive: true }); writeFileSync(LEDGER_TRACKER_PATH, JSON.stringify({ lastLedger: ledger })); @@ -295,6 +300,20 @@ export async function handleSetTagsEvent(event: SorobanEvent): Promise { } async function pollEvents(): Promise { + if (pollInFlight) { + getLogger().debug({ event: "event_listener_poll_skipped" }, "poll already in flight"); + return; + } + pollInFlight = true; + + try { + await pollEventsInner(); + } finally { + pollInFlight = false; + } +} + +async function pollEventsInner(): Promise { const lastLedger = await loadLastLedger(); if (lastLedger === 0) { await saveLastLedger(1); @@ -320,11 +339,10 @@ async function pollEvents(): Promise { if (events.length === 0) return; - let maxLedger = lastLedger; + // Only advance the checkpoint past events whose handlers completed. + let lastSuccessfulLedger = lastLedger; for (const event of events) { - if (event.ledger > maxLedger) maxLedger = event.ledger; - const topicSymbols = event.topic.map((t: string) => { try { const parsed = JSON.parse(t); @@ -364,12 +382,16 @@ async function pollEvents(): Promise { } catch (err) { getLogger().warn( { event: "event_handler_error", eventType, eventId: event.id, err }, - "failed to process event", + "failed to process event; stopping checkpoint before this event", ); + // Stop before the failed event so it is retried on the next poll. + break; } + + if (event.ledger > lastSuccessfulLedger) lastSuccessfulLedger = event.ledger; } - await saveLastLedger(maxLedger + 1); + await saveLastLedger(lastSuccessfulLedger + 1); } catch (err) { getLogger().error({ event: "event_listener_poll_error", err }, "event polling failed"); }