fix(workers): make delivery recording recoverable when contract call fails (#507)

src/escrow/escrow.repository.ts

@@ -273,17 +273,52 @@ export class EscrowRepository {
   }
 
   /**
-   * Returns all SHIPPED escrows that have a non-null trackingId,
+   * Returns all SHIPPED escrows that have a non-null trackingId and have not
+   * been claimed for delivery recording (deliveryRecordedAt is null),
    * used by the tracking poll worker to check for delivery updates.
    */
   findShippedWithTracking(): Promise<EscrowRecord[]> {
     return this.prisma.escrow
-      .findMany({ where: { state: 'SHIPPED' } })
+      .findMany({
+        where: { state: 'SHIPPED', deliveryRecordedAt: null },
+      })
       .then((escrows) =>
         escrows.filter((escrow) => Boolean(escrow.trackingId)),
       );
   }
 
+  /**
+   * Atomically claims an escrow for delivery recording by setting
+   * deliveryRecordedAt. Returns null if the escrow is already claimed
+   * (deliveryRecordedAt is not null).
+   * Follows the same claim-and-release pattern as markAutoReleaseSubmitting.
+   */
+  async claimDelivery(id: string): Promise<EscrowRecord | null> {
+    const escrow = await this.prisma.escrow.findUnique({ where: { id } });
+    if (!escrow || escrow.state !== 'SHIPPED' || escrow.deliveryRecordedAt !== null) {
+      return null;
+    }
+    const result = await this.prisma.escrow.update({
+      where: { id },
+      data: { deliveryRecordedAt: new Date() },
+    });
+    await this.invalidate(id);
+    return result;
+  }
+
+  /**
+   * Releases the delivery claim by setting deliveryRecordedAt back to null,
+   * allowing the next poll cycle to retry.
+   */
+  async clearDeliveryClaim(id: string): Promise<EscrowRecord> {
+    const result = await this.prisma.escrow.update({
+      where: { id },
+      data: { deliveryRecordedAt: null },
+    });
+    await this.invalidate(id);
+    return result;
+  }
+
   /**
    * Returns SHIPPED escrows whose deliveredAt is at or before the given
    * referenceTime and have no open dispute or existing auto-release transaction.

src/prisma/prisma.service.ts

@@ -505,6 +505,7 @@
       > & {
         shippedAt?: { lte: Date };
         deliveredAt?: { lte: Date } | null;
+        deliveryRecordedAt?: Date | null;
         createdAt?: { gte: Date; lte: Date };
       };
       select?: Partial<Record<keyof EscrowRecord, boolean>>;
@@ -546,6 +547,13 @@
             return escrow.createdAt >= gte && escrow.createdAt <= lte;
           }
 
+          if (
+            key === 'deliveryRecordedAt' &&
+            (value === null || value === undefined)
+          ) {
+            return escrow.deliveryRecordedAt === value;
+          }
+
           return escrow[key as keyof EscrowRecord] === value;
         });
       });
@@ -671,10 +679,105 @@
           | 'autoReleaseTxHash'
           | 'autoReleaseSubmittedAt'
         >
-      >;
+      > & {
+        deliveryRecordedAt?: Date | null;
+      };
     } = {}): Promise<number> => {
       return this.escrow.findMany({ where }).then((records) => records.length);
     },
+    aggregate: ({
+      _sum,
+      _avg,
+      _count,
+    }: {
+      _sum?: { amount?: boolean };
+      _avg?: { amount?: boolean };
+      _count?: { vendorAddress?: boolean; buyerAddress?: boolean };
+    } = {}): Promise<{
+      _sum?: { amount: number | null };
+      _avg?: { amount: number | null };
+      _count?: { vendorAddress: number; buyerAddress: number };
+    }> => {
+      // Mirror the CANCELLED filter that escrow.findMany applies by default
+      const allEscrows = [...this.escrows.values()].filter(
+        (e) => e.state !== 'CANCELLED',
+      );
+
+      const result: Record<string, unknown> = {};
+
+      if (_sum?.amount) {
+        const sum = allEscrows.reduce(
+          (s, e) => s + Number(e.amount),
+          0,
+        );
+        result._sum = { amount: sum };
+      }
+
+      if (_avg?.amount) {
+        const sum = allEscrows.reduce(
+          (s, e) => s + Number(e.amount),
+          0,
+        );
+        const avg = allEscrows.length > 0 ? sum / allEscrows.length : 0;
+        result._avg = { amount: avg };
+      }
+
+      if (_count?.vendorAddress) {
+        const unique = new Set(allEscrows.map((e) => e.vendorAddress)).size;
+        result._count = { ...(result._count as Record<string, number> ?? {}), vendorAddress: unique };
+      }
+
+      if (_count?.buyerAddress) {
+        const unique = new Set(allEscrows.map((e) => e.buyerAddress)).size;
+        result._count = { ...(result._count as Record<string, number> ?? {}), buyerAddress: unique };
+      }
+
+      return Promise.resolve(result as {
+        _sum?: { amount: number | null };
+        _avg?: { amount: number | null };
+        _count?: { vendorAddress: number; buyerAddress: number };
+      });
+    },
+    groupBy: ({
+      by,
+      _count,
+    }: {
+      by: string[];
+      _count?: boolean;
+    }): Promise<
+      Array<
+        Record<string, unknown> & { _count?: number }
+      >
+    > => {
+      // Mirror the CANCELLED filter that escrow.findMany applies by default
+      const allEscrows = [...this.escrows.values()].filter(
+        (e) => e.state !== 'CANCELLED',
+      );
+      const groups = new Map<string, EscrowRecord[]>();
+
+      for (const escrow of allEscrows) {
+        const groupKey = by.map((field) => escrow[field as keyof EscrowRecord]).join('|');
+        if (!groups.has(groupKey)) {
+          groups.set(groupKey, []);
+        }
+        groups.get(groupKey)!.push(escrow);
+      }
+
+      const result: Array<Record<string, unknown> & { _count?: number }> = [];
+      for (const [groupKey, records] of groups.entries()) {
+        const group: Record<string, unknown> = {};
+        const keys = groupKey.split('|');
+        for (let i = 0; i < by.length; i++) {
+          group[by[i]] = (records[0] as Record<string, unknown>)[by[i]];
+        }
+        if (_count) {
+          group._count = records.length;
+        }
+        result.push(group);
+      }
+
+      return Promise.resolve(result);
+    },
     deleteMany: (): Promise<{ count: number }> => {
       const count = this.escrows.size;
       this.escrows.clear();
@@ -727,10 +830,18 @@
     },
     findMany: ({
       where,
+      orderBy,
+      skip,
+      take,
     }: {
-      where?: Partial<Pick<DisputeRecord, 'escrowId' | 'status'>>;
+      where?: Partial<Pick<DisputeRecord, 'escrowId' | 'status'>> & {
+        status?: DisputeState | { in?: DisputeState[] };
+      };
+      orderBy?: Partial<Record<keyof DisputeRecord, 'asc' | 'desc'>>;
+      skip?: number;
+      take?: number;
     } = {}): Promise<DisputeRecord[]> => {
-      const disputes = [...this.disputes.values()].filter((dispute) => {
+      let disputes = [...this.disputes.values()].filter((dispute) => {
         if (!where) {
           return true;
         }
@@ -740,10 +851,54 @@
             return true;
           }
 
+          // Support { in: [...] } for status filtering
+          if (
+            key === 'status' &&
+            typeof value === 'object' &&
+            value !== null &&
+            'in' in value
+          ) {
+            return (value as { in: DisputeState[] }).in.includes(
+              dispute.status,
+            );
+          }
+
           return dispute[key as keyof DisputeRecord] === value;
         });
       });
 
+      if (orderBy) {
+        const [field, dir] = Object.entries(orderBy)[0] as [
+          keyof DisputeRecord,
+          'asc' | 'desc',
+        ];
+        disputes = [...disputes].sort((a, b) => {
+          const aVal = a[field];
+          const bVal = b[field];
+          if (aVal instanceof Date && bVal instanceof Date) {
+            return dir === 'asc'
+              ? aVal.getTime() - bVal.getTime()
+              : bVal.getTime() - aVal.getTime();
+          }
+          if (typeof aVal === 'number' && typeof bVal === 'number') {
+            return dir === 'asc' ? aVal - bVal : bVal - aVal;
+          }
+          if (typeof aVal === 'string' && typeof bVal === 'string') {
+            return dir === 'asc'
+              ? aVal.localeCompare(bVal)
+              : bVal.localeCompare(aVal);
+          }
+          return 0;
+        });
+      }
+
+      if (skip !== undefined) {
+        disputes = disputes.slice(skip);
+      }
+      if (take !== undefined) {
+        disputes = disputes.slice(0, take);
+      }
+
       return Promise.resolve(disputes.map((dispute) => ({ ...dispute })));
     },
     update: ({
@@ -771,6 +926,17 @@
         .findMany({ where })
         .then((records) => records[0] ?? null);
     },
+    count: ({
+      where,
+    }: {
+      where?: Partial<Pick<DisputeRecord, 'escrowId' | 'status'>> & {
+        status?: DisputeState | { in?: DisputeState[] };
+      };
+    } = {}): Promise<number> => {
+      return this.dispute
+        .findMany({ where })
+        .then((records) => records.length);
+    },
     deleteMany: (): Promise<{ count: number }> => {
       const count = this.disputes.size;
       this.disputes.clear();

src/workers/tracking-poll.worker.ts

@@ -55,9 +55,29 @@ export class TrackingPollWorker implements OnModuleInit, OnApplicationShutdown {
             continue;
           }
 
-          const deliveredAt = new Date();
-          await this.escrowRepository.markDelivered(escrow.id, deliveredAt);
-          await this.contractService.recordDelivery(escrow.id);
+          // Claim the escrow before any network call. This follows the same
+          // claim-and-release pattern as AutoReleaseWorker (#507): if the
+          // contract call fails, the claim is cleared so the next poll cycle
+          // retries. Without this, a failed recordDelivery leaves the escrow
+          // in DELIVERED state permanently out of sync with the chain.
+          const claimed = await this.escrowRepository.claimDelivery(
+            escrow.id,
+          );
+          if (!claimed) {
+            continue;
+          }
+
+          try {
+            await this.contractService.recordDelivery(escrow.id);
+            await this.escrowRepository.markDelivered(
+              escrow.id,
+              new Date(),
+            );
+          } catch (error) {
+            // Release the claim so the next poll cycle can retry.
+            await this.escrowRepository.clearDeliveryClaim(escrow.id);
+            throw error;
+          }
         } catch (error) {
           this.logger.error(
             JSON.stringify({