CashuMints.space/api/src/discovery.ts

import { SimplePool } from 'nostr-tools/pool';
import type { Event as NostrEvent, Filter } from 'nostr-tools';
import {
  ANNOUNCEMENT_KINDS,
  KIND_FEDIMINT_ANNOUNCEMENT,
  KIND_LNURL_ANNOUNCEMENT,
  KIND_MINT_ANNOUNCEMENT,
  KIND_REVIEW,
  mintPubkeyRef,
  mintUrlSpellings,
  mintUrlsFromEvent,
  normalizeMintUrl,
  parseFedimintAnnouncement,
  parseLnurlAnnouncement,
  parseRating,
  reviewEcosystem,
  reviewTargetId,
  reviewTargetKind,
  type FedimintAnnouncement,
  type LnurlAnnouncement,
  type LnurlFields,
  type RelayHealth,
} from '@cashumints/shared';
import { config } from './config.ts';
import { getDb, setState, getState, getStateNumber } from './db.ts';
import type { Sql } from './db-driver.ts';
import { log } from './log.ts';
import { insertMintIfNew, upsertFedimint, upsertLnurl } from './mints.ts';

const QUERY_LIMIT = 500;
const MAX_WAIT_MS = 12_000;
/** Backfill page cap. 20 pages x 500 is far more than the network holds today. */
const MAX_PAGES = 20;
/** The old site's second filter: a recent window alongside the unbounded one, because
 *  some relays answer an unbounded query with an arbitrary slice. See NOTES.md. */
const RECENT_WINDOW_S = 30 * 24 * 60 * 60;

export interface DiscoveryResult {
  events: number;
  newMints: string[];
  newReviews: number;
  ok: boolean;
  /** What each configured relay actually did, in `config.relays` order. */
  relays: RelayHealth[];
  /** A backfill that came in under `config.backfillMinEvents`. */
  starved: boolean;
}

/**
 * What the last cycle did, kept so /api/health can answer for it.
 *
 * Written to the `state` table rather than held in memory, because the question it
 * answers — "is discovery actually reading anything?" — has to survive the restart that
 * would otherwise reset it to "no cycle yet, nothing to report". A process that crash
 * loops would clear an in-memory flag on every attempt.
 */
export interface DiscoveryReport {
  at: number;
  mode: 'backfill' | 'incremental';
  events: number;
  ok: boolean;
  relays: RelayHealth[];
  starved: boolean;
}

/** `state` key holding the JSON of the above. */
const REPORT_KEY = 'last_discovery_report';

/**
 * The last cycle's report, or null before any cycle has run.
 *
 * A row that will not parse reads as null — the same as no cycle — because the caller
 * is a health endpoint and "I cannot tell you" must not be dressed up as "fine".
 */
export async function lastDiscoveryReport(): Promise<DiscoveryReport | null> {
  const raw = await getState(REPORT_KEY);
  if (!raw) return null;
  try {
    const parsed = JSON.parse(raw) as DiscoveryReport;
    return Array.isArray(parsed.relays) ? parsed : null;
  } catch {
    return null;
  }
}

/**
 * Per-relay bookkeeping for one cycle.
 *
 * Every relay in `config.relays` gets a row up front, including the ones that are never
 * reached, because a relay that produced no row at all is exactly the one worth naming:
 * the year-long starvation was a relay list that connected cleanly and simply did not
 * hold the archive, and the only field that would have shown it is a zero here.
 *
 * `events` counts what a relay sent *before* cross-relay deduplication, so five relays
 * carrying the same 400 events report 400 each rather than 400 once and 0 four times.
 * Attribution is the whole point; the deduplicated total is reported separately.
 */
class RelayTally {
  private readonly rows = new Map<string, { events: number; subs: number; eoses: number; connected: boolean }>();

  constructor(urls: readonly string[]) {
    for (const url of urls) {
      this.rows.set(url, { events: 0, subs: 0, eoses: 0, connected: false });
    }
  }

  private row(url: string) {
    let found = this.rows.get(url);
    if (!found) {
      found = { events: 0, subs: 0, eoses: 0, connected: false };
      this.rows.set(url, found);
    }
    return found;
  }

  connected(url: string): void {
    this.row(url).connected = true;
  }

  subscribed(url: string): void {
    this.row(url).subs++;
  }

  event(url: string): void {
    this.row(url).events++;
  }

  eose(url: string): void {
    this.row(url).eoses++;
  }

