Feature/contract metadata

README.md

@@ -136,6 +136,8 @@ All configuration comes from environment variables (see `.env.example`):
 | `RPC_URL` | `https://soroban-testnet.stellar.org` | Stellar RPC endpoint (JSON-RPC 2.0). Point at a provider URL for mainnet. |
 | `RPC_URLS` | unset | Comma-separated, priority-ordered list of Stellar RPC endpoints. When set, `RPC_URL` is ignored and the multi-provider failover client is used. List order is priority: index 0 is tried first. |
 | `RPC_RATE_LIMIT_RPS` | `10` | Per-provider request rate limit (`requests/second`) applied to each RPC endpoint independently. Only used when `RPC_URLS` is set. |
+| `HORIZON_URL` | `https://horizon-testnet.stellar.org` | Stellar Horizon REST endpoint used by `sorotrail backfill` only. Live ingestion does not touch Horizon. |
+| `BACKFILL_RATE_RPS` | `10` | Pace against Horizon when backfilling. 10 req/s is the public-instance cap; private deployments can lift this. |
 | `DATABASE_URL` | — (required) | Postgres connection string. |
 | `POLL_INTERVAL` | `5s` | Sleep between polls once caught up. |
 | `HTTP_ADDR` | `:8080` | API listen address. |
@@ -155,6 +157,7 @@ All configuration comes from environment variables (see `.env.example`):
 | `AUDIT_MAX_REPAIR_ATTEMPTS` | `3` | Repair iterations before a finding is kept open as `unrecoverable`. |
 | `AUDIT_FINDING_MAX_LEDGERS` | `100` | Largest range a single finding is allowed to span. |
 | `API_MAX_LIMIT` | `500` | Maximum page size accepted for list endpoints (`/events`, `/subscriptions/{id}/deliveries`). Values above this are rejected with 400. |
+| `API_KEY` | empty | Required to use the runtime `/watched-contracts` surface; empty means every request there is rejected with 503. This is a placeholder until #17 (real auth) lands — at that point `API_KEY` will be replaced. |
 | `RATE_LIMIT_RPS` | unset | Per-client HTTP request rate limit (`requests/second`). Both `RATE_LIMIT_RPS` and `RATE_LIMIT_BURST` must be set together; otherwise no rate limiting is applied. |
 | `RATE_LIMIT_BURST` | unset | Maximum instantaneous burst size for the rate limiter. Pairs with `RATE_LIMIT_RPS`. |
 | `RATE_LIMIT_TRUSTED_PROXY` | `false` | Honor `X-Forwarded-For` for client IP detection. Must only be enabled behind a proxy you trust to strip/rewrite the header — clients control `X-Forwarded-For` themselves, so enabling it on an Internet-facing surface lets any caller pick their own rate-limit key. |
@@ -643,6 +646,42 @@ Fetch a single event by its ID (the TOID-based identifier from the RPC).
 Shell
 
 curl -s localhost:8080/events/0001099511627776-0000000001
+### `GET /contracts`
+
+Lists every contract the indexer has seen, with cached token metadata
+(name, symbol, decimals) when available. Metadata is `null` for contracts
+that haven't been enriched yet or that don't implement the SEP-41 token
+interface.
+
+```sh
+curl -s localhost:8080/contracts
+```
+
+```json
+{
+  "contracts": [
+    {"contract_id":"CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC"},
+    {"contract_id":"CCW67...","name":"USD Coin","symbol":"USDC","decimals":6}
+  ]
+}
+```
+
+### `GET /contracts/{id}/stats`
+
+```sh
+curl -s localhost:8080/contracts/CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC/stats
+```
+
+```json
+{
+  "contract_id":"CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC",
+  "name":"USD Coin",
+  "symbol":"USDC",
+  "decimals":6,
+  "event_count":1423
+}
+```
+
 GET /contracts/{id}/events
 Convenience wrapper for GET /events?contract_id={id}; accepts the same
 remaining query parameters.
@@ -1396,6 +1435,45 @@ be defined with a foreign key to `events(id)` and `ON DELETE CASCADE`. The
 pruner therefore deletes only from `events` and lets Postgres cascade the
 dependent rows; today's single-table DELETE is correct because no such
 table exists yet, and the contract is “derived tables ride along”.
