feat: add Parquet reader for Iceberg-backed CloudEvent retrieval

.gitignore

@@ -32,3 +32,4 @@ __debug_bin*
 .env
 
 
+fetch-api

Makefile

@@ -1,4 +1,4 @@
-.PHONY: clean run build install dep test lint format docker
+.PHONY: clean run build install dep test lint format docker gqlgen podman
 
 SHELL := /bin/bash
 PATHINSTBIN = $(abspath ./bin)
@@ -20,7 +20,7 @@ VER_CUT   := $(shell echo $(VERSION) | cut -c2-)
 
 # Dependency versions
 GOLANGCI_VERSION   = latest
-PROTOC_VERSION             = 28.3
+PROTOC_VERSION             = 33.4
 PROTOC_GEN_GO_VERSION      = $(shell go list -m -f '{{.Version}}' google.golang.org/protobuf)
 PROTOC_GEN_GO_GRPC_VERSION = v1.5.1
 
@@ -56,9 +56,26 @@ lint: ## run linter
 	@PATH=$$PATH golangci-lint run --timeout 10m
 
 docker: dep ## build docker image
-	@docker build -f ./Dockerfile . -t dimozone/$(BIN_NAME):$(VER_CUT)
+	@docker build --build-arg APP_NAME=$(BIN_NAME) -f ./Dockerfile . -t dimozone/$(BIN_NAME):$(VER_CUT)
 	@docker tag dimozone/$(BIN_NAME):$(VER_CUT) dimozone/$(BIN_NAME):latest
 
+# Build multi-arch (amd64 + arm64) and push with a random tag. Does not trigger GitHub workflows.
+# Requires: docker buildx, docker login. Run from repo root.
+docker-push-multiarch:
+	$(eval TAG := dev-$(shell openssl rand -hex 6))
+	@echo "Building and pushing dimozone/$(BIN_NAME):$(TAG) (linux/amd64, linux/arm64)"
+	@docker buildx build --build-arg APP_NAME=$(BIN_NAME) --platform linux/amd64,linux/arm64 -f ./Dockerfile --push -t dimozone/$(BIN_NAME):$(TAG) .
+	@echo "Pushed dimozone/$(BIN_NAME):$(TAG)"
+
+# Same as docker-push-multiarch but using podman (manifest list + manifest push --all).
+podman-push-multiarch:
+	$(eval TAG := dev-$(shell openssl rand -hex 6))
+	$(eval IMAGE := dimozone/$(BIN_NAME):$(TAG))
+	@echo "Building and pushing $(IMAGE) (linux/amd64, linux/arm64)"
+	@podman build --build-arg APP_NAME=$(BIN_NAME) --platform linux/amd64,linux/arm64 --manifest $(IMAGE) -f ./Dockerfile .
+	@podman manifest push --all $(IMAGE) docker://$(IMAGE)
+	@echo "Pushed $(IMAGE)"
+
 tools-golangci-lint: ## install golangci-lint
 	@mkdir -p $(PATHINSTBIN)
 	curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | BINARY=golangci-lint bash -s -- ${GOLANGCI_VERSION}
@@ -80,11 +97,10 @@ endif
 
 make tools: tools-golangci-lint tools-protoc ## install all tools
 
-generate: generate-swagger generate-go generate-grpc ## run all file generation for the project
+gqlgen: ## Generate gqlgen code.
+	@go tool gqlgen generate
 
-generate-swagger: ## generate swagger documentation
-	@go tool swag -version
-	go tool swag init -g cmd/${BIN_NAME}/main.go --parseDependency --parseInternal
+generate: gqlgen generate-go generate-grpc ## run all file generation for the project
 
 generate-go:## run go generate
 	@go generate ./...

charts/fetch-api/values-prod.yaml

@@ -11,7 +11,7 @@ env:
   TOKEN_EXCHANGE_ISSUER_URL: https://auth-roles-rights.dimo.zone
   CLOUDEVENT_BUCKET: dimo-ingest-cloudevent-prod
   EPHEMERAL_BUCKET: dimo-ingest-ephemeral-prod
-  VC_BUCKET: dimo-network-vc-prod
+  PARQUET_BUCKET: dimo-iceberg-prod
   VEHICLE_NFT_ADDRESS: '0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF'
   CHAIN_ID: 137
 ingress:

charts/fetch-api/values.yaml

@@ -33,7 +33,7 @@ env:
   TOKEN_EXCHANGE_ISSUER_URL: https://auth-roles-rights.dev.dimo.zone
   CLOUDEVENT_BUCKET: dimo-ingest-cloudevent-dev
   EPHEMERAL_BUCKET: dimo-ingest-ephemeral-dev
-  VC_BUCKET: dimo-network-vc-dev
+  PARQUET_BUCKET: dimo-iceberg-dev
   S3_AWS_REGION: us-east-2
   VEHICLE_NFT_ADDRESS: '0x45fbCD3ef7361d156e8b16F5538AE36DEdf61Da8'
   CHAIN_ID: 80002

cmd/fetch-api/main.go