  /** One row per configured relay, in configuration order. */
  list(): RelayHealth[] {
    return [...this.rows.entries()].map(([url, row]) => ({
      url,
      connected: row.connected,
      events: row.events,
      // A cycle asks a relay many questions. It only counts as having reached the end
      // of the stream if it reached the end of every one of them.
      eose: row.subs > 0 && row.eoses === row.subs,
    }));
  }
}

/**
 * Long enough that the relay's own EOSE timer never wins.
 *
 * `Subscription` fires `oneose` both when an EOSE frame arrives and when its internal
 * timer expires, so the two are indistinguishable from the callback. Pushing that timer
 * out of reach and running the deadline here instead is what makes `eose` in the report
 * mean "the relay said it was done" rather than "something gave up".
 */
const NEVER_EOSE_MS = 24 * 60 * 60 * 1000;

let pool: SimplePool | null = null;

/**
 * The process's one relay pool.
 *
 * Exported because `relay-lookup.ts` asks the same relays a different question — "has
 * anyone ever mentioned this address?", for a mint a reader just submitted that does
 * not answer — and opening a second pool for it would mean a second set of sockets to
 * the same five relays, with its own reconnect behaviour and its own lifetime to get
 * wrong at shutdown.
 */
export function getPool(): SimplePool {
  pool ??= new SimplePool();
  return pool;
}

export function closePool(): void {
  try {
    pool?.close(config.relays);
  } catch {
    // A relay that is already gone throws on close. Nothing to do about it.
  }
  pool = null;
}

/**
 * Ask every configured relay one filter, and record what each of them did.
 *
 * This replaces `pool.querySync(config.relays, …)`, which answers the same question and
 * throws the attribution away: it merges five relays into one deduplicated array, so a
 * relay list where four relays are empty and one carries everything is indistinguishable
 * from five healthy ones. That indistinguishability is the bug this whole file is being
 * changed for — a year of ~31-event backfills, `ok=true` every time.
 *
 * What it keeps from `querySync`, deliberately:
 *
 *   - One subscription per relay over the pool's existing sockets, so this is the same
 *     number of connections as before.
 *   - A single `alreadyHaveEvent` shared across all five. `AbstractRelay._onmessage`
 *     consults it *before* `JSON.parse` and signature verification, so an event five
 *     relays all carry is still verified once. Per-relay `querySync` calls would have
 *     verified it five times, which at 500 events a page is real CPU.
 *   - `receivedEvent`, which fires on the way past that check, so the per-relay count is
 *     what the relay sent rather than what was new because of it.
 *
 * What it changes: the deadline is run here rather than by each `Subscription`'s own
 * EOSE timer, so `oneose` firing means an EOSE frame actually arrived. See NEVER_EOSE_MS.
 *
 * Never throws. A relay that will not connect is a fact to record, not a reason to
 * abandon the four that did.
 */
async function queryRelays(filter: Filter, tally: RelayTally | null): Promise<NostrEvent[]> {
  const events: NostrEvent[] = [];
  const known = new Set<string>();
  const alreadyHaveEvent = (id: string): boolean => {
    if (known.has(id)) return true;
    known.add(id);
    return false;
  };

  await Promise.all(
    config.relays.map(async (url) => {
      let relay;
      try {
        // The same connection budget subscribeMap would have used for this maxWait.
        relay = await getPool().ensureRelay(url, {
          connectionTimeout: Math.max(MAX_WAIT_MS * 0.8, MAX_WAIT_MS - 1000),
        });
      } catch {
        // Left as connected=false in the tally, which is the whole report this needs.
        return;
      }
      tally?.connected(url);

      await new Promise<void>((resolve) => {
        let settled = false;
        let deadline: ReturnType<typeof setTimeout> | undefined;
        const finish = (): void => {
          if (settled) return;
          settled = true;
          if (deadline !== undefined) clearTimeout(deadline);
          resolve();
        };

        try {
          const sub = relay.subscribe([filter], {
            onevent: (event) => events.push(event),
            alreadyHaveEvent,
            receivedEvent: () => tally?.event(url),
            oneose: () => {
              tally?.eose(url);
              sub.close('closed automatically on eose');
            },
            onclose: finish,
            eoseTimeout: NEVER_EOSE_MS,
          });
          tally?.subscribed(url);
          deadline = setTimeout(() => sub.close('closed on maxWait'), MAX_WAIT_MS);
        } catch {
          // The socket went away between ensureRelay and the REQ.
          finish();
        }
      });
    }),
  );

  return events;
}

/**
 * Query one kind, paging backwards with `until` until a page yields nothing new.
 * Relays cap `limit` independently, so paging is the only way a fresh database
 * converges to the complete history.
 */
