# Connector architecture

# 1. Layers

                 ┌────────────────────────── connectors/<engine>/<id> ──────────────────────────┐
  source ──────► │ crawl(ctx)  → RawRecordInput (source-shaped payload, snapshot, externalId)      │
                 │ normalize() → NormalizedRecord (sale · listing · auction_lot · price_observation │
                 │               · catalog_item · population_report · news_item)                    │
                 │ healthCheck(), lookup(url) (scanner / cert verification)                          │
                 └───────────────▲──────────────────────────────────────────────────────────────────┘
                                 │ SDK  (packages/connectors)
   ctx.fetch ── Router ── HostGates (concurrency · interval · circuit breaker, per host, from domains.json)
                  │  engines in priority order, each result scored by the connector's parser (quality ≥ 0.6)
                  ├─ api/feed  direct HTTP (honest UA, retries/backoff, conditional requests, binary)
                  ├─ firecrawl v2 scrape/search/map (markdown + rawHtml + JSON extraction)
                  └─ scrapfly  render_js + ASP + geo (public pages only)
   adapters: shopify · woocommerce · sitemap/robots · rss · pdf · schema.org · pagination
   budget store (crawl_state): content hash + change-interval estimate → shouldFetch()
                                 │
  workers/crawler/run.ts ── raw_records (immutable, content-hash dedupe, snapshots on disk), connector_runs,
                            connector_backfills (resumable campaigns), engine stats (statuses, blocked, latency)
  workers/normalizer      ── normalized_records + connector_field_stats (schema drift) + price-distribution anomaly
  workers/entity-resolution ─ assets / asset_variants (identifiers → canonical key → fuzzy) → sales · listings ·
                            auction_lots · price_observations · population_reports · certificates (+ sightings)
  workers/health          ── connector_health snapshots (UP/DEGRADED/BROKEN/DISABLED/MAINTENANCE)
  workers/certs-verify    ── cert_lookup connectors verify cert numbers seen on marketplaces

# 2. The connector contract (packages/connectors/src/types.ts)

interface RareIndexConnector {
  meta: ConnectorMeta;            // registry entry (id, source, engines, categories, geography, capabilities, refresh class, requires…)
  version; parserVersion;         // bump on crawl / mapping changes; stored on every raw row
  crawl(ctx): AsyncIterable<RawRecordInput>;      // resumable through ctx.options.cursor + ctx.setCursor / ctx.progress
  normalize(raw): Promise<NormalizedRecord[]>;    // pure, deterministic, fixture-tested
  healthCheck(ctx): Promise<ConnectorHealth>;     // default: derived from recent runs (+ light probe when fresh)
  lookup?(url, ctx): Promise<RawRecordInput[]>;   // single-URL resolution (scanner, cert verification)
  urlPatterns?: RegExp[];
}

SPEC §5 method names map onto this surface: discover/search/fetchListing/fetchProduct/fetchSoldListings/ fetchAuctionResults/fetchPopulationData/fetchCatalog are what crawl() does for the record kinds the connector declares in capabilities; refresh() is the scheduler; fetchSeller() is the seller/ sellerReputation fields (capability seller_data). Capabilities are explicit (meta.capabilities, derived from the supports* flags plus cert_lookup, price_guide, seller_data, historical_backfill, url_lookup).

# 3. Engines and routing (SPEC §10–11)

createRouter() tries the connector's enginePriority (narrowed by the domain policy) in order:

1. api/feed — createHttpEngine: honest RareIndexBot UA (browser UA only where domains.json says userAgent: "browser"), exponential backoff on 429/5xx, Retry-After, responseType json|text|binary.
2. firecrawl — /v2/scrape (markdown + rawHtml, optional JSON schema/prompt extraction, waitFor, location), /v2/search, /v2/map (sitemap-like discovery). Credits recorded in costs.
3. scrapfly — /scrape with render_js, asp, country, rendering_wait, extraction models.

Each result is scored: the connector's parse() returns which expected fields were found (weights §199) or, without a parser, a document heuristic. isChallengePage() detects Cloudflare/Akamai/PerimeterX/DataDome interstitials → counted as blocked and escalated. 404/410 never escalate. All attempts feed engineStats[engine] = { attempts, success, credits, ms, statuses{code:n}, blocked, circuitOpen } and the cost ledger (onAttempt). The strategy that succeeds per domain is visible in connector_runs.engine_stats and in health (firecrawl_success_rate, scrapfly_fallback_rate).

# 4. Per-domain policy and fault isolation (SPEC §16, §26)

connectors/domains.json (+ domains.d/*.json fragments) → policyFor(url): minIntervalMs, concurrency, timeoutMs, maxRetries, allowed engines, Firecrawl/Scrapfly options, country/locale/currency, crawlDepth, backfillMaxPages, userAgent, circuitFailures, circuitCooldownMs. Resolution: defaults → parent domain → exact host. HostGates (one per process, shared by all connectors) enforces the concurrency cap and interval per host and opens a circuit after N consecutive failures; while open, requests fail fast (circuitOpen counted), then a single half-open trial closes it again. Health probes bypass the gate.

# 5. Adapters (SPEC §1, §5)

packages/connectors/src/adapters/:

• ShopifyStoreConnector — /collections/<handle>/products.json (250/page), variants → listings, SKU/barcode → upc/ean/jan/isbn, grade/condition from titles, lookup() for /products/<handle>; config-only connectors.
• WooCommerceStoreConnector — Store API /wp-json/wc/store/v1/products, minor-unit prices with currency.
• discoverFromSitemap / parseRobots / robotsAllows — sitemap indexes, gz, lastmod filters, robots rules.
• parseFeed — RSS 2.0 / Atom. extractPdfPages/extractPdfText/parsePricesRealised — PDF price lists (pdfjs).
• productsFromHtml/primaryProduct/metaTags — schema.org Product/Offer JSON-LD and Open Graph.
• numberedPages/cursorPages/offsetPages/withParams — pagination loops (page numbers, cursors, offsets/infinite scroll).
• html.jsonLd/nextData/inlineJson — embedded JSON.

# 6. Storage (SPEC §24, §27)

All large tables carry date indexes suitable for range partitioning (sales(sale_date), raw_records(fetched_at), price_observations(observation_date)); ids are text prefixes so partitions can be added by migration without changing application code.

# 7. Scheduling (SPEC §17)

refreshFrequencyMinutes ⇄ refresh class: HOT ≤ 5 min · ACTIVE ≤ 60 · NORMAL ≤ 24 h · ARCHIVE weekly+. The scheduler tick (every minute) enqueues due incremental crawls (priority high/medium/low), continues running backfill campaigns, skips gated connectors (missing requires) and connectors that already have a queued or running crawl. Each run is time-boxed (RI_MAX_RUN_MINUTES, default 40) and resumes from its cursor.

# 8. Observability (SPEC §25)

Structured pino logs with component/connector fields; connector_run_finished events; per-run engine stats; cost ledger; health snapshots every 30 min; /api/admin/connectors and /api/admin/coverage JSON for external monitors (MacLustr apps). OpenTelemetry export can wrap the pino logger and the queue handlers without touching connectors.