Merge branch 'develop' into feature/brc20-module-api

README.md

@@ -51,8 +51,6 @@ Here is our minimum database disk space requirement for each module.
 | ------ | -------------------------- | ---------------------------- |
 | Runes  | 10 GB                      | 150 GB                       |
 
-Here is our minimum database disk space requirement for each module.
-
 #### 4. Prepare `config.yaml` file.
 
 ```yaml
@@ -108,7 +106,7 @@ We will be using `docker-compose` for our installation guide. Make sure the `doc
 # docker-compose.yaml
 services:
   gaze-indexer:
-    image: ghcr.io/gaze-network/gaze-indexer:v1.0.0
+    image: ghcr.io/gaze-network/gaze-indexer:v0.2.1
     container_name: gaze-indexer
     restart: unless-stopped
     ports:

cmd/cmd_run.go

@@ -23,10 +23,15 @@ import (
 	"github.com/gaze-network/indexer-network/pkg/errorhandler"
 	"github.com/gaze-network/indexer-network/pkg/logger"
 	"github.com/gaze-network/indexer-network/pkg/logger/slogx"
+	"github.com/gaze-network/indexer-network/pkg/middleware/requestcontext"
+	"github.com/gaze-network/indexer-network/pkg/middleware/requestlogger"
 	"github.com/gaze-network/indexer-network/pkg/reportingclient"
 	"github.com/gofiber/fiber/v2"
 	"github.com/gofiber/fiber/v2/middleware/compress"
+	"github.com/gofiber/fiber/v2/middleware/cors"
+	"github.com/gofiber/fiber/v2/middleware/favicon"
 	fiberrecover "github.com/gofiber/fiber/v2/middleware/recover"
+	"github.com/gofiber/fiber/v2/middleware/requestid"
 	"github.com/samber/do/v2"
 	"github.com/samber/lo"
 	"github.com/spf13/cobra"
@@ -137,6 +142,14 @@ func runHandler(cmd *cobra.Command, _ []string) error {
 			ErrorHandler: errorhandler.NewHTTPErrorHandler(),
 		})
 		app.