async function fetchKind(
  kind: number,
  since: number | null,
  tally: RelayTally | null,
): Promise<NostrEvent[]> {
  const seen = new Map<string, NostrEvent>();
  let until: number | undefined;

  for (let page = 0; page < MAX_PAGES; page++) {
    const filter: Filter = { kinds: [kind], limit: QUERY_LIMIT };
    if (since !== null) filter.since = since;
    if (until !== undefined) filter.until = until;

    let batch: NostrEvent[];
    try {
      batch = await queryRelays(filter, tally);
    } catch (err) {
      log.warn('relay query failed', {
        kind,
        page,
        reason: err instanceof Error ? err.message : String(err),
      });
      break;
    }

    let added = 0;
    let oldest = Number.POSITIVE_INFINITY;
    for (const e of batch) {
      oldest = Math.min(oldest, e.created_at);
      if (!seen.has(e.id)) {
        seen.set(e.id, e);
        added++;
      }
    }

    // Nothing new, or the relays returned less than a full page: history is exhausted.
    if (added === 0 || batch.length < QUERY_LIMIT || !Number.isFinite(oldest)) break;
    until = oldest - 1;
    if (since !== null && until <= since) break;
  }

  return [...seen.values()];
}

/**
 * Reviews for one specific mint, asked for by name.
 *
 * The global sweep above does NOT converge on its own: relays answer an unbounded
 * `{kinds:[38000]}` with an arbitrary capped slice, but answer a filtered query for one
 * mint in full. Measured against the live network, the global sweep found 43 reviews for
 * mint.minibits.cash/Bitcoin while this targeted query finds 80. Without this pass the
 * count in the API disagrees with the list the browser renders on the same page.
 */
async function fetchReviewsForMint(
  target: ReviewTarget,
  since: number | null,
  tally: RelayTally | null,
): Promise<NostrEvent[]> {
  const filters: Filter[] = [];
  const base: Filter = { kinds: [KIND_REVIEW], limit: QUERY_LIMIT };
  if (since !== null) base.since = since;

  if (target.type === 'fedimint') {
    /*
     * A federation is asked for by id, and only by id.
     *
     * The mint side asks by `#u` as well, and the obvious translation of that is `#u`
     * with the invite codes. Two of the five relays in the pool answer such a filter
     * with `ERROR: bad req: filter item too large` and nothing else — an invite code is
     * 150+ characters — so the query costs two round trips and returns less than
     * sending nothing would.
     *
     * It also buys nothing today: every kind 38000 with `k` = 38173 on the network
     * carries the federation id in `d` and no `u` tag at all. A review that arrives
     * with only an invite code is still matched, by the global sweep, through
     * `byInvite` in the index above; it just is not asked for by name.
     */
    if (!target.federationId) return [];
    filters.push({
      ...base,
      '#d': [target.federationId],
      '#k': [String(KIND_FEDIMINT_ANNOUNCEMENT)],
    });
  } else if (target.type === 'lnurl') {
    /*
     * Asked for by every identifier it has ever had, not just its current one.
     *
     * An LNURL mint's `d` is its funding node's pubkey when it has one and its host
     * otherwise, so a mint that gained a funding source has reviews filed under both
     * spellings — the older ones under the host, the newer under the pubkey. One `#d`
     * filter carrying both is one round trip and finds all of them; asking for only the
     * current identifier would silently strand the earlier half. See the kind document.
     *
     * `#u` is asked as well, and unlike a federation's invite codes an LNURL mint's URLs
     * are short enough that no relay rejects the filter.
     */
    if (target.identifiers.length > 0) {
      filters.push({
        ...base,
        '#d': target.identifiers,
        '#k': [String(KIND_LNURL_ANNOUNCEMENT)],
      });
    }
    filters.push({ ...base, '#u': mintUrlSpellings(target.baseUrl ?? target.url) });
  } else {
    if (target.pubkey) {
      filters.push({ ...base, '#d': [target.pubkey], '#k': [String(KIND_MINT_ANNOUNCEMENT)] });
    }
    filters.push({ ...base, '#u': mintUrlSpellings(target.url) });
  }

  const batches = await Promise.all(
    filters.map((filter) => queryRelays(filter, tally).catch(() => [] as NostrEvent[])),
  );

  return batches.flat();
}

/**
 * One thing the targeted second pass asks the relays about.
 *
 * A mint is asked for by URL and by the pubkey it publishes; a federation by its id and
 * its invite codes. Nothing else about the row is needed, so the pass reads five
 * columns rather than the whole table.
 */
