Bound collection candidate D1 work

apps/api/src/queries/candidates.ts

@@ -6,8 +6,9 @@
  * asset to nine thousand wallets manufactures holders, not collectors (ground truth: XCERPASS reads
  * 81% "collectors" by raw holding and 0% by choice; FoldingCoin reads 98% by choice).
  *
- * Two bounded stages so no single D1 query approaches the CPU limit: seeds (raw collector-held
- * ranking) then chosen counts over the seeds in small batches, composed by the route.
+ * Two bounded stages keep the work below D1's CPU limit: seeds (raw collector-held ranking), then
+ * chosen counts over only those seeds. The second stage uses one JSON bind, avoiding both repeated
+ * collector scans and D1's SQL-variable ceiling.
  */
 import { PERSONA } from "#api/reputation/config";
 import { q } from "#api/db";
@@ -16,12 +17,9 @@ const COLLECTION_SOURCES = "'manual','issuer','discovered','collection','digirar
 
 // The H2 global persona, collector-primary — the same thresholds the address header and the
 // collection holder-makeup use (reputation/config.ts), so "collector" means one thing everywhere.
-const COLLECTOR_SET = `SELECT s.address_id
-  FROM address_signals s
-  JOIN address_reputations r ON r.address_id=s.address_id
-  WHERE s.is_exchange=0 AND s.is_deposit=0 AND s.is_emblem_vault=0 AND s.is_burn=0
+const COLLECTOR_PREDICATE = `s.is_exchange=0 AND s.is_deposit=0 AND s.is_emblem_vault=0 AND s.is_burn=0
     AND s.vault_scams+s.shell_scams+s.dump_scams=0 AND r.reputation IS NOT NULL
