spb/coinexplorer Public MIT

1#!/usr/bin/env python32# Author: Simon-Pierre Boucher3# Mail: contact@spboucher.ai4"""Run the indexer: one worker thread per chain + price/supply workers,5all under a watchdog that restarts any thread that dies.67    python3 run_indexer.py                        # all configured chains8    python3 run_indexer.py --chains ethereum base # a subset9"""1011import argparse12import logging13import threading1415from indexer import config, db16from indexer.adapters.bitcoin import BitcoinIndexer17from indexer.adapters.solana import SolanaIndexer18from indexer.adapters.tron import TronIndexer19from indexer.enrich import PriceWorker, StatsWorker, SupplyWorker20from indexer.ingest import ChainIndexer2122ADAPTERS = {23    "evm": ChainIndexer,24    "tron": TronIndexer,25    "solana": SolanaIndexer,26    "bitcoin": BitcoinIndexer,27}282930def main():31    ap = argparse.ArgumentParser()32    ap.add_argument("--chains", nargs="*", help="subset of chains (default: all)")33    ap.add_argument("--db", default=config.db_path())34    args = ap.parse_args()3536    logging.basicConfig(37        level=logging.INFO,38        format="%(asctime)s %(levelname)-7s %(message)s",39        datefmt="%H:%M:%S",40    )4142    chains = config.load_chains()43    tokens = config.load_tokens()44    selected = config.selected_chains(chains, tokens, args.chains, set(ADAPTERS))4546    conn = db.connect(args.db)  # main-thread conn: schema + token metadata only47    for chain in selected:48        db.upsert_chain(conn, chain, chains[chain].get("family", "evm"),49                        chains[chain].get("chain_id"))50        db.upsert_tokens(conn, chain, tokens[chain])51    conn.close()5253    stop = threading.Event()5455    # every worker is a (name, factory) pair so the watchdog can rebuild it —56    # workers keep their own retry loops; this is the belt AND the suspenders57    def chain_factory(c):58        return lambda: ADAPTERS[chains[c].get("family", "evm")](59            c, chains[c], tokens[c], args.db).run(stop)6061    factories = {c: chain_factory(c) for c in selected}62    factories["prices"] = lambda: PriceWorker(63        {c: tokens[c] for c in selected}, args.db).run(stop)64    factories["supply"] = lambda: SupplyWorker(65        chains, {c: tokens[c] for c in selected}, args.db).run(stop)66    factories["stats"] = lambda: StatsWorker(args.db).run(stop)6768    def spawn(name):69        t = threading.Thread(target=guarded(name), name=name, daemon=True)70        t.start()71        return t7273    def guarded(name):74        def inner():75            try:76                factories[name]()77            except Exception as e:  # last-resort: workers shouldn't get here78                logging.error("worker %s died: %s", name, e)79        return inner8081    threads = {name: spawn(name) for name in factories}82    logging.info("started %d chain workers + prices + supply (watchdog on)",83                 len(selected))8485    try:86        while not stop.is_set():87            stop.wait(30)88            for name, t in list(threads.items()):89                if not t.is_alive() and not stop.is_set():90                    logging.warning("watchdog: restarting dead worker %s", name)91                    threads[name] = spawn(name)92    except KeyboardInterrupt:93        logging.info("stopping...")94        stop.set()95        for t in threads.values():96            t.join(timeout=10)979899if __name__ == "__main__":100    main()101