fix: release undersized groups for individual processing

- When a settled group has fewer unprocessed events than minGroupSize,
  release those events for individual processing instead of keeping
  them stuck in the aggregator forever
- Added debug logging to OpenSea dedupe layer to show exactly which
  events are being filtered and why
- Updated getReadyGroups to return both groups and released individuals
- processEventsWithAggregator now includes released individuals in
  processableEvents

Author: ryanio

package.json

@@ -1,6 +1,6 @@
 {
   "name": "opensea-activity-bot",
-  "version": "3.5.9",
+  "version": "3.5.10",
   "description": "A bot that shares new OpenSea events for a collection to Discord and Twitter.",
   "author": "Ryan Ghods <ryan@ryanio.com>",
   "license": "MIT",

src/opensea.ts

@@ -471,12 +471,19 @@ const processEventFilters = (
   for (const event of processed) {
     const key = canonicalEventKeyFor(event);
     if (eventStateStore.hasKey(key)) {
+      logger.debug(
+        `[Dedupe] Skipping already-seen event: ${event.event_type} token=${event.nft?.identifier ?? "?"} key=${key}`
+      );
       continue;
     }
     deduped.push(event);
     newKeys.push(key);
   }
 
+  if (preDedup > 0 && deduped.length === 0) {
+    logger.debug(`[Dedupe] All ${preDedup} events were filtered as duplicates`);
+  }
+
   eventStateStore.markProcessed(newKeys);
   processed = deduped;
   stats.deduped = preDedup - deduped.length;

src/utils/event-grouping.ts

@@ -125,8 +125,11 @@ export class EventGroupManager {
     }
   }
 
-  // Get ready event groups
-  getReadyGroups(): Array<{ tx: string; events: OpenSeaAssetEvent[] }> {
+  // Get ready event groups and any individual events released from undersized groups
+  getReadyGroups(): {
+    groups: Array<{ tx: string; events: OpenSeaAssetEvent[] }>;
+    releasedIndividuals: OpenSeaAssetEvent[];
+  } {
     // Log pending groups state before checking
     if (this.actorAgg.size > 0) {
       const now = Date.now();
@@ -163,7 +166,8 @@ export class EventGroupManager {
   private processSettledGroup(
     key: string,
     agg: { events: OpenSeaAssetEvent[]; rawCount: number },
-    out: Array<{ tx: string; events: OpenSeaAssetEvent[] }>
+    out: Array<{ tx: string; events: OpenSeaAssetEvent[] }>,
+    releasedIndividuals: OpenSeaAssetEvent[]
   ): void {
     const unprocessed = this.filterUnprocessedEvents(agg.events, key);
 
@@ -182,11 +186,13 @@ export class EventGroupManager {
     }
 
     if (unprocessed.length > 0) {
-      // Events exist but below minGroupSize - keep for individual processing
-      logger.warn(
-        `[EventGroup] Group ${key} has ${unprocessed.length} unprocessed events ` +
-          `(below minGroupSize=${this.minGroupSize}), keeping for individual processing`
+      // Events exist but below minGroupSize - release for individual processing
+      logger.info(
+        `[EventGroup] 📤 Releasing ${unprocessed.length} event(s) from group ${key} for individual processing ` +
+          `(below minGroupSize=${this.minGroupSize})`
       );
+      releasedIndividuals.push(...unprocessed);
+      this.actorAgg.delete(key);
       return;
     }
 
@@ -197,11 +203,12 @@ export class EventGroupManager {
     this.actorAgg.delete(key);
   }
 
-  private collectReadyActorGroups(): Array<{
-    tx: string;
-    events: OpenSeaAssetEvent[];
-  }> {
-    const out: Array<{ tx: string; events: OpenSeaAssetEvent[] }> = [];
+  private collectReadyActorGroups(): {
+    groups: Array<{ tx: string; events: OpenSeaAssetEvent[] }>;
+    releasedIndividuals: OpenSeaAssetEvent[];
+  } {
+    const groups: Array<{ tx: string; events: OpenSeaAssetEvent[] }> = [];
+    const releasedIndividuals: OpenSeaAssetEvent[] = [];
     const now = Date.now();
     this.pruneStaleActorGroups(now);
 
@@ -211,10 +218,10 @@ export class EventGroupManager {
       const meetsMinSize = agg.rawCount >= this.minGroupSize;
 
       if (meetsMinSize && isSettled) {
-        this.processSettledGroup(key, agg, out);
+        this.processSettledGroup(key, agg, groups, releasedIndividuals);
       }
     }
-    return out;
+    return { groups, releasedIndividuals };
   }
 
   private actorKeyForEvent(event: OpenSeaAssetEvent): string | undefined {
@@ -684,17 +691,23 @@ export const processEventsWithAggregator = (
   }
 
   // Always flush ready groups (even if no new events)
-  const readyGroups = groupManager.getReadyGroups();
+  // Also get any individual events released from undersized groups
+  const { groups: readyGroups, releasedIndividuals } =
+    groupManager.getReadyGroups();
 
   // Filter processable events (only if we have new events)
   const { processableEvents, skippedDupes, skippedPending } =
     events.length > 0
       ? groupManager.filterProcessableEvents(events)
       : { processableEvents: [], skippedDupes: 0, skippedPending: 0 };
 
+  // Add released individuals to processable events
+  // These are events from settled groups that didn't meet minGroupSize
+  const allProcessable = [...processableEvents, ...releasedIndividuals];
+
   return {
     readyGroups,
-    processableEvents,
+    processableEvents: allProcessable,
     skippedDupes,
     skippedPending,
   };

test/utils/event-grouping.test.ts

@@ -397,17 +397,17 @@ describe("eventGrouping-utils", () => {
       groupManager.addEvents([mockEvent1, mockEvent2]);
 
       // Initially no ready groups
-      let readyGroups = groupManager.getReadyGroups();
-      expect(readyGroups).toHaveLength(0);
+      let result = groupManager.getReadyGroups();
+      expect(result.groups).toHaveLength(0);
 
       // Wait for settle time (config.settleMs is 1000, so wait 1100 to be safe)
       const SETTLE_BUFFER_MS = 1100;
       jest.advanceTimersByTime(SETTLE_BUFFER_MS);
 
-      readyGroups = groupManager.getReadyGroups();
-      expect(readyGroups).toHaveLength(1);
-      expect(readyGroups[0].tx).toBe("actor:purchase:0xbuyer1");
-      expect(readyGroups[0].events).toHaveLength(2);
+      result = groupManager.getReadyGroups();
+      expect(result.groups).toHaveLength(1);
+      expect(result.groups[0].tx).toBe("actor:purchase:0xbuyer1");
+      expect(result.groups[0].events).toHaveLength(2);
     });
 
     it("does not flush a group if duplicates reduce below min size", () => {
@@ -419,8 +419,10 @@ describe("eventGrouping-utils", () => {
 
       const SETTLE_BUFFER_MS = 1100;
       jest.advanceTimersByTime(SETTLE_BUFFER_MS);
-      const readyGroups = groupManager.getReadyGroups();
-      expect(readyGroups.length).toBe(0);
+      const result = groupManager.getReadyGroups();
+      // No groups returned (below minGroupSize), but the single event should be released
+      expect(result.groups.length).toBe(0);
+      expect(result.releasedIndividuals.length).toBe(1);
     });
   });