interface ReviewTarget {
  type: string;
  /** Row key: the mint URL, `fedimint:<id>`, or `lnurl:<base url>`. */
  url: string;
  pubkey: string | null;
  federationId: string | null;
  /** LNURL: every `d` this mint could be reviewed under. Empty for the others. */
  identifiers: string[];
  /** LNURL: the fetchable https URL inside the row key. */
  baseUrl: string | null;
}

/**
 * Every federation the batch announces, inserted or refreshed.
 *
 * Unlike mints, a federation's row content comes entirely from these events — there is
 * nothing to fetch afterwards — so this both creates rows and keeps existing ones
 * current. Deduped by federation id inside the batch first, newest announcement per id
 * winning, because the same federation is routinely announced by several npubs and by
 * the same npub more than once, and writing each of those would be one UPDATE per event
 * to land on the same final state.
 */
async function ingestFedimints(
  events: Iterable<NostrEvent>,
  now: number,
  newRows: Set<string>,
): Promise<void> {
  const newest = new Map<string, FedimintAnnouncement>();

  for (const event of events) {
    const announcement = parseFedimintAnnouncement(event);
    if (!announcement) continue;
    const existing = newest.get(announcement.federationId);
    if (existing && existing.announcedAt >= announcement.announcedAt) continue;
    newest.set(announcement.federationId, announcement);
  }

  for (const announcement of newest.values()) {
    const created = await upsertFedimint(announcement, now);
    if (created) newRows.add(created);
  }
}

/**
 * Every LNURL mint the batch announces, inserted or refreshed.
 *
 * Deduped **by normalized base URL**, not by the `d` tag, which is the one place this
 * genuinely differs from `ingestFedimints`. A federation has exactly one identity for
 * life; an LNURL mint's identifier is its funding node's pubkey when it has one and its
 * host otherwise, so the same mint legitimately announces under two different `d` values
 * across its life. Keying on `d` would split it into two pages with half its reviews on
 * each. The `u` tag is present in every form of the event, so it is the key.
 *
 * Newest announcement per URL wins inside the batch, for the same reason the Fedimint
 * side does it: the same mint is announced by more than one npub and more than once, and
 * writing each of those would be one UPDATE per event to reach the same final state.
 */
async function ingestLnurl(
  events: Iterable<NostrEvent>,
  now: number,
  newRows: Set<string>,
): Promise<void> {
  const newest = new Map<string, LnurlAnnouncement>();

  for (const event of events) {
    const announcement = parseLnurlAnnouncement(event);
    if (!announcement) continue;
    const existing = newest.get(announcement.baseUrl);
    if (existing && existing.announcedAt >= announcement.announcedAt) continue;
    newest.set(announcement.baseUrl, announcement);
  }

  for (const announcement of newest.values()) {
    const created = await upsertLnurl(announcement, now);
    if (created) newRows.add(created);
  }
}

/**
 * Every mint URL the batch points at, normalized and inserted if new.
 *
 * Deduped across the whole batch before anything is written. A popular mint's URL
 * appears in hundreds of events per cycle, and the old row-at-a-time version paid one
 * INSERT round trip for each of them — invisible against a local SQLite file, minutes
 * of latency against a Postgres server.
 */
async function ingestMintUrls(
  events: Iterable<NostrEvent>,
  now: number,
  newMints: Set<string>,
): Promise<void> {
  const seen = new Set<string>();
  for (const event of events) {
    // A review that says it is about a federation carries an invite code in `u`, not a
    // mint URL, and feeding those to the mint normalizer is how `fed11…` filled the
    // skipped-URL log. (A handful of kind 38172 announcements carry one too, which is
    // their publisher's mistake and still logs once.)
    if (reviewEcosystem(event) !== 'cashu') continue;
    for (const raw of mintUrlsFromEvent(event)) seen.add(raw);
  }

  for (const raw of seen) {
    const created = await insertMintIfNew(raw, now);
    if (created) newMints.add(created);
  }
}

/**
 * The mints table as two lookup maps, read once per cycle.
 *
 * Review resolution asks "which mint is this?" once per event, and the answer only
 * changes when a mint is inserted — which happens in ingestMintUrls, before any of
 * this runs. So it is a snapshot, not a query per event.
 */
