From 33113fafd4558496e70f96b72fb25296ec76de05 Mon Sep 17 00:00:00 2001 From: Dan Anderson Date: Wed, 28 Oct 2026 22:51:03 -0400 Subject: [PATCH] Bound collection candidate D1 work --- apps/api/src/queries/candidates.ts | 84 ++++++++++++++---------- apps/api/src/read/candidates.ts | 11 ++-- apps/api/tests/candidates.test.ts | 102 +++++++++++++++++++++++++++++ 3 files changed, 156 insertions(+), 41 deletions(-) create mode 100644 apps/api/tests/candidates.test.ts diff --git a/apps/api/src/queries/candidates.ts b/apps/api/src/queries/candidates.ts index 6c986c7..214e5e0 100644 --- a/apps/api/src/queries/candidates.ts +++ b/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 { return q( 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 { if (assetIds.length === 0) return Promise.resolve([]); - const placeholders = assetIds.map(() => "?").join(","); + const ids = [...new Set(assetIds)]; return q( 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), ); } diff --git a/apps/api/src/read/candidates.ts b/apps/api/src/read/candidates.ts index 4a2ad50..75d5d22 100644 --- a/apps/api/src/read/candidates.ts +++ b/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> => { const seeds = await collectorCandidateSeeds(c.env.CORE_DB, SEED_LIMIT); const chosen = new Map(); - 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) => ({ diff --git a/apps/api/tests/candidates.test.ts b/apps/api/tests/candidates.test.ts new file mode 100644 index 0000000..d988501 --- /dev/null +++ b/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() { + 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 }], + ); +});