apps/keeper/src/jobs/owners.ts

1// owners: who holds each NFT of a listed collection right now. Payouts go to whoever owns an NFT when they run.2//  - With HELIUS_API_KEY: Helius DAS getAssetsByGroup reads every NFT's owner and burn state (1,000 per call), its3//    on-chain image (arweave; public IPFS gateways rate-limit) and its name when HowRare had none.4//    Collections paired with a coin refresh every `owners.paired_every_s` (10 min); the rest every 6 h.5//  - Without it: Magic Eden's per-token endpoint, a bounded number per run (slow; enough to demo, not to pay a whole6//    collection). NFTs of paired collections first, rarest first, never-seen owners before stale ones.7// A program-owned NFT (escrow, pool, PDA) is marked owner_is_wallet = false: its rewards wait for the next wallet.8import { and, asc, desc, eq, inArray, isNotNull, isNull, lt, or, sql } from 'drizzle-orm';9import { collections, coins, nfts } from '@solary/db';10import { getSetting, type JobCtx, type Stats } from '../ctx';11import { isWalletAddress } from '../nft/sources';12 13interface OwnerRow {14  id: string;15  owner: string | null;16  burnt: boolean;17  /** the on-chain image (DAS content.links.image, usually arweave): replaces HowRare's IPFS mirror when present */18  image?: string | null;19  /** fills the name only when HowRare had none */20  name?: string | null;21}22 23async function writeOwners(ctx: Pick<JobCtx, 'db'>, rows: readonly OwnerRow[]): Promise<void> {24  for (let i = 0; i < rows.length; i += 1000) {25    const chunk = rows.slice(i, i + 1000);26    const values = sql.join(27      chunk.map((r) => sql`(${r.id}, ${r.owner}, ${r.owner ? isWalletAddress(r.owner) : false}, ${r.burnt}, ${r.image || null}, ${r.name || null})`),28      sql`, `,29    );30    await ctx.db.execute(sql`31      UPDATE nfts SET owner = v.owner, owner_is_wallet = v.wallet, burnt = v.burnt,32        image = COALESCE(v.image::text, nfts.image),33        name = CASE WHEN nfts.name = '' THEN COALESCE(v.name::text, '') ELSE nfts.name END,34        updated_at = now()35      FROM (VALUES ${values}) AS v(id, owner, wallet, burnt, image, name)36      WHERE nfts.id = v.id`);37  }38}39 40async function refreshTotals(ctx: Pick<JobCtx, 'db'>, collection: string): Promise<void> {41  await ctx.db.execute(sql`42    UPDATE collections SET43      total_shares = COALESCE((SELECT SUM(multiplier) FROM nfts WHERE collection = ${collection} AND NOT burnt), 0),44      owners = GREATEST((SELECT COUNT(DISTINCT owner) FROM nfts WHERE collection = ${collection} AND owner IS NOT NULL AND NOT burnt), owners),45      owners_at = now()46    WHERE address = ${collection}`);47}48 49/** Every owner and burn of one collection from Helius DAS (needs HELIUS_API_KEY); also used by the demo seed. */50export async function refreshCollectionOwners(ctx: Pick<JobCtx, 'db' | 'nft'>, address: string): Promise<{ nfts: number; unranked: number }> {51  if (!ctx.nft.das) throw new Error('HELIUS_API_KEY is not set');52  const assets = await ctx.nft.das.collectionAssets(address);53  const knownIds = new Set((await ctx.db.select({ id: nfts.id }).from(nfts).where(eq(nfts.collection, address))).map((r) => r.id));54  const rows = assets.filter((a) => knownIds.has(a.id)).map((a) => ({ id: a.id, owner: a.owner, burnt: a.burnt, image: a.image, name: a.name }));55  await writeOwners(ctx, rows);56  await refreshTotals(ctx, address);57  return { nfts: rows.length, unranked: assets.length - rows.length };58}59 60export async function ownersJob(ctx: JobCtx): Promise<Stats> {61  const { db, nft } = ctx;62  const pairedRows = await db.selectDistinct({ c: coins.collection }).from(coins).where(isNotNull(coins.collection));63  const paired = new Set(pairedRows.map((r) => r.c!));64  const listed = await db65    .select({ address: collections.address, name: collections.name, ownersAt: collections.ownersAt })66    .from(collections)67    .where(inArray(collections.status, ['listed', 'disabled']));68 69  if (nft.das) {70    const pairedEvery = (await getSetting<number>(db, 'owners.paired_every_s', 600)) * 1000;71    const otherEvery = (await getSetting<number>(db, 'owners.other_every_s', 6 * 3600)) * 1000;72    const due = listed73      .filter((c) => Date.now() - (c.ownersAt?.getTime() ?? 0) > (paired.has(c.address) ? pairedEvery : otherEvery))74      .sort((a, b) => Number(paired.has(b.address)) - Number(paired.has(a.address)) || (a.ownersAt?.getTime() ?? 0) - (b.ownersAt?.getTime() ?? 0))75      .slice(0, 4);76    let updated = 0;77    for (const c of due) {78      const r = await refreshCollectionOwners(ctx, c.address);79      updated += r.nfts;80      ctx.log(`${c.name}: ${r.nfts} owners from Helius (${r.unranked} assets not in the ranked list)`);81    }82    return { source: 'helius', collections: due.length, nfts: updated };83  }84 85  // Magic Eden fallback, bounded86  const budget = await getSetting<number>(db, 'owners.me_per_run', ctx.env.fast ? 30 : 40);87  const pairedList = [...paired];88  if (!pairedList.length) return { source: 'magic-eden', skipped: 'no paired collections yet' };89  const staleBefore = new Date(Date.now() - 24 * 3600_000);90  const todo = await db91    .select({ id: nfts.id, collection: nfts.collection })92    .from(nfts)93    .where(and(inArray(nfts.collection, pairedList), eq(nfts.burnt, false), or(isNull(nfts.owner), lt(nfts.updatedAt, staleBefore))))94    .orderBy(sql`${nfts.owner} IS NOT NULL`, desc(nfts.multiplier), asc(nfts.rank))95    .limit(budget);96  const rows: OwnerRow[] = [];97  let failed = 0;98  for (const t of todo) {99    try {100      const o = await nft.me.owner(t.id);101      rows.push({ id: t.id, owner: o.owner, burnt: false });102    } catch {103      failed++;104    }105  }106  await writeOwners(ctx, rows);107  for (const c of new Set(todo.map((t) => t.collection))) await refreshTotals(ctx, c);108  return { source: 'magic-eden', looked_up: rows.length, failed };109}