interface MintIndex {
  byHost: Map<string, string>;
  byPubkey: Map<string, string>;
  /**
   * Federation id to row key. A federation has no URL, so its `d` tag is the only thing
   * a review can be matched on, and it is matched exactly.
   */
  byFederation: Map<string, string>;
  /** Invite code to row key, for a review that carries `u` but no usable `d`. */
  byInvite: Map<string, string>;
  /**
   * Every LNURL identifier to its row key: the mint pubkey *and* the normalized host,
   * both pointing at the same row.
   *
   * Kept apart from `byPubkey`, which is the Cashu index over `mints.pubkey`. An LNURL
   * mint's key is its funding node's identity, not a Cashu mint pubkey, and mixing the
   * two namespaces would let a review of one resolve to the other. Nothing writes
   * `mints.pubkey` for an LNURL row, for exactly that reason.
   */
  byLnurlId: Map<string, string>;
  /** Normalized base URL to row key, for a review that carries `u` but no usable `d`. */
  byLnurlUrl: Map<string, string>;
}

async function loadMintIndex(): Promise<MintIndex> {
  const db = await getDb();
  const rows = await db.all<{
    url: string;
    host: string;
    type: string;
    pubkey: string | null;
    ecosystem_json: string | null;
  }>('SELECT url, host, type, pubkey, ecosystem_json FROM mints');

  const byHost = new Map<string, string>();
  const byPubkey = new Map<string, string>();
  const byFederation = new Map<string, string>();
  const byInvite = new Map<string, string>();
  const byLnurlId = new Map<string, string>();
  const byLnurlUrl = new Map<string, string>();

  for (const row of rows) {
    byHost.set(row.host, row.url);
    // First writer wins, matching the LIMIT 1 the per-event query used.
    if (row.pubkey && !byPubkey.has(row.pubkey)) byPubkey.set(row.pubkey, row.url);

    if (row.type === 'lnurl' && row.ecosystem_json) {
      try {
        const fields = JSON.parse(row.ecosystem_json) as Partial<LnurlFields>;
        // Both identifiers, always: a mint announced by host and later by pubkey has
        // reviews under each, and both belong to this one row.
        for (const id of [fields.lnurl_id, fields.mint_pubkey]) {
          if (id) byLnurlId.set(id.toLowerCase(), row.url);
        }
        if (fields.base_url) byLnurlUrl.set(fields.base_url, row.url);
      } catch {
        // Unparseable column: this row simply cannot be matched by identifier. It still
        // has a page and still resolves by its key, which is honest.
      }
      continue;
    }

    if (row.type !== 'fedimint' || !row.ecosystem_json) continue;
    try {
      const fields = JSON.parse(row.ecosystem_json) as {
        federation_id?: string;
        invite_codes?: string[];
      };
      if (fields.federation_id) byFederation.set(fields.federation_id.toLowerCase(), row.url);
      for (const code of fields.invite_codes ?? []) byInvite.set(code.toLowerCase(), row.url);
    } catch {
      // A row whose ecosystem column will not parse simply cannot be matched by id.
      // It still has a page and still resolves by nothing else, which is honest.
    }
  }

  return { byHost, byPubkey, byFederation, byInvite, byLnurlId, byLnurlUrl };
}

/**
 * Resolve which mint a review is about.
 *
 * `u` tag first, resolved through the slug so that every spelling of a mint URL (with
 * or without a trailing slash, either path case) lands on the one canonical row. Then
 * the `d` tag pubkey against the pubkey a mint publishes in its own /v1/info, which
 * catches reviews whose `u` tag is missing or wrong.
 */
function resolveReviewTarget(event: NostrEvent, index: MintIndex): string | null {
  /*
   * The `k` tag decides which ecosystem's resolver runs, and it is not merely a hint:
   * a federation id and a Cashu mint pubkey are both 64 hex characters in a `d` tag, so
   * without this a Fedimint review could in principle be filed against a mint. A review
   * with no `k` at all is Cashu, because every one of those predates anything else.
   */
  const ecosystem = reviewEcosystem(event);
  if (ecosystem === 'fedimint') return resolveFedimintReview(event, index);
  if (ecosystem === 'lnurl') return resolveLnurlReview(event, index);
  // A `k` naming a kind this build has no ecosystem for: not ours to file.
  if (ecosystem === null) return null;

  for (const raw of mintUrlsFromEvent(event)) {
    const normalized = normalizeMintUrl(raw);
    if (!normalized) continue;
    // The row was inserted moments ago by ingestMintUrls, so this almost always hits.
    const canonical = index.byHost.get(normalized.host);
    if (canonical) return canonical;
  }

  const pubkey = mintPubkeyRef(event);
  return pubkey ? index.byPubkey.get(pubkey) ?? null : null;
}

