Files
zeekayandHanzo Dev 2d8e8c756e A subgraph's health reports the store's cursor, not this process's
Status() counts what the running indexer has ingested since it started. On a
chain already at its head a restart ingests nothing, so /v1/graph/{slug}/{sub}/
health answered {"block":0,"indexed":0} with every event still in the store —
measured on cchain, which reported block 19 before a restart and block 0 after,
while the same subgraph's GraphQL kept answering with its token.

That is the one reply a reader checks this endpoint to avoid. Read `block` from
store.GetLastBlock(), which is the cursor that persists, and keep the process
count where it is higher — a fresh store mid-first-pass has moved further than
it has committed. `indexed` stays the process count: nothing persists an event
total, and a second number meaning something else under the same name is how
this went wrong once.

Co-authored-by: Hanzo Dev <[email protected]>
2026-09-08 19:55:00 -07:00

671 lines
23 KiB
Go

package main
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"path/filepath"
"strings"
"sync"
"time"
// ONE sqlite C library in this binary, under a name nothing else claims.
// This process links graph and indexer as well as its own store, and
// csqlite is the house build of the amalgamation all three now open —
// registered as "sqlite3". Upstream mattn compiles a second copy of the
// same C, and defining every sqlite3_* symbol twice fails at link on
// darwin. Not its hanzoai/sqlite wrapper either: that registers
// "sqlite", and so does modernc.org/sqlite, which arrives with
// hanzoai/replicate — two packages under one name is a panic in init.
_ "github.com/hanzoai/csqlite"
"github.com/luxfi/graph/engine"
graphidx "github.com/luxfi/graph/indexer"
graphstor "github.com/luxfi/graph/storage"
"github.com/luxfi/indexer/chain"
"github.com/luxfi/indexer/evm"
"github.com/luxfi/indexer/explorer"
"github.com/luxfi/indexer/platform"
idxstor "github.com/luxfi/indexer/storage"
)
// ChainSupervisor owns the per-chain indexer + graph goroutines plus the
// route dispatcher that maps /v1/indexer/{slug}/* and /v1/graph/{slug}/*
// requests to the correct handler. Routes are mutable at runtime: adding a
// chain spawns its workers, removing a chain cancels them.
type ChainSupervisor struct {
dataDir string
cfg Config // resolves a request host to its chain; see Config.Serves
mu sync.Mutex
state map[string]*chainState
indexerRoutes sync.Map // map[slug]http.Handler
graphRoutes sync.Map // map[slug + "/" + subgraph]http.Handler
swapRoutes sync.Map // map[chainID]*swapMount — the swap form's routes, by the chain they name
defaultSlug string // serves /v1/indexer/* when the host names no chain
// defaultChainID answers a swap request that names no chain. A form that
// has not finished loading its network sends one, and the alternative to a
// default is an error on the first render.
defaultChainID int64
// realtimeHub is the explorer's RealtimeHub. When set (via
// AttachRealtime), the supervisor wires each per-chain EVM indexer's
// Subscriber.OnBroadcast callback so block events flow into the hub
// → out to WebSocket + SSE subscribers. Nil = no live broadcast.
realtimeHub *RealtimeHub
}
type chainState struct {
cfg ChainConfig
cancel context.CancelFunc
wg sync.WaitGroup
}
// NewChainSupervisor creates a supervisor rooted at cfg.DataDir.
func NewChainSupervisor(cfg Config) *ChainSupervisor {
return &ChainSupervisor{
dataDir: cfg.DataDir,
cfg: cfg,
state: make(map[string]*chainState),
}
}
// AttachRealtime wires per-chain block events into the explorer's
// RealtimeHub. Called once from main.go after both the supervisor and
// the registry's hub exist. Subsequent chain spawns (runChain below)
// install the Subscriber.OnBroadcast callback automatically.
func (s *ChainSupervisor) AttachRealtime(h *RealtimeHub) {
s.realtimeHub = h
}
// MountRoutes installs the /v1/indexer and /v1/graph dispatchers on mux. The
// dispatchers look up per-chain handlers from the supervisor's route maps so
// chains added at runtime become reachable without re-mounting.
func (s *ChainSupervisor) MountRoutes(mux *http.ServeMux) {
mux.Handle("/v1/indexer/", http.HandlerFunc(s.dispatchIndexer))
mux.Handle("/v1/explorer/", http.HandlerFunc(s.dispatchExplorerLegacy))
mux.Handle("/v1/graph/", http.HandlerFunc(s.dispatchGraph))
}
// SetDefaultSlug picks which chain serves the unprefixed /v1/indexer/* route.
func (s *ChainSupervisor) SetDefaultSlug(slug string) {
s.mu.Lock()
s.defaultSlug = slug
s.mu.Unlock()
}
// start spawns workers for cfg. Idempotent: starting an already-running slug
// is a no-op.
func (s *ChainSupervisor) start(cfg ChainConfig) {
s.mu.Lock()
if _, ok := s.state[cfg.Slug]; ok {
s.mu.Unlock()
return
}
ctx, cancel := context.WithCancel(context.Background())
st := &chainState{cfg: cfg, cancel: cancel}
s.state[cfg.Slug] = st
if cfg.Default && s.defaultSlug == "" {
s.defaultSlug = cfg.Slug
}
s.mu.Unlock()
st.wg.Add(1)
go func() {
defer st.wg.Done()
s.runChain(ctx, cfg)
}()
if cfg.Graph.Enabled {
for _, sg := range cfg.Graph.Subgraphs {
if !sg.Enabled {
continue
}
sg := sg
st.wg.Add(1)
go func() {
defer st.wg.Done()
s.runSubgraph(ctx, cfg, sg)
}()
}
}
}
// stop cancels and waits for all workers belonging to slug.
func (s *ChainSupervisor) stop(slug string) {
s.mu.Lock()
st, ok := s.state[slug]
if !ok {
s.mu.Unlock()
return
}
delete(s.state, slug)
if s.defaultSlug == slug {
s.defaultSlug = ""
}
s.mu.Unlock()
st.cancel()
st.wg.Wait()
s.indexerRoutes.Delete(slug)
s.graphRoutes.Range(func(k, _ any) bool {
if key, ok := k.(string); ok && strings.HasPrefix(key, slug+"/") {
s.graphRoutes.Delete(k)
}
return true
})
log.Printf("[supervisor] stopped %s", slug)
}
// Wait blocks until ctx is done, then cancels every chain and waits for
// goroutines to drain.
func (s *ChainSupervisor) Wait(ctx context.Context) {
<-ctx.Done()
s.mu.Lock()
slugs := make([]string, 0, len(s.state))
for k := range s.state {
slugs = append(slugs, k)
}
s.mu.Unlock()
for _, slug := range slugs {
s.stop(slug)
}
}
// runChain runs one chain's indexer to completion or context cancel and then
// mounts the standalone explorer API once the SQLite schema is ready.
func (s *ChainSupervisor) runChain(ctx context.Context, cfg ChainConfig) {
chainDir := filepath.Join(s.dataDir, cfg.Slug)
if err := os.MkdirAll(chainDir, 0755); err != nil {
log.Printf("[%s] mkdir: %v", cfg.Slug, err)
return
}
store, err := idxstor.NewUnified(idxstor.DefaultUnifiedConfig(chainDir))
if err != nil {
log.Printf("[%s] storage: %v", cfg.Slug, err)
return
}
defer store.Close()
if err := store.Init(ctx); err != nil {
log.Printf("[%s] storage init: %v", cfg.Slug, err)
return
}
dbPath := filepath.Join(chainDir, "query", "indexer.db")
log.Printf("[%s] indexing %s (%s) -> %s", cfg.Slug, cfg.Name, cfg.CoinSymbol, dbPath)
go s.mountIndexerAPI(ctx, cfg, dbPath)
switch cfg.Type {
case "", "evm":
poll := 30 * time.Second
if cfg.Indexer.PollInterval != "" {
if d, err := time.ParseDuration(cfg.Indexer.PollInterval); err == nil {
poll = d
}
}
idx, err := evm.NewIndexer(evm.Config{
ChainName: cfg.Name,
ChainID: cfg.ChainID,
RPCEndpoint: cfg.RPC,
HTTPPort: 0,
PollInterval: poll,
StartBlock: cfg.Indexer.StartBlock,
}, store)
if err != nil {
log.Printf("[%s] evm indexer: %v", cfg.Slug, err)
return
}
// Live broadcast bridge — when the supervisor has been given a
// RealtimeHub (see AttachRealtime), every block the indexer
// commits is re-published into the hub on the chain's slug. The
// hub then fans out to WebSocket and SSE subscribers (see
// realtime.go + realtime_sse.go). slug captured by value so the
// closure doesn't see a mutated `cfg` if the supervisor loops.
if s.realtimeHub != nil && idx.Subscriber() != nil {
slug := cfg.Slug
hub := s.realtimeHub
idx.Subscriber().OnBroadcast = func(eventType string, data any) {
hub.Broadcast(eventType, slug, data)
}
}
if err := idx.Init(ctx); err != nil {
log.Printf("[%s] evm init: %v", cfg.Slug, err)
return
}
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Printf("[%s] evm run: %v", cfg.Slug, err)
}
case "linear", "platform":
// P-Chain (platform VM) indexing. The generic chain indexer supplies
// the canonical pchain_blocks schema; the platform adapter supplies the
// validator/delegator/etc schema and the RPC block/validator fetchers.
poll := 30 * time.Second
if cfg.Indexer.PollInterval != "" {
if d, err := time.ParseDuration(cfg.Indexer.PollInterval); err == nil {
poll = d
}
}
adapter := platform.New(cfg.RPC)
chainCfg := platform.NewConfig() // ChainType=pchain -> pchain_blocks table
chainCfg.ChainName = cfg.Name
chainCfg.RPCEndpoint = cfg.RPC
chainCfg.PollInterval = poll
idx, err := chain.New(chainCfg, store, adapter)
if err != nil {
log.Printf("[%s] platform indexer: %v", cfg.Slug, err)
return
}
// Init once: creates pchain_blocks (generic schema) and the adapter's
// pchain_validators/pchain_delegators/... tables. We drive our own poll
// loop rather than idx.Run() so we don't stand up the chain indexer's
// WS-gated poller + HTTP server (the standalone API reads the SQLite DB
// directly). Blocks are written via store.StoreBlock, not idx.StoreBlock,
// to avoid the indexer's subscriber broadcast (nothing drains it here).
if err := idx.Init(ctx); err != nil {
log.Printf("[%s] platform init: %v", cfg.Slug, err)
return
}
const platformBlocksTable = "pchain_blocks"
sync := func() {
raw, err := adapter.GetRecentBlocks(ctx, 50)
if err != nil {
if ctx.Err() == nil {
log.Printf("[%s] platform blocks: %v", cfg.Slug, err)
}
} else {
for _, rb := range raw {
b, perr := adapter.ParseBlock(rb)
if perr != nil {
continue
}
sb := &idxstor.Block{
ID: b.ID,
ParentID: b.ParentID,
Height: b.Height,
Timestamp: b.Timestamp,
Status: string(b.Status),
TxCount: b.TxCount,
TxIDs: b.TxIDs,
Data: b.Data,
CreatedAt: time.Now(),
}
if err := store.StoreBlock(ctx, platformBlocksTable, sb); err != nil && ctx.Err() == nil {
log.Printf("[%s] platform store block %d: %v", cfg.Slug, b.Height, err)
}
}
}
if err := adapter.SyncValidators(ctx, store); err != nil && ctx.Err() == nil {
log.Printf("[%s] platform validators: %v", cfg.Slug, err)
}
}
sync() // first pass immediately so tables populate before the first tick
ticker := time.NewTicker(poll)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
sync()
}
}
default:
log.Printf("[%s] %s indexer: passive (no chain-specific implementation)", cfg.Slug, cfg.Type)
<-ctx.Done()
}
}
// mountIndexerAPI waits for the per-chain SQLite schema to settle, then
// publishes the standalone explorer API in the indexer/explorer route maps.
func (s *ChainSupervisor) mountIndexerAPI(ctx context.Context, cfg ChainConfig, dbPath string) {
apiPrefix := "/v1/indexer/" + cfg.Slug
deadline := time.Now().Add(5 * time.Minute)
for {
select {
case <-ctx.Done():
return
case <-time.After(time.Second):
}
if _, err := os.Stat(dbPath); err != nil {
if time.Now().After(deadline) {
log.Printf("[%s] api waiting for %s: %v", cfg.Slug, dbPath, err)
deadline = time.Now().Add(5 * time.Minute)
}
continue
}
// Gate readiness on the primary blocks table for this chain type:
// EVM writes evm_blocks; platform/linear writes pchain_blocks (created
// by the generic chain indexer's Init — NOT by platform.InitSchema,
// which only creates the validator/delegator/etc tables).
gateTable := ""
switch cfg.Type {
case "", "evm":
gateTable = "evm_blocks"
case "platform", "linear":
gateTable = "pchain_blocks"
}
if gateTable != "" {
db, err := sql.Open("sqlite3", "file:"+dbPath+"?mode=ro")
if err == nil {
var n int
_ = db.QueryRow("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?", gateTable).Scan(&n)
db.Close()
if n == 0 {
if time.Now().After(deadline) {
log.Printf("[%s] api waiting for %s", cfg.Slug, gateTable)
deadline = time.Now().Add(5 * time.Minute)
}
continue
}
}
}
srv, err := explorer.NewStandaloneServer(explorer.Config{
IndexerDBPath: dbPath,
ChainID: cfg.ChainID,
ChainName: cfg.Name,
CoinSymbol: cfg.CoinSymbol,
APIPrefix: apiPrefix,
// Native-coin balances come from the chain, not from the
// index — the same RPC this chain is indexed from.
RPCEndpoint: cfg.RPC,
})
if err != nil {
if time.Now().After(deadline) {
log.Printf("[%s] api: %v", cfg.Slug, err)
deadline = time.Now().Add(5 * time.Minute)
}
continue
}
s.indexerRoutes.Store(cfg.Slug, srv.Handler())
log.Printf("[%s] api mounted at %s/*", cfg.Slug, apiPrefix)
return
}
}
// runSubgraph spawns a graph indexer + engine for a single subgraph and
// publishes its handler at /v1/graph/{chain-slug}/{subgraph-name}/.
func (s *ChainSupervisor) runSubgraph(ctx context.Context, cfg ChainConfig, sg Subgraph) {
dir := filepath.Join(s.dataDir, cfg.Slug, "graph", sg.Name)
if err := os.MkdirAll(dir, 0755); err != nil {
log.Printf("[%s/%s] mkdir: %v", cfg.Slug, sg.Name, err)
return
}
store, err := graphstor.New(dir)
if err != nil {
log.Printf("[%s/%s] storage: %v", cfg.Slug, sg.Name, err)
return
}
defer store.Close()
if err := store.Init(ctx); err != nil {
log.Printf("[%s/%s] storage init: %v", cfg.Slug, sg.Name, err)
return
}
idx := graphidx.NewWithConfig(graphidx.Config{
RPC: cfg.RPC,
PoolManager: cfg.PoolManager, // empty => indexer defaults to 0x9999
FactoryV2: cfg.FactoryV2,
FactoryV3: cfg.FactoryV3,
Native: cfg.Native,
// What the token is actually worth: minted at genesis, less what the
// treasury still holds and what is bonded to validators.
GenesisSupply: cfg.GenesisSupply,
Treasury: cfg.Treasury,
StartBlock: cfg.Indexer.StartBlock,
// Every other line this supervisor emits is prefixed "[slug/name]"; the
// graph indexer's were not, so in a six-chain process there was no way to
// tell which chain a line came from — or which chain had gone quiet.
Label: cfg.Slug + "/" + sg.Name,
}, store)
// What a start is for: ask the chain about the Token rows still short an
// answer — a symbol an earlier read never got, a supply a build that never
// asked for one left empty. Before Run, so live indexing is not racing the
// same writes. A row with everything on it costs nothing, which is why this
// is simply what a start does rather than something to remember to turn on.
if n, err := idx.BackfillTokens(ctx); err != nil {
log.Printf("[%s/%s] token backfill: %v (settled=%d)", cfg.Slug, sg.Name, err, n)
} else {
log.Printf("[%s/%s] token backfill: settled=%d", cfg.Slug, sg.Name, n)
}
go func() {
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
log.Printf("[%s/%s] indexer: %v", cfg.Slug, sg.Name, err)
}
}()
eng := engine.New(store, &engine.Config{
MaxQueryDepth: 10,
MaxResultSize: 1 << 20,
QueryTimeoutMs: 30000,
})
schema := sg.Schema
if schema == "" {
schema = "amm"
}
if err := eng.LoadBuiltin(schema); err != nil {
log.Printf("[%s/%s] schema %q: %v", cfg.Slug, sg.Name, schema, err)
return
}
prefix := fmt.Sprintf("/v1/graph/%s/%s", cfg.Slug, sg.Name)
mux := http.NewServeMux()
gqlHandler := func(w http.ResponseWriter, r *http.Request) {
r.Body = http.MaxBytesReader(w, r.Body, 10<<20)
var req engine.Request
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, `{"errors":[{"message":"invalid JSON"}]}`, http.StatusBadRequest)
return
}
resp := eng.Execute(r.Context(), &req)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
}
// `block` is the cursor the STORE holds, not the one this process has moved.
// Status() counts what the running indexer has ingested since it started, so
// a restart on a chain already at its head reports `block:0, indexed:0` while
// the store holds every event — which reads as an empty index and is the
// reply a reader checks this endpoint to avoid. `indexed` stays the process
// count: nothing persists an event total, and a second number that means
// something else under the same name is how this went wrong once.
statusHandler := func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
st := idx.Status()
block := store.GetLastBlock()
if st.LatestBlock > block {
block = st.LatestBlock
}
fmt.Fprintf(w, `{"status":"ok","block":%d,"indexed":%d}`, block, st.IndexedEvents)
}
mux.HandleFunc("POST "+prefix+"/graphql", gqlHandler)
mux.HandleFunc("GET "+prefix+"/graphql", eng.HandleGraphiQL)
mux.HandleFunc("POST "+prefix+"/ql", gqlHandler)
mux.HandleFunc("GET "+prefix+"/ql", eng.HandleGraphiQL)
mux.HandleFunc("GET "+prefix+"/health", statusHandler)
// The swap form asks four questions the GraphQL schema does not answer in a
// shape a form can use: what can I trade, what do I get, may the router
// spend it, and what do I sign. All four read the same store this subgraph
// already indexes, so they belong beside it rather than behind a second
// service with a second copy of the book.
quoter := engine.NewQuoter(store, cfg.ChainID, cfg.RPC, cfg.QuoterV2, cfg.Native)
mux.HandleFunc("POST "+prefix+"/quote", engine.HandleQuote(quoter))
mux.HandleFunc("GET "+prefix+"/swappable_tokens", engine.HandleSwappableTokens(store, cfg.ChainID, cfg.CoinSymbol, cfg.Native))
// A swap is a transaction addressed to this chain's router, and an approval
// grants that same router the right to spend. Both are rendered into a
// twenty-byte argument, as is the wrapped native a route substitutes for
// the coin, so anything else here produces a call whose arguments have slid
// — signed against an address nobody configured. Without both the routes
// are simply absent, which a caller can see.
if canSwap(cfg) {
mux.HandleFunc("POST "+prefix+"/swap", engine.HandleSwap(cfg.ChainID, cfg.Router, cfg.Native))
mux.HandleFunc("POST "+prefix+"/check_approval", engine.HandleCheckApproval(cfg.ChainID, cfg.RPC, cfg.Router))
} else {
log.Printf("[%s/%s] router=%q native=%q — both must be addresses; /swap and /check_approval not served",
cfg.Slug, sg.Name, cfg.Router, cfg.Native)
}
s.graphRoutes.Store(cfg.Slug+"/"+sg.Name, http.Handler(mux))
// The same mux answers the unprefixed /v1/{quote,swap,…} routes for this
// chain. A swap request already carries the chain it is for, so naming it
// again in the path is a second place for the two to disagree. Registering
// the handler VALUES here rather than building a second set is what keeps
// the prefixed and unprefixed routes one implementation.
//
// A chain may declare more than one subgraph holding a book, and each runs
// in its own goroutine, so "whichever registered last" would pick a
// different book across restarts. The one named in the config first wins,
// decided from the config rather than from the scheduler.
if canSwap(cfg) && sg.Name == firstAMM(cfg) {
s.swapRoutes.Store(cfg.ChainID, &swapMount{prefix: prefix, h: mux})
if cfg.Default {
s.mu.Lock()
s.defaultChainID = cfg.ChainID
s.mu.Unlock()
}
}
log.Printf("[%s/%s] graph mounted at %s/{graphql,ql,health,quote,swappable_tokens,swap,check_approval} (schema=%s)",
cfg.Slug, sg.Name, prefix, schema)
<-ctx.Done()
}
// dispatchIndexer routes /v1/indexer/{slug}/* and /v1/indexer/* (default
// chain) to the per-chain handler installed by mountIndexerAPI.
//
// Edge case the older code got wrong: when the first path segment matches
// the slug regex but is NOT a registered chain (e.g. `main-page` in
// `/v1/indexer/main-page/blocks`), splitSlug peels it off as a "slug" and
// the dispatcher then forwarded only the REMAINING segments to the
// default chain. That turned `/v1/indexer/main-page/blocks` into
// `/v1/indexer/evm/blocks` server-side — a completely different endpoint
// with a different response shape (paginated envelope vs. raw array). The
// SPA's `?.slice(0,8)` on the home page's Latest Blocks widget tripped
// over the envelope and threw "o.slice is not a function".
//
// Fix: when the parsed slug isn't a real chain, re-join it with the rest
// before delegating, so the unrecognized segment is preserved as part of
// the path under the default chain.
func (s *ChainSupervisor) dispatchIndexer(w http.ResponseWriter, r *http.Request) {
slug, rest, ok := splitSlug(r.URL.Path, "/v1/indexer/")
if !ok {
s.serveHost(w, r, "")
return
}
if h, found := s.indexerRoutes.Load(slug); found {
h.(http.Handler).ServeHTTP(w, r)
return
}
fullTail := slug
if rest != "" {
fullTail = slug + "/" + rest
}
s.serveHost(w, r, fullTail)
}
// serveHost forwards a slug-less /v1/indexer/* request to the chain the
// request host is served. The standalone server is mounted under its slug, so
// the URL is rewritten to include the slug before delegating.
//
// Pinning this to the configured default instead of the host answered Lux's
// blocks, addresses and tokens on api-explore.hanzo.network and
// api-explore.zoo.network — the same window of Lux heights, to the block, on
// all three brands.
func (s *ChainSupervisor) serveHost(w http.ResponseWriter, r *http.Request, rest string) {
slug := s.hostSlug(r.Host)
if slug == "" {
http.Error(w, `{"error":"no chain for host"}`, http.StatusNotFound)
return
}
h, ok := s.indexerRoutes.Load(slug)
if !ok {
http.Error(w, `{"error":"chain not ready"}`, http.StatusServiceUnavailable)
return
}
r2 := r.Clone(r.Context())
r2.URL.Path = "/v1/indexer/" + slug + "/" + strings.TrimPrefix(rest, "/")
h.(http.Handler).ServeHTTP(w, r2)
}
// hostSlug is the chain a slug-less request is answered from: the chain the
// host names, when that chain is running, else the default. Chains can be
// added and removed at runtime, so a host naming a chain that is not up falls
// back rather than 404ing.
func (s *ChainSupervisor) hostSlug(host string) string {
if c, ok := s.cfg.Chain(host); ok {
if _, running := s.indexerRoutes.Load(c.Slug); running {
return c.Slug
}
}
s.mu.Lock()
defer s.mu.Unlock()
return s.defaultSlug
}
// dispatchExplorerLegacy preserves the older /v1/explorer/{slug}/* prefix
// for clients that haven't migrated to /v1/indexer/.
func (s *ChainSupervisor) dispatchExplorerLegacy(w http.ResponseWriter, r *http.Request) {
slug, rest, ok := splitSlug(r.URL.Path, "/v1/explorer/")
if !ok {
http.NotFound(w, r)
return
}
h, found := s.indexerRoutes.Load(slug)
if !found {
http.NotFound(w, r)
return
}
r2 := r.Clone(r.Context())
r2.URL.Path = "/v1/indexer/" + slug + "/" + strings.TrimPrefix(rest, "/")
h.(http.Handler).ServeHTTP(w, r2)
}
// dispatchGraph routes /v1/graph/{slug}/{subgraph}/* to the matching engine.
func (s *ChainSupervisor) dispatchGraph(w http.ResponseWriter, r *http.Request) {
rest := strings.TrimPrefix(r.URL.Path, "/v1/graph/")
if rest == r.URL.Path {
http.NotFound(w, r)
return
}
parts := strings.SplitN(rest, "/", 3)
if len(parts) < 2 {
http.NotFound(w, r)
return
}
key := parts[0] + "/" + parts[1]
h, ok := s.graphRoutes.Load(key)
if !ok {
http.NotFound(w, r)
return
}
h.(http.Handler).ServeHTTP(w, r)
}
// splitSlug returns the chain slug and the remainder of the path after a
// known prefix (e.g. "/v1/indexer/"). Returns ok=false if the path does not
// match the prefix or contains no slug.
func splitSlug(path, prefix string) (slug, rest string, ok bool) {
if !strings.HasPrefix(path, prefix) {
return "", "", false
}
tail := strings.TrimPrefix(path, prefix)
if tail == "" {
return "", "", false
}
parts := strings.SplitN(tail, "/", 2)
if !slugPattern.MatchString(parts[0]) {
return "", "", false
}
if len(parts) == 1 {
return parts[0], "", true
}
return parts[0], parts[1], true
}