+			Use(favicon.New()).
+			Use(cors.New()).
+			Use(requestid.New()).
+			Use(requestcontext.New(
+				requestcontext.WithRequestId(),
+				requestcontext.WithClientIP(conf.HTTPServer.RequestIP),
+			)).
+			Use(requestlogger.New(conf.HTTPServer.Logger)).
 			Use(fiberrecover.New(fiberrecover.Config{
 				EnableStackTrace: true,
 				StackTraceHandler: func(c *fiber.Ctx, e interface{}) {

config.example.yaml

@@ -23,6 +23,14 @@ reporting:
 # HTTP server configuration options.
 http_server:
   port: 8080 # Port to run the HTTP server on for modules with HTTP API handlers.
+  logger:
+    disable: false # disable logger if logger level is `INFO`
+    request_header: false
+    request_query: false
+  requestip: # Client IP extraction configuration options. This is unnecessary if you don't care about the real client IP or if you're not using a reverse proxy.
+    trusted_proxies_ip: # Cloudflare, GCP Public LB. See: server/internal/middleware/requestcontext/PROXY-IP.md
+    trusted_proxies_header: # X-Real-IP, CF-Connecting-IP
+    enable_reject_malformed_request: false # return 403 if request is malformed (invalid IP)
 
 # Meta-protocol modules configuration options.
 modules:

core/constants/constants.go

@@ -1,5 +1,5 @@
 package constants
 
 const (
-	Version = "v0.0.1"
+	Version = "v0.2.1"
 )

core/indexer/indexer.go

@@ -91,6 +91,10 @@ func (i *Indexer[T]) Run(ctx context.Context) (err error) {
 		select {
 		case <-i.quit:
 			logger.InfoContext(ctx, "Got quit signal, stopping indexer")
+			if err := i.Processor.Shutdown(ctx); err != nil {
+				logger.ErrorContext(ctx, "Failed to shutdown processor", slogx.Error(err))
+				return errors.Wrap(err, "processor shutdown failed")
+			}
 			return nil
 		case <-ctx.Done():
 			return nil
@@ -204,9 +208,9 @@ func (i *Indexer[T]) process(ctx context.Context) (err error) {
 			}
 
 			// validate is input is continuous and no reorg
-			for i := 1; i < len(inputs); i++ {
-				header := inputs[i].BlockHeader()
-				prevHeader := inputs[i-1].BlockHeader()
+			prevHeader := i.currentBlock
+			for i, input := range inputs {
+				header := input.BlockHeader()
 				if header.Height != prevHeader.Height+1 {
 					return errors.Wrapf(errs.InternalError, "input is not continuous, input[%d] height: %d, input[%d] height: %d", i-1, prevHeader.Height, i, header.Height)
 				}
@@ -217,6 +221,7 @@ func (i *Indexer[T]) process(ctx context.Context) (err error) {
 					// end current round
 					return nil
 				}
+				prevHeader = header
 			}
 
 			ctx = logger.WithContext(ctx, slog.Int("total_inputs", len(inputs)))

core/indexer/interface.go

@@ -29,6 +29,9 @@ type Processor[T Input] interface {
 	// VerifyStates verifies the states of the indexed data and the indexer
 	// to ensure the last shutdown was graceful and no missing data.
 	VerifyStates(ctx context.Context) error
+
+	// Shutdown gracefully stops the processor. Database connections, network calls, leftover states, etc. should be closed and cleaned up here.
+	Shutdown(ctx context.Context) error
 }
 
 type IndexerWorker interface {

internal/config/config.go

@@ -12,6 +12,8 @@ import (
 	runesconfig "github.com/gaze-network/indexer-network/modules/runes/config"
 	"github.com/gaze-network/indexer-network/pkg/logger"
 	"github.com/gaze-network/indexer-network/pkg/logger/slogx"
+	"github.com/gaze-network/indexer-network/pkg/middleware/requestcontext"
+	"github.com/gaze-network/indexer-network/pkg/middleware/requestlogger"
 	"github.com/gaze-network/indexer-network/pkg/reportingclient"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
@@ -65,7 +67,9 @@ type Modules struct {
 }
 
 type HTTPServerConfig struct {
-	Port int `mapstructure:"port"`
+	Port      int                               `mapstructure:"port"`
+	Logger    requestlogger.Config              `mapstructure:"logger"`
+	RequestIP requestcontext.WithClientIPConfig `mapstructure:"requestip"`
 }
 
 // Parse parse the configuration from environment variables

modules/runes/processor.go

@@ -31,6 +31,7 @@ type Processor struct {
 	bitcoinClient   btcclient.Contract
 	network         common.Network
 	reportingClient *reportingclient.ReportingClient
+	cleanupFuncs    []func(context.Context) error
 
 	newRuneEntries      map[runes.RuneId]*runes.RuneEntry
 	newRuneEntryStates  map[runes.RuneId]*runes.RuneEntry
@@ -40,13 +41,14 @@ type Processor struct {
 	newRuneTxs          []*entity.RuneTransaction
 }
 
-func NewProcessor(runesDg datagateway.RunesDataGateway, indexerInfoDg datagateway.IndexerInfoDataGateway, bitcoinClient btcclient.Contract, network common.Network, reportingClient *reportingclient.ReportingClient) *Processor {
+func NewProcessor(runesDg datagateway.RunesDataGateway, indexerInfoDg datagateway.IndexerInfoDataGateway, bitcoinClient btcclient.Contract, network common.Network, reportingClient *reportingclient.ReportingClient, cleanupFuncs []func(context.Context) error) *Processor {
 	return &Processor{
 		runesDg:             runesDg,
 		indexerInfoDg:       indexerInfoDg,
 		bitcoinClient:       bitcoinClient,
 		network:             network,
 		reportingClient:     reportingClient,
+		cleanupFuncs:        cleanupFuncs,
 		newRuneEntries:      make(map[runes.RuneId]*runes.RuneEntry),
 		newRuneEntryStates:  make(map[runes.RuneId]*runes.RuneEntry),
 		newOutPointBalances: make(map[wire.OutPoint][]*entity.OutPointBalance),
@@ -228,3 +230,13 @@ func (p *Processor) RevertData(ctx context.Context, from int64) error {
 	}
 	return nil
 }
+
+func (p *Processor) Shutdown(ctx context.Context) error {
+	var errs []error
+	for _, cleanup := range p.cleanupFuncs {
+		if err := cleanup(ctx); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.WithStack(errors.Join(errs...))
+}

modules/runes/runes.go

@@ -33,6 +33,7 @@ func New(injector do.Injector) (indexer.IndexerWorker, error) {
 		runesDg       runesdatagateway.RunesDataGateway
 		indexerInfoDg runesdatagateway.IndexerInfoDataGateway
 	)
+	var cleanupFuncs []func(context.Context) error
 	switch strings.ToLower(conf.Modules.Runes.Database) {
 	case "postgresql", "postgres", "pg":
 		pg, err := postgres.NewPool(ctx, conf.Modules.Runes.Postgres)
@@ -42,7 +43,10 @@ func New(injector do.Injector) (indexer.IndexerWorker, error) {
 			}
 			return nil, errors.Wrap(err, "can't create Postgres connection pool")
 		}
-		defer pg.Close()
+		cleanupFuncs = append(cleanupFuncs, func(ctx context.Context) error {
+			pg.Close()
+			return nil
+		})
 		runesRepo := runespostgres.NewRepository(pg)
 		runesDg = runesRepo
 		indexerInfoDg = runesRepo
@@ -62,7 +66,7 @@ func New(injector do.Injector) (indexer.IndexerWorker, error) {
 		return nil, errors.Wrapf(errs.Unsupported, "%q datasource is not supported", conf.Modules.Runes.Datasource)
 	}
 
-	processor := NewProcessor(runesDg, indexerInfoDg, bitcoinClient, conf.Network, reportingClient)
+	processor := NewProcessor(runesDg, indexerInfoDg, bitcoinClient, conf.Network, reportingClient, cleanupFuncs)
 	if err := processor.VerifyStates(ctx); err != nil {
 		return nil, errors.WithStack(err)
 	}

pkg/middleware/requestcontext/PROXY-IP.md

@@ -0,0 +1,7 @@
+# Proxies IP Range Resources
+
+- Cloudflare - https://www.cloudflare.com/ips/
+- GCP Load Balancer - https://cloud.google.com/load-balancing/docs/health-check-concepts#ip-ranges
+- GCP Compute Engine, Customer-usable external IP address ranges - https://www.gstatic.com/ipranges/cloud.json
+- Other GCP Services - https://cloud.google.com/compute/docs/faq#networking
+- Other Resources - https://github.com/lord-alfred/ipranges

pkg/middleware/requestcontext/errors.go

@@ -0,0 +1,21 @@
+package requestcontext
+
+// requestcontextError implements error interface
+var _ error = requestcontextError{}
+
+type requestcontextError struct {
+	err     error
+	status  int
+	message string
+}
+
+func (r requestcontextError) Error() string {
+	if r.err != nil {
+		return r.err.Error()
+	}
+	return r.message
+}
+
+func (r requestcontextError) Unwrap() error {
+	return r.err
+}

pkg/middleware/requestcontext/requestcontext.go

@@ -0,0 +1,44 @@
+package requestcontext
+
+import (
+	"context"
+	"log/slog"
+	"net/http"
+
+	"github.com/cockroachdb/errors"
+	"github.com/gaze-network/indexer-network/pkg/logger"
+	"github.com/gofiber/fiber/v2"
+)
+
+type Response struct {
+	Result any    `json:"result"`
+	Error  string `json:"error,omitempty"`
+}
+
+type Option func(ctx context.Context, c *fiber.Ctx) (context.Context, error)
+
+func New(opts ...Option) fiber.Handler {
+	return func(c *fiber.Ctx) error {
+		var err error
+		ctx := c.UserContext()
+		for i, opt := range opts {
+			ctx, err = opt(ctx, c)
+			if err != nil {
+				rErr := requestcontextError{}
+				if errors.As(err, &rErr) {
+					return c.Status(rErr.status).JSON(Response{Error: rErr.message})
+				}
+
+				logger.ErrorContext(ctx, "failed to extract request context",
+					err,
+					slog.String("event", "requestcontext/error"),
+					slog.String("module", "requestcontext"),
+					slog.Int("optionIndex", i),
+				)
+				return c.Status(http.StatusInternalServerError).JSON(Response{Error: "internal server error"})
+			}
+		}
+		c.SetUserContext(ctx)
+		return c.Next()
+	}
+}

pkg/middleware/requestcontext/with_clientip.go

@@ -0,0 +1,150 @@
+package requestcontext
+
+import (
+	"context"
+	"log/slog"
+	"net"
+
+	"github.com/cockroachdb/errors"
+	"github.com/gaze-network/indexer-network/pkg/logger"
+	"github.com/gofiber/fiber/v2"
+)
+
+type clientIPKey struct{}
+
+type WithClientIPConfig struct {
+	// [Optional] TrustedProxiesIP is a list of all proxies IP ranges that's between the server and the client.
+	//
+	// If it's provided, it will walk backwards from the last IP in `X-Forwarded-For` header
+	// and use first IP that's not trusted proxy(not in the given IP ranges.)
+	//
+	// **If you want to use this option, you should provide all of probable proxies IP ranges.**
+	//
+	// This is lowest priority.
+	TrustedProxiesIP []string `env:"TRUSTED_PROXIES_IP" mapstructure:"trusted_proxies_ip"`
+
+	// [Optional] TrustedHeader is a header name for getting client IP. (e.g. X-Real-IP, CF-Connecting-IP, etc.)
+	//
+	// This is highest priority, it will ignore rest of the options if it's provided.
+	TrustedHeader string `env:"TRUSTED_HEADER" mapstructure:"trusted_proxies_header"`
+
+	// EnableRejectMalformedRequest return 403 Forbidden if the request is from proxies, but can't extract client IP
+	EnableRejectMalformedRequest bool `env:"ENABLE_REJECT_MALFORMED_REQUEST" envDefault:"false" mapstructure:"enable_reject_malformed_request"`
+}
+
+// WithClientIP setup client IP context with XFF Spoofing prevention support.
+//
+// If request is from proxies, it will use first IP from `X-Forwarded-For` header by default.
+func WithClientIP(config WithClientIPConfig) Option {
+	var trustedProxies trustedProxy
+	if len(config.TrustedProxiesIP) > 0 {
+		proxy, err := newTrustedProxy(config.TrustedProxiesIP)
+		if err != nil {
+			logger.Panic("Failed to parse trusted proxies", err)
+		}
+		trustedProxies = proxy
+	}
+
+	return func(ctx context.Context, c *fiber.Ctx) (context.Context, error) {
+		// Extract client IP from given header
+		if config.TrustedHeader != "" {
+			headerIP := c.Get(config.TrustedHeader)
+
+			// validate ip from header
+			if ip := net.ParseIP(headerIP); ip != nil {
+				return context.WithValue(ctx, clientIPKey{}, headerIP), nil
+			}
+		}
+
+		// Extract client IP from XFF header
+		rawIPs := c.IPs()
+		ips := parseIPs(rawIPs)
+
+		// If the request is directly from client, we can use direct remote IP address
+		if len(ips) == 0 {
+			return context.WithValue(ctx, clientIPKey{}, c.IP()), nil
+		}
+
+		// Walk back and find first IP that's not trusted proxy
+		if len(trustedProxies) > 0 {
+			for i := len(ips) - 1; i >= 0; i-- {
+				if !trustedProxies.IsTrusted(ips[i]) {
+					return context.WithValue(ctx, clientIPKey{}, ips[i].String()), nil
+				}
+			}
+
+			// If all IPs are trusted proxies, return first IP in XFF header
+			return context.WithValue(ctx, clientIPKey{}, rawIPs[0]), nil
+		}
+
+		// Finally, if we can't extract client IP, return forbidden
+		if config.EnableRejectMalformedRequest {
+			logger.WarnContext(ctx, "IP Spoofing detected, returning 403 Forbidden",
+				slog.String("event", "requestcontext/ip_spoofing_detected"),
+				slog.String("module", "requestcontext/with_clientip"),
+				slog.String("ip", c.IP()),
+				slog.Any("ips", rawIPs),
+			)
+			return nil, requestcontextError{
+				status:  fiber.StatusForbidden,
+				message: "not allowed to access",
+			}
+		}
+
+		// Fallback to first IP in XFF header
+		return context.WithValue(ctx, clientIPKey{}, rawIPs[0]), nil
+	}
+}
+
+// GetClientIP get clientIP from context. If not found, return empty string
+//
+// Warning: Request context should be setup before using this function
+func GetClientIP(ctx context.Context) string {
+	if ip, ok := ctx.Value(clientIPKey{}).(string); ok {
+		return ip
+	}
+	return ""
+}
+
+type trustedProxy []*net.IPNet
+
+// newTrustedProxy create a new trusted proxies instance for preventing IP spoofing (XFF Attacks)
+func newTrustedProxy(ranges []string) (trustedProxy, error) {
+	nets, err := parseCIDRs(ranges)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return trustedProxy(nets), nil
+}
+
+func (t trustedProxy) IsTrusted(ip net.IP) bool {
+	if ip == nil {
+		return false
+	}
+	for _, r := range t {
+		if r.Contains(ip) {
+			return true
+		}
+	}
+	return false
+}
+
+func parseCIDRs(ranges []string) ([]*net.IPNet, error) {
+	nets := make([]*net.IPNet, 0, len(ranges))
+	for _, r := range ranges {
+		_, ipnet, err := net.ParseCIDR(r)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to parse CIDR for %q", r)
+		}
+		nets = append(nets, ipnet)
+	}
+	return nets, nil
+}
+
+func parseIPs(ranges []string) []net.IP {
+	ip := make([]net.IP, 0, len(ranges))
+	for _, r := range ranges {
+		ip = append(ip, net.ParseIP(r))
+	}
+	return ip
+}

pkg/middleware/requestcontext/with_requestid.go

@@ -0,0 +1,47 @@
+package requestcontext
+
+import (
+	"context"
+
+	"github.com/gaze-network/indexer-network/pkg/logger"
+	"github.com/gofiber/fiber/v2"
+	"github.com/gofiber/fiber/v2/middleware/requestid"
+	fiberutils "github.com/gofiber/fiber/v2/utils"
+)
+
+type requestIdKey struct{}
+
+// GetRequestId get requestId from context. If not found, return empty string
+//
+// Warning: Request context should be setup before using this function
+func GetRequestId(ctx context.Context) string {
+	if id, ok := ctx.Value(requestIdKey{}).(string); ok {
+		return id
+	}
+	return ""
+}
+
+func WithRequestId() Option {
+	return func(ctx context.Context, c *fiber.Ctx) (context.Context, error) {
+		// Try to get id from fiber context.
+		requestId, ok := c.Locals(requestid.ConfigDefault.ContextKey).(string)
+		if !ok || requestId == "" {
+			// Try to get id from request, else we generate one
+			requestId = c.Get(requestid.ConfigDefault.Header, fiberutils.UUID())
+
+			// Set new id to response header
+			c.Set(requestid.ConfigDefault.Header, requestId)
+
+			// Add the request ID to locals (fasthttp UserValue storage)
+			c.Locals(requestid.ConfigDefault.ContextKey, requestId)
+		}
+
+		// Add the request ID to context
+		ctx = context.WithValue(ctx, requestIdKey{}, requestId)
+
+		// Add the requuest ID to context logger
+		ctx = logger.WithContext(ctx, "requestId", requestId)
+
+		return ctx, nil
+	}
+}

pkg/middleware/requestlogger/requestlogger.go

@@ -0,0 +1,115 @@
+package requestlogger
+
+import (
+	"log/slog"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/cockroachdb/errors"
+	"github.com/gaze-network/indexer-network/pkg/logger"
+	"github.com/gaze-network/indexer-network/pkg/middleware/requestcontext"
+	"github.com/gofiber/fiber/v2"
+)
+
+type Config struct {
+	WithRequestHeader    bool     `env:"REQUEST_HEADER" envDefault:"false" mapstructure:"request_header"`
+	WithRequestQuery     bool     `env:"REQUEST_QUERY" envDefault:"false" mapstructure:"request_query"`
+	Disable              bool     `env:"DISABLE" envDefault:"false" mapstructure:"disable"` // Disable logger level `INFO`
+	HiddenRequestHeaders []string `env:"HIDDEN_REQUEST_HEADERS" mapstructure:"hidden_request_headers"`
+}
+
+// New setup request context and information
+func New(config Config) fiber.Handler {
+	hiddenRequestHeaders := make(map[string]struct{}, len(config.HiddenRequestHeaders))
+	for _, header := range config.HiddenRequestHeaders {
+		hiddenRequestHeaders[strings.TrimSpace(strings.ToLower(header))] = struct{}{}
+	}
+	return func(c *fiber.Ctx) error {
+		start := time.Now()
+
+		// Continue stack
+		err := c.Next()
+
+		end := time.Now()
+		latency := end.Sub(start)
+		status := c.Response().StatusCode()
+
+		baseAttrs := []slog.Attr{
+			slog.String("event", "api_request"),
+			slog.Int64("latency", latency.Milliseconds()),
+			slog.String("latencyHuman", latency.String()),
+		}
+
+		// prep request attributes
+		requestAttributes := []slog.Attr{
+			slog.Time("time", start),
+			slog.String("method", c.Method()),
+			slog.String("host", c.Hostname()),
+			slog.String("path", c.Path()),
+			slog.String("route", c.Route().Path),
+			slog.String("ip", requestcontext.GetClientIP(c.UserContext())),
+			slog.String("remoteIP", c.Context().RemoteIP().String()),
+			slog.Any("x-forwarded-for", c.IPs()),
+			slog.String("user-agent", string(c.Context().UserAgent())),
+			slog.Any("params", c.AllParams()),
+			slog.Int("length", len((c.Body()))),
+		}
+
+		// prep response attributes
+		responseAttributes := []slog.Attr{
+			slog.Time("time", end),
+			slog.Int("status", status),
+			slog.Int("length", len(c.Response().Body())),
+		}
+
+		// request query
+		if config.WithRequestQuery {
+			requestAttributes = append(requestAttributes, slog.String("query", string(c.Request().URI().QueryString())))
+		}
+
+		// request headers
+		if config.WithRequestHeader {
+			kv := []any{}
+
+			for k, v := range c.GetReqHeaders() {
+				if _, found := hiddenRequestHeaders[strings.ToLower(k)]; found {
+					continue
+				}
+				kv = append(kv, slog.Any(k, v))
+			}
+
+			requestAttributes = append(requestAttributes, slog.Group("header", kv...))
+		}
+
+		level := slog.LevelInfo
+		if err != nil || status >= http.StatusInternalServerError {
+			level = slog.LevelError
+
+			// error attributes
+			logErr := err
+			if logErr == nil {
+				logErr = fiber.NewError(status)
+			}
+			baseAttrs = append(baseAttrs, slog.Any("error", logErr))
+		}
+
+		if config.Disable && level == slog.LevelInfo {
+			return errors.WithStack(err)
+		}
+
+		logger.LogAttrs(c.UserContext(), level, "Request Completed", append([]slog.Attr{
+			{
+				Key:   "request",
+				Value: slog.GroupValue(requestAttributes...),
+			},
+			{
+				Key:   "response",
+				Value: slog.GroupValue(responseAttributes...),
+			},
+		}, baseAttrs...)...,
+		)
+
+		return errors.WithStack(err)
+	}
+}