/**
 * Which federation a review is about.
 *
 * `d` first and almost always: every kind 38000 with `k` = 38173 seen on the network
 * carries the federation id there and nothing else — no `u`, no `a`. The invite code
 * fallback is for publishers that write `u` instead, and matches the code exactly
 * rather than decoding it, because an invite code is opaque to this codebase.
 *
 * A review of a federation this site has never seen announced resolves to nothing and
 * is dropped, exactly as a review of an unknown mint is. There is no page to put it on.
 */
function resolveFedimintReview(event: NostrEvent, index: MintIndex): string | null {
  const d = reviewTargetId(event);
  if (d && /^[0-9a-f]{64}$/i.test(d)) {
    const found = index.byFederation.get(d.toLowerCase());
    if (found) return found;
  }

  for (const raw of mintUrlsFromEvent(event)) {
    const found = index.byInvite.get(raw.trim().toLowerCase());
    if (found) return found;
  }

  return null;
}

/**
 * Which LNURL mint a review is about.
 *
 * `d` first, matched against both identifiers a mint can have — its funding node's
 * pubkey and its normalized host — because a mint that gained a funding source has
 * reviews written under each and both name the same page.
 *
 * The `u` fallback is not the afterthought it is on the Fedimint side. An LNURL mint's
 * `u` tag is a real, normalizable https URL, so a review that carries only an address
 * resolves exactly as a Cashu review does, through the same normalizer. That matters
 * for reviews written by clients that do not know this kind: `u` is the tag they are
 * most likely to get right.
 *
 * A review of a mint this site has never seen resolves to nothing and is dropped, as it
 * is for the other two ecosystems. There is no page to put it on.
 */
function resolveLnurlReview(event: NostrEvent, index: MintIndex): string | null {
  const d = reviewTargetId(event);
  if (d) {
    const found = index.byLnurlId.get(d.trim().toLowerCase());
    if (found) return found;
  }

  for (const raw of mintUrlsFromEvent(event)) {
    const normalized = normalizeMintUrl(raw);
    if (!normalized) continue;
    const found = index.byLnurlUrl.get(normalized.url);
    if (found) return found;
  }

  return null;
}

/** Postgres caps a statement at 65535 bounds parameters; this keeps every batch clear of it. */
const INSERT_BATCH = 500;

interface ReviewRow {
  eventId: string;
  mintUrl: string;
  pubkey: string;
  rating: number | null;
  /** The `k` tag verbatim, or null when the event carried none. */
  k: string | null;
  createdAt: number;
}

/** Which of these event ids the reviews table already holds. */
async function knownEventIds(ids: string[]): Promise<Set<string>> {
  const db = await getDb();
  const known = new Set<string>();

  for (let i = 0; i < ids.length; i += INSERT_BATCH) {
    const chunk = ids.slice(i, i + INSERT_BATCH);
    const rows = await db.all<{ event_id: string }>(
      `SELECT event_id FROM reviews WHERE event_id IN (${chunk.map(() => '?').join(',')})`,
      ...chunk,
    );
    for (const row of rows) known.add(row.event_id);
  }

  return known;
}

/**
 * Write a batch of reviews and report how many were genuinely new.
 *
 * Deduped by event id first, keeping the last spelling seen. That is not just a saving:
 * Postgres refuses an ON CONFLICT DO UPDATE that would touch the same row twice in one
 * statement, so a duplicate inside a single VALUES list is an error, not a no-op. The
 * targeted second pass below queries by `#d` and by `#u` and routinely returns both.
 *
 * `newReviews` is a health signal — is discovery finding anything? — so it counts rows
 * that did not exist, not rows written. `changes` would count every re-seen event and
 * hide a stalled crawl behind a busy-looking number.
 */
