Fix: 4 issues (#99, #100, #101, #102)

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),
   }),
 );
 

src/routes/publishers.test.ts

@@ -31,6 +31,10 @@ const authedPublisher = {
 let publishersLookupRows: Array<typeof authedPublisher>;
 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<string, { totalSales: number; totalEarned: number }>();
+    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<string, any>(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" });
+  });
 });

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<number>`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;