@@ -4,61 +4,81 @@ import (
 	"context"
 	"errors"
 	"flag"
+	"net/http"
 	"os"
 	"os/signal"
+	"runtime/debug"
 	"strconv"
+	"syscall"
 
-	_ "github.com/DIMO-Network/fetch-api/docs"
+	"github.com/99designs/gqlgen/graphql/playground"
 	"github.com/DIMO-Network/fetch-api/internal/app"
 	"github.com/DIMO-Network/fetch-api/internal/config"
-	"github.com/DIMO-Network/server-garage/pkg/env"
-	"github.com/DIMO-Network/server-garage/pkg/logging"
 	"github.com/DIMO-Network/server-garage/pkg/monserver"
 	"github.com/DIMO-Network/server-garage/pkg/runner"
+	"github.com/DIMO-Network/shared/pkg/settings"
+	"github.com/rs/zerolog"
 	"golang.org/x/sync/errgroup"
 )
 
-// @title                       DIMO Fetch API
-// @version                     1.0
-// @securityDefinitions.apikey  BearerAuth
-// @in                          header
-// @name                        Authorization
 func main() {
-	settingsFile := flag.String("env", ".env", "env file")
-	flag.Parse()
+	logger := zerolog.New(os.Stdout).With().Timestamp().Str("app", app.AppName).Logger()
+	if info, ok := debug.ReadBuildInfo(); ok {
+		for _, s := range info.Settings {
+			if s.Key == "vcs.revision" && len(s.Value) == 40 {
+				logger = logger.With().Str("commit", s.Value[:7]).Logger()
+				break
+			}
+		}
+	}
+	zerolog.DefaultContextLogger = &logger
 
-	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
-	defer cancel()
+	mainCtx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-mainCtx.Done()
+		logger.Info().Msg("Received signal, shutting down...")
+		cancel()
+	}()
 
-	logger := logging.GetAndSetDefaultLogger("fetch-api")
+	runnerGroup, runnerCtx := errgroup.WithContext(mainCtx)
 
-	settings, err := env.LoadSettings[config.Settings](*settingsFile)
+	settingsFile := flag.String("settings", "settings.yaml", "settings file")
+	flag.Parse()
+
+	cfg, err := settings.LoadConfig[config.Settings](*settingsFile)
 	if err != nil {
 		logger.Fatal().Err(err).Msg("Couldn't load settings.")
 	}
 
-	webServer, err := app.CreateWebServer(&settings)
+	application, err := app.New(cfg)
 	if err != nil {
-		logger.Fatal().Err(err).Msg("Failed to create web server.")
+		logger.Fatal().Err(err).Msg("Couldn't create application.")
 	}
-	rpcServer, err := app.CreateGRPCServer(&logger, &settings)
+	defer application.Cleanup()
+
+	rpcServer, err := app.CreateGRPCServer(&logger, &cfg)
 	if err != nil {
-		logger.Fatal().Err(err).Msg("Failed to create RPC server.")
+		logger.Fatal().Err(err).Msg("Failed to create gRPC server.")
 	}
 
-	group, gCtx := errgroup.WithContext(ctx)
+	monSrv := monserver.NewMonitoringServer(&logger, cfg.EnablePprof)
+	runner.RunHandler(runnerCtx, runnerGroup, monSrv, ":"+strconv.Itoa(cfg.MonPort))
 
-	monApp := monserver.NewMonitoringServer(&logger, settings.EnablePprof)
-	logger.Info().Str("port", strconv.Itoa(settings.MonPort)).Msgf("Starting monitoring server")
-	runner.RunHandler(gCtx, group, monApp, ":"+strconv.Itoa(settings.MonPort))
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /health", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("healthy"))
+	})
+	mux.Handle("/", app.LoggerMiddleware(app.PanicRecoveryMiddleware(playground.Handler("GraphQL playground", "/query"))))
+	mux.Handle("/query", application.Handler)
 
-	logger.Info().Str("port", strconv.Itoa(settings.Port)).Msgf("Starting web server")
-	runner.RunFiber(gCtx, group, webServer, ":"+strconv.Itoa(settings.Port))
+	logger.Info().Msgf("Server started on port: %d", cfg.Port)
+	runner.RunHandler(runnerCtx, runnerGroup, mux, ":"+strconv.Itoa(cfg.Port))
 
-	logger.Info().Str("port", strconv.Itoa(settings.GRPCPort)).Msgf("Starting gRPC server")
-	runner.RunGRPC(gCtx, group, rpcServer, ":"+strconv.Itoa(settings.GRPCPort))
+	logger.Info().Msgf("gRPC server started on port: %d", cfg.GRPCPort)
+	runner.RunGRPC(runnerCtx, runnerGroup, rpcServer, ":"+strconv.Itoa(cfg.GRPCPort))
 
-	err = group.Wait()
+	err = runnerGroup.Wait()
 	if err != nil && !errors.Is(err, context.Canceled) {
 		logger.Fatal().Err(err).Msg("Server shut down due to an error.")
 	}