async function ingestReviews(events: Iterable<NostrEvent>, index: MintIndex): Promise<number> {
  const rows = new Map<string, ReviewRow>();

  for (const e of events) {
    /*
     * The `k` tag is what marks a 38000 as being about a Cashu mint rather than a
     * federation or something this site does not list. Events without it but with a
     * resolvable mint URL are still accepted: the old publisher's own events are the
     * reason, and every one of them is about a mint.
     *
     * A federation review has already proved itself by resolving: `resolveReviewTarget`
     * only returns a federation row for an event whose `k` says 38173, so nothing here
     * has to re-test that.
     */
    const target = resolveReviewTarget(e, index);
    if (!target) continue;
    /*
     * A review whose `k` names an ecosystem this build knows has already proved itself
     * by resolving: `resolveReviewTarget` runs one resolver per ecosystem and returns a
     * row only for an event whose `k` matches that resolver, so nothing here has to
     * re-test it. Written against `ANNOUNCEMENT_KINDS` rather than against a list of
     * kind numbers, so a fourth ecosystem needs no edit here either.
     *
     * The `u` clause is what keeps the legacy events: a great many kind 38000 on the
     * network carry no `k` at all, every one of them is about a Cashu mint, and they are
     * accepted on having a resolvable mint URL exactly as they always were.
     */
    const kindKnown = reviewTargetKind(e) !== null && reviewEcosystem(e) !== null;
    if (!kindKnown && mintUrlsFromEvent(e).length === 0) continue;

    const k = reviewTargetKind(e);

    rows.set(e.id, {
      eventId: e.id,
      mintUrl: target,
      pubkey: e.pubkey,
      rating: parseRating(e),
      k: k === null ? null : String(k),
      createdAt: e.created_at,
    });
  }

  if (rows.size === 0) return 0;

  const batch = [...rows.values()];
  const known = await knownEventIds(batch.map((r) => r.eventId));
  const db = await getDb();

  await db.transaction(async (tx: Sql) => {
    for (let i = 0; i < batch.length; i += INSERT_BATCH) {
      const chunk = batch.slice(i, i + INSERT_BATCH);
      await tx.run(
        `INSERT INTO reviews (event_id, mint_url, pubkey, rating, k, created_at)
         VALUES ${chunk.map(() => '(?, ?, ?, ?, ?, ?)').join(', ')}
         ON CONFLICT (event_id) DO UPDATE SET
           mint_url = excluded.mint_url,
           rating = excluded.rating,
           k = excluded.k`,
        ...chunk.flatMap((r) => [r.eventId, r.mintUrl, r.pubkey, r.rating, r.k, r.createdAt]),
      );
    }
  });

  return batch.filter((r) => !known.has(r.eventId)).length;
}