-    AND s.assets_held >= ${PERSONA.collectorFloor}
+    AND s.assets_held>0 AND s.assets_held >= ${PERSONA.collectorFloor}
     AND MIN(ln(1+s.assets_held+0.5*s.assets_received)/ln(1+${PERSONA.collectorCap}), 1.0) > MAX(
       CASE WHEN s.assets_issued+s.stamps_created+2*s.src20_deploys >= ${PERSONA.creatorFloor}
         THEN MIN(ln(1+s.assets_issued+s.stamps_created+2*s.src20_deploys)/ln(1+${PERSONA.creatorCap}), 1.0) ELSE 0 END,
@@ -30,6 +28,11 @@ const COLLECTOR_SET = `SELECT s.address_id
       CASE WHEN s.dex_trades >= ${PERSONA.traderFloor}
         THEN MIN(ln(1+s.dex_trades)/ln(1+${PERSONA.traderCap}), 1.0) ELSE 0 END)`;
 
+const COLLECTOR_SET = `SELECT s.address_id
+  FROM address_signals s INDEXED BY idx_address_signals_assets_held
+  JOIN address_reputations r ON r.address_id=s.address_id
+  WHERE ${COLLECTOR_PREDICATE}`;
+
 export interface CandidateSeedRow {
   asset_id: number;
   asset: string;
@@ -45,27 +48,37 @@ export interface CandidateSeedRow {
 export function collectorCandidateSeeds(db: D1Database, limit = 150): Promise<CandidateSeedRow[]> {
   return q<CandidateSeedRow>(
     db,
-    `WITH collectors AS MATERIALIZED (${COLLECTOR_SET})
-     SELECT b.asset_id, d.asset, a.asset_longname,
+    `WITH collectors AS MATERIALIZED (${COLLECTOR_SET}),
+     holder_counts AS MATERIALIZED (
+       SELECT b.asset_id,COUNT(*) collector_holders
+       FROM collectors c
+       CROSS JOIN balances b INDEXED BY idx_balances_address_asset
+       WHERE b.address_id=c.address_id AND CAST(b.quantity AS INTEGER)>0
+       GROUP BY b.asset_id HAVING COUNT(*)>=5
+     ), tagged_assets AS MATERIALIZED (
+       SELECT DISTINCT dictionary.asset_id
+       FROM collection_membership_evidence evidence
+       JOIN entity_dictionary entity ON entity.entity_id=evidence.entity_id AND entity.entity_type='asset'
+       JOIN asset_dictionary dictionary ON dictionary.asset=entity.entity_key
+       WHERE evidence.source IN (${COLLECTION_SOURCES})
+     ), curated_lowq AS MATERIALIZED (
+       SELECT dictionary.asset_id FROM curated
+       JOIN asset_dictionary dictionary ON dictionary.asset=curated.key
+       WHERE curated.kind='lowq'
+     )
+     SELECT counts.asset_id, d.asset, a.asset_longname,
             issuer_address.address issuer,
-            COUNT(DISTINCT b.address_id) collector_holders, sig.holders
-     FROM balances b
-     JOIN collectors c ON c.address_id=b.address_id
-     JOIN asset_dictionary d ON d.asset_id=b.asset_id
-     LEFT JOIN assets a ON a.asset_id=b.asset_id
+            counts.collector_holders, sig.holders
+     FROM holder_counts counts
+     JOIN asset_dictionary d ON d.asset_id=counts.asset_id
+     LEFT JOIN assets a ON a.asset_id=counts.asset_id
      LEFT JOIN address_dictionary issuer_address ON issuer_address.address_id=a.issuer_id
-     LEFT JOIN asset_signals sig ON sig.asset_id=b.asset_id
-     WHERE CAST(b.quantity AS INTEGER) > 0 AND d.asset NOT IN ('XCP','BTC')
+     LEFT JOIN asset_signals sig ON sig.asset_id=counts.asset_id
+     WHERE d.asset NOT IN ('XCP','BTC')
        AND COALESCE(sig.low_quality,0)=0
-       AND d.asset NOT IN (SELECT key FROM curated WHERE kind='lowq')
-       AND NOT EXISTS (
-         SELECT 1 FROM collection_membership_evidence evidence
-         JOIN entity_dictionary entity ON entity.entity_id=evidence.entity_id
-         WHERE entity.entity_type='asset' AND entity.entity_key=d.asset
-           AND evidence.source IN (${COLLECTION_SOURCES})
-       )
-     GROUP BY b.asset_id HAVING collector_holders >= 5
-     ORDER BY collector_holders DESC LIMIT ?`,
+       AND counts.asset_id NOT IN (SELECT asset_id FROM curated_lowq)
+       AND counts.asset_id NOT IN (SELECT asset_id FROM tagged_assets)
+     ORDER BY counts.collector_holders DESC,counts.asset_id LIMIT ?`,
     limit,
   );
 }
@@ -75,15 +88,14 @@ export interface ChosenCountRow {
   chosen_collectors: number;
 }
 
-/** Stage 2 (batch ≤ ~30 asset ids): collector holders with a CHOSEN acquisition path — any dispense,
- *  any DEX match, or a send from anyone but the asset's single widest-reaching sender. */
+/** Stage 2: collector holders with a CHOSEN acquisition path — any dispense, any DEX match, or a
+ *  send from anyone but the asset's single widest-reaching sender. */
 export function chosenCollectorCounts(db: D1Database, assetIds: number[]): Promise<ChosenCountRow[]> {
   if (assetIds.length === 0) return Promise.resolve([]);
-  const placeholders = assetIds.map(() => "?").join(",");
+  const ids = [...new Set(assetIds)];
   return q<ChosenCountRow>(
     db,
-    `WITH target(asset_id) AS (SELECT asset_id FROM asset_dictionary WHERE asset_id IN (${placeholders})),
-     collectors AS MATERIALIZED (${COLLECTOR_SET}),
+    `WITH target(asset_id) AS (SELECT CAST(value AS INTEGER) FROM json_each(?)),
      sender_reach AS (
        SELECT s.asset_id, s.source_id,
               ROW_NUMBER() OVER (PARTITION BY s.asset_id ORDER BY COUNT(DISTINCT s.destination_id) DESC) rn
@@ -104,11 +116,13 @@ export function chosenCollectorCounts(db: D1Database, assetIds: number[]): Promi
         WHERE om.tx0_address_id IS NOT NULL
      )
      SELECT b.asset_id, COUNT(DISTINCT b.address_id) chosen_collectors
-     FROM balances b
-     JOIN collectors c ON c.address_id=b.address_id
-     JOIN acquirers acq ON acq.asset_id=b.asset_id AND acq.address_id=b.address_id
-     WHERE CAST(b.quantity AS INTEGER) > 0
+     FROM acquirers acq
+     JOIN balances b INDEXED BY idx_balances_asset_address
+       ON b.asset_id=acq.asset_id AND b.address_id=acq.address_id
+     JOIN address_signals s ON s.address_id=b.address_id
+     JOIN address_reputations r ON r.address_id=b.address_id
+     WHERE CAST(b.quantity AS INTEGER)>0 AND ${COLLECTOR_PREDICATE}
      GROUP BY b.asset_id`,
-    ...assetIds,
+    JSON.stringify(ids),
   );
 }

apps/api/src/read/candidates.ts

@@ -12,7 +12,6 @@ import { collectorCandidateSeeds, chosenCollectorCounts } from "#api/queries/can
 export const candidates = router();
 
 const SEED_LIMIT = 150;
-const CHOSEN_BATCH = 30;
 const BOARD_SIZE = 100;
 
 candidates.get("/v2/collections/candidates", (c) =>
@@ -23,11 +22,11 @@ candidates.get("/v2/collections/candidates", (c) =>
     async (): Promise<Envelope<CollectionCandidatesPayload>> => {
       const seeds = await collectorCandidateSeeds(c.env.CORE_DB, SEED_LIMIT);
       const chosen = new Map<number, number>();
-      for (let index = 0; index < seeds.length; index += CHOSEN_BATCH) {
-        const batch = seeds.slice(index, index + CHOSEN_BATCH).map((seed) => seed.asset_id);
-        for (const row of await chosenCollectorCounts(c.env.CORE_DB, batch)) {
-          chosen.set(row.asset_id, row.chosen_collectors);
-        }
+      for (const row of await chosenCollectorCounts(
+        c.env.CORE_DB,
+        seeds.map((seed) => seed.asset_id),
+      )) {
+        chosen.set(row.asset_id, row.chosen_collectors);
       }
       const board: CollectionCandidate[] = seeds
         .map((seed) => ({

apps/api/tests/candidates.test.ts

@@ -0,0 +1,102 @@
+import assert from "node:assert/strict";
+import { readFileSync, readdirSync } from "node:fs";
+import { DatabaseSync } from "node:sqlite";
+import { test } from "node:test";
+import { chosenCollectorCounts, collectorCandidateSeeds } from "#api/queries/candidates";
+
+const migrations = readdirSync("migrations-core")
+  .filter((name) => name.endsWith(".sql"))
+  .sort()
+  .map((name) => readFileSync(`migrations-core/${name}`, "utf8"));
+
+class Statement {
+  private values: unknown[] = [];
+  constructor(
+    private readonly db: DatabaseSync,
+    private readonly sql: string,
+  ) {}
+  bind(...values: unknown[]) {
+    this.values = values;
+    return this;
+  }
+  async all<T>() {
+    return { results: this.db.prepare(this.sql).all(...this.values) as T[] };
+  }
+}
+
+const d1 = (db: DatabaseSync) => ({ prepare: (sql: string) => new Statement(db, sql) }) as unknown as D1Database;
+
+test("collection candidates aggregate holders before enrichment and retain the chosen-acquisition gate", async () => {
+  const db = new DatabaseSync(":memory:");
+  for (const migration of migrations) db.exec(migration);
+  db.exec(`
+    INSERT INTO address_dictionary(address)
+      VALUES('1distributor'),('1issuer'),('1collector1'),('1collector2'),('1collector3'),('1collector4'),('1collector5');
+    INSERT INTO address_signals(address_id,assets_held,assets_received)
+      SELECT address_id,10,20 FROM address_dictionary WHERE address LIKE '1collector%';
+    INSERT INTO address_reputations(
+      address_id,reputation,rank_position,population,duration_score,creation_score,
+      economic_score,participation_score,calculated_at,model_version)
+      SELECT address_id,50,1,5,0,0,0,0,1,1 FROM address_dictionary WHERE address LIKE '1collector%';
+
+    INSERT INTO asset_dictionary(asset) VALUES('CHOSEN'),('AIRDROP'),('TAGGED'),('LOWQ');
+    INSERT INTO assets(asset_id,type,issuer_id,divisible,locked,first_issuance_block_index)
+      SELECT asset_id,'asset',(SELECT address_id FROM address_dictionary WHERE address='1issuer'),0,1,1
+      FROM asset_dictionary WHERE asset IN ('CHOSEN','AIRDROP','TAGGED','LOWQ');
+    INSERT INTO asset_signals(asset_id,holders,low_quality)
+      SELECT asset_id,5,CASE asset WHEN 'LOWQ' THEN 1 ELSE 0 END
+      FROM asset_dictionary WHERE asset IN ('CHOSEN','AIRDROP','TAGGED','LOWQ');
+    INSERT INTO balances(address_id,asset_id,quantity,updated_event_index)
+      SELECT address.address_id,asset.asset_id,'1',1
+      FROM address_dictionary address CROSS JOIN asset_dictionary asset
+      WHERE address.address LIKE '1collector%' AND asset.asset IN ('CHOSEN','AIRDROP','TAGGED','LOWQ');
+
+    INSERT INTO entity_dictionary(entity_type,entity_key) VALUES('asset','TAGGED');
+    INSERT INTO collection_membership_evidence(entity_id,tag,source)
+      SELECT entity_id,'Known Collection','manual' FROM entity_dictionary
+      WHERE entity_type='asset' AND entity_key='TAGGED';
+
+    INSERT INTO sends(event_index,tx_index,tx_hash,block_index,source_id,destination_id,
+      source_address_id,destination_address_id,asset_id,quantity,msg_index)
+      SELECT 100+address.address_id,100+address.address_id,randomblob(32),1,
+        distributor.address_id,address.address_id,distributor.address_id,address.address_id,
+        asset.asset_id,'1',0
+      FROM address_dictionary address
+      JOIN address_dictionary distributor ON distributor.address='1distributor'
+      JOIN asset_dictionary asset ON asset.asset='AIRDROP'
+      WHERE address.address LIKE '1collector%';
+    INSERT INTO sends(event_index,tx_index,tx_hash,block_index,source_id,destination_id,
+      source_address_id,destination_address_id,asset_id,quantity,msg_index)
+      SELECT 200+address.address_id,200+address.address_id,randomblob(32),1,
+        distributor.address_id,address.address_id,distributor.address_id,address.address_id,
+        asset.asset_id,'1',0
+      FROM address_dictionary address
+      JOIN address_dictionary distributor ON distributor.address='1distributor'
+      JOIN asset_dictionary asset ON asset.asset='CHOSEN'
+      WHERE address.address LIKE '1collector%';
+    INSERT INTO dispenses(event_index,tx_index,dispense_index,tx_hash,dispenser_tx_index,
+      source_id,destination_id,asset_id,dispense_quantity,btc_amount,block_index)
+      SELECT 1,1,0,randomblob(32),1,distributor.address_id,collector.address_id,
+        asset.asset_id,'1','10000',1
+      FROM address_dictionary distributor,address_dictionary collector,asset_dictionary asset
+      WHERE distributor.address='1distributor' AND collector.address='1collector1' AND asset.asset='CHOSEN';
+  `);
+
+  const core = d1(db);
+  const seeds = await collectorCandidateSeeds(core, 150);
+  assert.deepEqual(
+    seeds.map(({ asset, collector_holders }) => ({ asset, collector_holders })),
+    [
+      { asset: "CHOSEN", collector_holders: 5 },
+      { asset: "AIRDROP", collector_holders: 5 },
+    ],
+  );
+
+  // One JSON bind supports the route's full 150-seed batch without crossing D1's SQL-variable ceiling.
+  const fillerIds = Array.from({ length: 148 }, (_, index) => 1_000_000 + index);
+  const chosen = await chosenCollectorCounts(core, [...seeds.map((seed) => seed.asset_id), ...fillerIds]);
+  assert.deepEqual(
+    chosen.map((row) => ({ ...row })),
+    [{ asset_id: seeds[0]!.asset_id, chosen_collectors: 1 }],
+  );
+});