+## Contract metadata enrichment
+
+SoroTrail runs a background worker that enriches contracts with
+human-readable token metadata (name, symbol, decimals) by calling the
+SEP-41 token interface via RPC simulation.
+
+**How it works:**
+
+1. The worker periodically scans the events table for contracts that don't
+   yet have cached metadata (or whose cached metadata has exceeded the
+   TTL).
+2. For each candidate, it simulates calling `name()`, `symbol()`, and
+   `decimals()` on the contract via the RPC's `simulateTransaction`
+   endpoint.
+3. Results are stored in the `contract_meta` table with a `fetched_at`
+   timestamp.
+4. Contracts that don't implement the SEP-41 token interface (the
+   simulation traps or returns an error) are negatively cached so they're
+   never re-probed.
+5. Metadata surfaces in `/contracts` and `/contracts/{id}/stats` responses;
+   fields are `null` when unknown.
+
+**Design principles:**
+
+- **Never blocks ingestion** — the enrichment worker runs on its own
+  goroutine with its own poll interval.
+- **Rate-limited** — simulation calls go through the same RPC client with
+  the same ~10 req/s limiter, so enrichment doesn't overwhelm the endpoint.
+- **Failures are data, not errors** — failed RPC calls are logged at debug
+  level and the contract is retried on the next pass.
+- **Negative caching is permanent** — non-token contracts are cached with
+  `is_token=false` and never re-fetched, so contracts that emit events but
+  aren't tokens don't cause wasteful RPC calls.
+- **TTL-based refresh** — token metadata is re-fetched after
+  `CONTRACT_META_TTL` (default 24h) so name/symbol changes (e.g. a token
+  upgrade) are eventually picked up.
+
+Disable enrichment entirely with `CONTRACT_META_ENABLED=false`.
+
 ## Caching
 
 Stored events are immutable — a row written by ingest is never rewritten

cmd/sorotrail/backfill.go

@@ -119,7 +119,6 @@ flags:
 	minInterval := time.Duration(float64(time.Second) / rpsFinal)
 
 	st := store.NewPostgres(pool, int64(cfg.PartitionLedgerSpan))