export async function runDiscovery(backfill: boolean): Promise<DiscoveryResult> {
  const started = Date.now();
  const now = Math.floor(Date.now() / 1000);
  const lastRun = backfill ? null : await getStateNumber('discovery_since');
  // Overlap the window by an hour so an event that arrived late is not missed.
  const since = lastRun === null ? null : Math.max(0, lastRun - 3600);

  const newMints = new Set<string>();
  const tally = new RelayTally(config.relays);
  let newReviews = 0;
  let events = 0;
  let ok = true;

  try {
    /*
     * One query per announcement kind, driven by ANNOUNCEMENT_KINDS rather than by a
     * list written out here: adding an ecosystem is a line in shared/, not an edit in
     * this loop. They run together because they are independent reads against the same
     * relay pool, and running them in series doubled the cycle's wall clock for nothing.
     */
    const announcementsByType = new Map<string, NostrEvent[]>();
    await Promise.all(
      Object.entries(ANNOUNCEMENT_KINDS).map(async ([type, kind]) => {
        announcementsByType.set(type, await fetchKind(kind, since, tally));
      }),
    );

    const announcements = announcementsByType.get('cashu') ?? [];
    const federations = announcementsByType.get('fedimint') ?? [];
    const lnurlMints = announcementsByType.get('lnurl') ?? [];
    events += announcements.length + federations.length + lnurlMints.length;

    await ingestMintUrls(announcements, now, newMints);
    await ingestFedimints(federations, now, newMints);
    await ingestLnurl(lnurlMints, now, newMints);

    // Announcements alone miss mints that only ever appear in a review's `u` tag,
    // so reviews feed discovery too.
    const reviews = await fetchKind(KIND_REVIEW, since, tally);
    // The recent window catches anything a relay dropped from the unbounded query.
    const recent =
      since === null ? await fetchKind(KIND_REVIEW, now - RECENT_WINDOW_S, tally) : [];

    const byId = new Map<string, NostrEvent>();
    for (const e of [...reviews, ...recent]) byId.set(e.id, e);
    events += byId.size;

    await ingestMintUrls(byId.values(), now, newMints);

    // Built after every mint this cycle found has been inserted, so review resolution
    // below sees them. Nothing after this point inserts a mint.
    const index = await loadMintIndex();
    newReviews += await ingestReviews(byId.values(), index);

    // Second pass: ask each known mint's reviews by name, which is the only way the
    // counts converge (see fetchReviewsForMint). Runs after the global sweep so mints
    // discovered in this cycle are included.
    const db = await getDb();
    const rows = await db.all<{
      url: string;
      type: string;
      pubkey: string | null;
      ecosystem_json: string | null;
    }>('SELECT url, type, pubkey, ecosystem_json FROM mints');

    const targets: ReviewTarget[] = rows.map((row) => {
      let federationId: string | null = null;
      let identifiers: string[] = [];
      let baseUrl: string | null = null;

      if (row.ecosystem_json) {
        try {
          if (row.type === 'fedimint') {
            federationId =
              (JSON.parse(row.ecosystem_json) as { federation_id?: string }).federation_id ?? null;
          } else if (row.type === 'lnurl') {
            const fields = JSON.parse(row.ecosystem_json) as Partial<LnurlFields>;
            baseUrl = fields.base_url ?? null;
            // Deduped, because a mint with no funding source has `lnurl_id` and
            // `mint_pubkey` describing the same thing and a relay filter listing one
            // value twice is one wasted slot.
            identifiers = [...new Set(
              [fields.lnurl_id, fields.mint_pubkey]
                .filter((id): id is string => Boolean(id))
                .map((id) => id.toLowerCase()),
            )];
          }
        } catch {
          // Nothing to ask by. The global sweep above already had its chance.
        }
      }

      return { type: row.type, url: row.url, pubkey: row.pubkey, federationId, identifiers, baseUrl };
    });

    const CONCURRENCY = 6;
    let cursor = 0;
    await Promise.all(
      Array.from({ length: Math.min(CONCURRENCY, targets.length) }, async () => {
        while (cursor < targets.length) {
          const target = targets[cursor++];
          if (!target) continue;
          const found = await fetchReviewsForMint(target, since, tally);
          if (found.length > 0) {
            events += found.length;
            newReviews += await ingestReviews(found, index);
          }
        }
      }),
    );

    await setState('discovery_since', String(now));
    await setState('last_discovery_at', String(now));
    await setState('last_discovery_ok', '1');
  } catch (err) {
    ok = false;
    await setState('last_discovery_ok', '0').catch(() => undefined);
    log.error('discovery failed', { reason: err instanceof Error ? err.message : String(err) });
  }

  const mode = backfill ? 'backfill' : 'incremental';
  const relays = tally.list();

  /*
   * Name the relay, every time, one line each.
   *
   * A relay that would not connect is worth saying on any cycle: the address is wrong,
   * or it is down, and neither gets better by itself. A relay that connected and sent
   * nothing is only news on a backfill — an incremental cycle asking for the last hour
   * of four kinds legitimately comes back empty, and warning about that hourly would
   * train everyone to skip the line that eventually matters.
   */
  for (const relay of relays) {
    if (!relay.connected) {
      log.warn('discovery relay unreachable', { relay: relay.url, mode });
      continue;
    }
    if (backfill && relay.events === 0) {
      log.warn('discovery relay returned no events', { relay: relay.url, mode });
    } else if (!relay.eose) {
      log.warn('discovery relay never reached EOSE', {
        relay: relay.url,
        mode,
        events: relay.events,
      });
    }
  }

  /*
   * The floor, and the flag the health endpoint reads.
   *
   * Only a backfill is measured against it. A backfill asks for the entire history of
   * every announcement kind and every review, so on a working relay set it is thousands
   * of events; an incremental cycle asks for one interval and is supposed to be small.
   *
   * The flag is sticky across incremental cycles: an hourly cycle that finds four
   * events must not clear a starvation a backfill diagnosed, so a non-backfill carries
   * forward whatever the last backfill concluded.
   */
  let starved: boolean;
  if (backfill) {
    starved = events < config.backfillMinEvents;
    if (starved) {
      log.error('discovery starvation suspected', {
        events,
        floor: config.backfillMinEvents,
        relays: relays.length,
        silent: relays.filter((r) => r.events === 0).length,
        unreachable: relays.filter((r) => !r.connected).length,
        hint: 'check RELAYS: a relay list missing the announcement archive looks exactly like this',
      });
    }
  } else {
    // No backfill has ever run in this deployment: nothing has confirmed the relay set
    // reads anything, and saying "fine" would be the whole original bug.
    starved = (await lastDiscoveryReport())?.starved ?? true;
  }

  const report: DiscoveryReport = { at: now, mode, events, ok, relays, starved };
  // A report that cannot be written is not worth failing a cycle over; the cycle's own
  // work is already committed, and health degrades on the stale timestamp instead.
  await setState(REPORT_KEY, JSON.stringify(report)).catch(() => undefined);

  log.info('discovery cycle', {
    mode,
    events,
    new_mints: newMints.size,
    new_reviews: newReviews,
    ok,
    starved,
    relays: relays.map((r) => `${r.url}=${r.connected ? r.events : 'down'}`).join(' '),
    ms: Date.now() - started,
  });

  return { events, newMints: [...newMints], newReviews, ok, relays, starved };
}