-	tokenProc := ingester.NewTokenBalanceProcessor(st, log)
 	hClient := horizon.NewHTTPClient(hURL, minInterval)
 
 	b := backfill.New(hClient, st, decode.XDRDecoder{}, log, backfill.Options{
@@ -155,20 +154,6 @@ flags:
 		return err
 	}
 
-	if !*dryRun && sum.Completed && sum.Extracted > 0 {
-		// Process token balances for newly backfilled events. We query the
-		// store for events in the backfilled range and feed them through
-		// the token balance processor so holders are up-to-date.
-		log.Info("processing token balances for backfilled events",
-			"contract_id", *contractID,
-			"from_ledger", *fromLedger,
-			"to_ledger", sum.ThroughLedger,
-			"extracted", sum.Extracted)
-		if err := processBackfillTokenBalances(ctx, st, tokenProc, *contractID, *fromLedger, sum.ThroughLedger, cfg.DefaultNetworkName()); err != nil {
-			log.Warn("token balance processing for backfilled events", "error", err)
-		}
-	}
-
 	printBackfillSummary(sum, *dryRun)
 	if !sum.Completed {
 		return errInterrupted
@@ -291,41 +276,6 @@ func contains(haystack, needle string) bool {
 	return false
 }
 
-// processBackfillTokenBalances reads events for the backfilled range and feeds
-// them through the token balance processor so holder balances stay accurate.
-func processBackfillTokenBalances(ctx context.Context, st store.Store, proc *ingester.TokenBalanceProcessor, contractID string, fromLedger, toLedger int64, network string) error {
-	if toLedger <= 0 {
-		return nil
-	}
-	// Query events in batches to avoid loading everything into memory.
-	// We use the store's QueryEvents with order=asc to page through.
-	var cursor string
-	limit := 500
-	for {
-		events, next, err := st.QueryEvents(ctx, store.EventFilter{
-			ContractID: contractID,
-			Network:    network,
-			FromLedger: fromLedger,
-			ToLedger:   toLedger,
-			Limit:      limit,
-			Order:      "asc",
-			Cursor:     cursor,
-		})
-		if err != nil {
-			return fmt.Errorf("querying backfilled events: %w", err)
-		}
-		if len(events) == 0 {
-			break
-		}
-		proc.NotifyEvents(ctx, events)
-		if next == "" || len(events) < limit {
-			break
-		}
-		cursor = next
-	}
-	return nil
-}
-
 // printBackfillSummary mirrors replay's printReplaySummary shape so
 // operator scripts see consistent backfill output.
 func printBackfillSummary(s backfill.Summary, dryRun bool) {

cmd/sorotrail/main.go

@@ -20,7 +20,6 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
-	_ "modernc.org/sqlite"
 
 	"github.com/sorotrail/sorotrail/internal/api"
 	"github.com/sorotrail/sorotrail/internal/api/graphql"
@@ -33,18 +32,10 @@ import (
 	"github.com/sorotrail/sorotrail/internal/rpc"
 	"github.com/sorotrail/sorotrail/internal/spec"
 	"github.com/sorotrail/sorotrail/internal/store"
+	"github.com/sorotrail/sorotrail/internal/telemetry"
 	"github.com/sorotrail/sorotrail/internal/webhook"
 )
 
-// compositeNotifier fans out to multiple EventNotifiers.
-type compositeNotifier []ingester.EventNotifier
-
-func (n compositeNotifier) NotifyEvents(ctx context.Context, events []store.Event) {
-	for _, notifier := range n {
-		notifier.NotifyEvents(ctx, events)
-	}
-}
-
 var errInterrupted = errors.New("interrupted")
 
 func main() {
@@ -59,6 +50,8 @@ func main() {
 	}
 }
 
+// dispatch routes to a subcommand, defaulting to the indexer so existing
+// deployments (and the Dockerfile entrypoint) keep working unchanged.
 func dispatch(args []string) error {
 	if len(args) == 0 {
 		return run()
@@ -188,54 +181,19 @@ func run() error {
 		st = pg
 	}
 
-	// Tag any events that have empty network with the default network.
-	// This handles the upgrade path for single-network deployments.
-	if pool != nil {
-		defaultNetwork := cfg.DefaultNetworkName()
-		if defaultNetwork == "" {
-			networks := cfg.NetworksOrDefault()
-			if len(networks) > 0 {
-				defaultNetwork = networks[0].Name
-			}
-		}
-		if defaultNetwork != "" {
-			if _, err := pool.Exec(ctx, `UPDATE events SET network = $1 WHERE network = '' OR network IS NULL`, defaultNetwork); err != nil {
-				log.Warn("tagging legacy events with default network", "error", err)
-			}
-			if _, err := pool.Exec(ctx, `INSERT INTO ingestion_state (network, last_ingested_ledger, last_cursor, updated_at)
-				SELECT $1, last_ingested_ledger, last_cursor, updated_at FROM ingestion_state WHERE network = '' OR network IS NULL
-				ON CONFLICT (network) DO NOTHING`, defaultNetwork); err != nil {
-				log.Warn("migrating ingestion state", "error", err)
-			}
-			if _, err := pool.Exec(ctx, `INSERT INTO audit_state (network, verified_through_ledger, updated_at)
-				SELECT $1, verified_through_ledger, updated_at FROM audit_state WHERE network = '' OR network IS NULL
-				ON CONFLICT (network) DO NOTHING`, defaultNetwork); err != nil {
-				log.Warn("migrating audit state", "error", err)
-			}
-		}
-	}
-
 	for _, id := range cfg.WatchedContracts {
-		// In multi-network mode, watched contracts apply to all networks.
-		for _, n := range cfg.NetworksOrDefault() {
-			if err := st.AddWatchedContract(ctx, id); err != nil {
-				return err
-			}
-			_ = n // preserve for per-network contract lists
+		if err := st.AddWatchedContract(ctx, id); err != nil {
+			return err
 		}
 	}
 
 	// Shared broadcaster for live event streaming across all networks.
 	bcast := broadcast.New(broadcast.DefaultBufferSize)
 
-	// Build per-network components.
-	networks := cfg.NetworksOrDefault()
-	type networkIngester struct {
-		ing     *ingester.Ingester
-		auditor *audit.Auditor
-		rpc     rpc.Client
-	}
-	ingesters := make([]networkIngester, 0, len(networks))
+	rpcClient := rpc.NewHTTPClient(cfg.RPCURL)
+	wh := webhook.NewNotifier(st, log)
+
+	// Wire the spec cache and enricher for spec-decoded event views.
 	specCache := spec.NewCache(st)
 	specFetcher := spec.NewFetcher(rpcClient)
 	specEnricher := spec.NewEnricher(specFetcher, specCache, log)
@@ -340,10 +298,6 @@ func run() error {
 	limiter.Start(ctx)
 	defer limiter.Stop()
 
-	// Wire spec enricher for the API using the first network's fetcher.
-	firstSpecFetcher := spec.NewFetcher(ingesters[0].rpc)
-	firstSpecEnricher := spec.NewEnricher(firstSpecFetcher, specCache, log)
-
 	// Guarded store for API-originated reads with timeout and slow-query logging.
 	apiStore := store.NewGuardedStore(st, store.GuardedStoreOptions{
 		Timeout:            cfg.APIQueryTimeout,
@@ -429,7 +383,7 @@ func run() error {
 	if ingesterEnabled {
 		remaining++ // + ingester
 		go func() {
-			log.Info("ingester starting", "rpc_url", cfg.RPCURL, "poll_interval", cfg.PollInterval)
+			log.Info("ingester starting", "rpc_urls", rpcURLsForLog(cfg), "poll_interval", cfg.PollInterval)
 			if err := ing.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 				errCh <- fmt.Errorf("ingester: %w", err)
 			} else {
@@ -446,13 +400,27 @@ func run() error {
 		}
 	}()
 
-	// Count expected goroutines that send to errCh.
-	remaining := 2 + len(ingesters) // http server + webhook + ingesters
-	for _, ni := range ingesters {
-		if ni.auditor != nil {
-			remaining++
-		}
+	// The auditor runs alongside ingestion and reports into the same
+	// error channel when enabled.
+	if aud != nil {
+		remaining++ // + auditor
+		go func() {
+			log.Info("auditor starting",
+				"budget_share", cfg.AuditBudgetShare,
+				"batch_ledgers", cfg.AuditBatchLedgers,
+				"lag_threshold", cfg.AuditLagThreshold,
+				"max_repair_attempts", cfg.AuditMaxRepair)
+			if err := aud.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
+				errCh <- fmt.Errorf("auditor: %w", err)
+			} else {
+				errCh <- nil
+			}
+		}()
 	}
+
+	// The pruner goroutine always runs; without a retention policy it
+	// returns immediately and reports nil so shutdown accounting holds.
+	remaining++ // + pruner
 	go func() {
 		if cfg.RetentionEnabled() {
 			log.Info("pruner starting",
@@ -470,9 +438,6 @@ func run() error {
 	}()
 
 	var firstErr error
-	if aud != nil {
-		remaining++
-	}
 	select {
 	case <-ctx.Done():
 		log.Info("shutdown signal received")
@@ -483,7 +448,6 @@ func run() error {
 
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout)
 	defer cancel()
-	broker.Shutdown()
 	if err := server.Shutdown(shutdownCtx); err != nil {
 		log.Error("http shutdown", "error", err)
 	}
@@ -513,7 +477,7 @@ func bootstrapAdminKey(ctx context.Context, ts store.TenantStore, key string, lo
 	}
 	prefix, digest, ok := api.ParseAPIKeyForBootstrap(key)
 	if !ok {
-		return fmt.Errorf("MULTI_TENANT_BOOTSTRAP_KEY is not a valid key; " +
+		return fmt.Errorf("MULTI_TENANT_BOOTSTRAP_KEY is not a valid key; "+
 			"generate one with `sorotrail help` format st_<12 chars>_<secret>")
 	}
 	tenant, err := ts.GetTenantByName(ctx, "default")