server.go

package server

import (
	"context"
	"fmt"

	"net/http"
	"time"

	"encoding/binary"
	"errors"

	"connectrpc.com/connect"
	"connectrpc.com/grpcreflect"
	goheader "github.com/celestiaorg/go-header"
	datypes "github.com/evstack/ev-node/pkg/da/types"
	ds "github.com/ipfs/go-datastore"
	"github.com/rs/zerolog"
	"golang.org/x/net/http2"
	"golang.org/x/net/http2/h2c"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"

	"github.com/evstack/ev-node/pkg/config"
	"github.com/evstack/ev-node/pkg/p2p"
	"github.com/evstack/ev-node/pkg/store"
	"github.com/evstack/ev-node/pkg/telemetry"
	"github.com/evstack/ev-node/types"
	pb "github.com/evstack/ev-node/types/pb/evnode/v1"
	rpc "github.com/evstack/ev-node/types/pb/evnode/v1/v1connect"
)

var _ rpc.StoreServiceHandler = (*StoreServer)(nil)

// StoreServer implements the StoreService defined in the proto file
type StoreServer struct {
	store       store.Store
	headerStore goheader.Store[*types.SignedHeader]
	dataStore   goheader.Store[*types.Data]
	logger      zerolog.Logger
}

// NewStoreServer creates a new StoreServer instance
func NewStoreServer(
	store store.Store,
	headerStore goheader.Store[*types.SignedHeader],
	dataStore goheader.Store[*types.Data],
	logger zerolog.Logger,
) *StoreServer {
	return &StoreServer{
		store:       store,
		headerStore: headerStore,
		dataStore:   dataStore,
		logger:      logger,
	}
}

// GetBlock implements the GetBlock RPC method
func (s *StoreServer) GetBlock(
	ctx context.Context,
	req *connect.Request[pb.GetBlockRequest],
) (*connect.Response[pb.GetBlockResponse], error) {
	var header *types.SignedHeader
	var data *types.Data
	var err error

	switch identifier := req.Msg.Identifier.(type) {
	case *pb.GetBlockRequest_Height:
		fetchHeight := identifier.Height
		if fetchHeight == 0 {
			// Subcase 2a: Height is 0 -> Fetch latest block
			fetchHeight, err = s.store.Height(ctx)
			if err != nil {
				return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to get latest height: %w", err))
			}
			if fetchHeight == 0 {
				return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("store is empty, no latest block available"))
			}
		}
		// Fetch by the determined height (either specific or latest)
		header, data, err = s.store.GetBlockData(ctx, fetchHeight)

	case *pb.GetBlockRequest_Hash:
		hash := types.Hash(identifier.Hash)
		header, data, err = s.store.GetBlockByHash(ctx, hash)

	default:
		// This case handles potential future identifier types or invalid states
		return nil, connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("invalid or unsupported identifier type provided"))
	}

	if err != nil {
		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to retrieve block data: %w", err))
	}

	// Convert retrieved types to protobuf types
	pbHeader, err := header.ToProto()
	if err != nil {
		// Error during conversion indicates an issue with the retrieved data or proto definition
		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to convert block header to proto format: %w", err))
	}
	pbData := data.ToProto() // Assuming data.ToProto() exists and doesn't return an error

	// Return the successful response
	resp := &pb.GetBlockResponse{
		Block: &pb.Block{
			Header: pbHeader,
			Data:   pbData,
		},
	}

	// Fetch and set DA heights
	blockHeight := header.Height()
	if blockHeight > 0 { // DA heights are not stored for genesis/height 0 in the current impl
		headerDAHeightKey := store.GetHeightToDAHeightHeaderKey(blockHeight)
		headerDAHeightBytes, err := s.store.GetMetadata(ctx, headerDAHeightKey)
		if err == nil && len(headerDAHeightBytes) == 8 {
			resp.HeaderDaHeight = binary.LittleEndian.Uint64(headerDAHeightBytes)
		} else if err != nil && !errors.Is(err, ds.ErrNotFound) {
			s.logger.Error().Uint64("height", blockHeight).Err(err).Msg("Error fetching header DA height for block")
		}

		dataDAHeightKey := store.GetHeightToDAHeightDataKey(blockHeight)
		dataDAHeightBytes, err := s.store.GetMetadata(ctx, dataDAHeightKey)
		if err == nil && len(dataDAHeightBytes) == 8 {
			resp.DataDaHeight = binary.LittleEndian.Uint64(dataDAHeightBytes)
		} else if err != nil && !errors.Is(err, ds.ErrNotFound) {
			s.logger.Error().Uint64("height", blockHeight).Err(err).Msg("Error fetching data DA height for block")
		}
	}

	return connect.NewResponse(resp), nil
}

// GetState implements the GetState RPC method
func (s *StoreServer) GetState(
	ctx context.Context,
	req *connect.Request[emptypb.Empty],
) (*connect.Response[pb.GetStateResponse], error) {
	state, err := s.store.GetState(ctx)
	if err != nil {
		return nil, connect.NewError(connect.CodeNotFound, err)
	}

	// Convert state to protobuf type
	pbState := &pb.State{
		AppHash:         state.AppHash,
		LastBlockHeight: state.LastBlockHeight,
		LastBlockTime:   timestamppb.New(state.LastBlockTime),
		DaHeight:        state.DAHeight,
		ChainId:         state.ChainID,
		Version: &pb.Version{
			Block: state.Version.Block,
			App:   state.Version.App,
		},
		InitialHeight: state.InitialHeight,
	}

	return connect.NewResponse(&pb.GetStateResponse{
		State: pbState,
	}), nil
}

// GetGenesisDaHeight implements the GetGenesisDaHeight RPC method
func (s *StoreServer) GetGenesisDaHeight(
	ctx context.Context,
	_ *connect.Request[emptypb.Empty],
) (*connect.Response[pb.GetGenesisDaHeightResponse], error) {
	resp, err := s.GetMetadata(ctx, connect.NewRequest(&pb.GetMetadataRequest{
		Key: store.GenesisDAHeightKey,
	}))
	if err != nil {
		return nil, err
	}

	if len(resp.Msg.GetValue()) != 8 {
		return nil, connect.NewError(connect.CodeNotFound, errors.New("invalid metadata value"))
	}

	return connect.NewResponse(&pb.GetGenesisDaHeightResponse{
		Height: binary.LittleEndian.Uint64(resp.Msg.GetValue()),
	}), nil

}

// GetP2PStoreInfo implements the GetP2PStoreInfo RPC method
func (s *StoreServer) GetP2PStoreInfo(
	ctx context.Context,
	_ *connect.Request[emptypb.Empty],
) (*connect.Response[pb.GetP2PStoreInfoResponse], error) {
	snapshots := make([]*pb.P2PStoreSnapshot, 0, 2)

	if s.headerStore != nil {
		snapshot, err := collectP2PStoreSnapshot(ctx, s.headerStore, "Header Store")
		if err != nil {
			return nil, connect.NewError(connect.CodeInternal, err)
		}
		snapshots = append(snapshots, snapshot)
	}

	if s.dataStore != nil {
		snapshot, err := collectP2PStoreSnapshot(ctx, s.dataStore, "Data Store")
		if err != nil {
			return nil, connect.NewError(connect.CodeInternal, err)
		}
		snapshots = append(snapshots, snapshot)
	}

	return connect.NewResponse(&pb.GetP2PStoreInfoResponse{
		Stores: snapshots,
	}), nil
}

// GetMetadata implements the GetMetadata RPC method
func (s *StoreServer) GetMetadata(
	ctx context.Context,
	req *connect.Request[pb.GetMetadataRequest],
) (*connect.Response[pb.GetMetadataResponse], error) {
	value, err := s.store.GetMetadata(ctx, req.Msg.Key)
	if err != nil {
		return nil, connect.NewError(connect.CodeNotFound, err)
	}

	return connect.NewResponse(&pb.GetMetadataResponse{
		Value: value,
	}), nil
}

func collectP2PStoreSnapshot[H goheader.Header[H]](
	ctx context.Context,
	store goheader.Store[H],
	label string,
) (*pb.P2PStoreSnapshot, error) {
	snapshot := &pb.P2PStoreSnapshot{
		Label:  label,
		Height: store.Height(),
	}

	if head, err := store.Head(ctx); err == nil {
		snapshot.Head = toP2PStoreEntry(head)
	} else if !errors.Is(err, goheader.ErrEmptyStore) && !errors.Is(err, goheader.ErrNotFound) {
		return nil, fmt.Errorf("failed to read %s head: %w", label, err)
	}

	if tail, err := store.Tail(ctx); err == nil {
		snapshot.Tail = toP2PStoreEntry(tail)
	} else if !errors.Is(err, goheader.ErrEmptyStore) && !errors.Is(err, goheader.ErrNotFound) {
		return nil, fmt.Errorf("failed to read %s tail: %w", label, err)
	}

	return snapshot, nil
}

func toP2PStoreEntry[H goheader.Header[H]](item H) *pb.P2PStoreEntry {
	if any(item) == nil {
		return nil
	}

	entry := &pb.P2PStoreEntry{
		Height: item.Height(),
		Hash:   append([]byte(nil), item.Hash()...),
	}

	if ts := item.Time(); !ts.IsZero() {
		entry.Time = timestamppb.New(ts)
	}

	return entry
}

type ConfigServer struct {
	config config.Config
	signer []byte
	logger zerolog.Logger
}

func NewConfigServer(config config.Config, proposerAddress []byte, logger zerolog.Logger) *ConfigServer {
	return &ConfigServer{
		config: config,
		signer: proposerAddress,
		logger: logger,
	}
}

func (cs *ConfigServer) GetNamespace(
	ctx context.Context,
	req *connect.Request[emptypb.Empty],
) (*connect.Response[pb.GetNamespaceResponse], error) {

	hns := datypes.NamespaceFromString(cs.config.DA.GetNamespace())
	dns := datypes.NamespaceFromString(cs.config.DA.GetDataNamespace())

	return connect.NewResponse(&pb.GetNamespaceResponse{
		HeaderNamespace: hns.HexString(),
		DataNamespace:   dns.HexString(),
	}), nil
}

func (cs *ConfigServer) GetSignerInfo(
	ctx context.Context,
	req *connect.Request[emptypb.Empty],
) (*connect.Response[pb.GetSignerInfoResponse], error) {

	// If no signer is available, return an error
	if cs.signer == nil {
		return nil, connect.NewError(connect.CodeUnavailable, fmt.Errorf("sequencer signer not available"))
	}

	return connect.NewResponse(&pb.GetSignerInfoResponse{
		Address: cs.signer,
	}), nil
}

// P2PServer implements the P2PService defined in the proto file
type P2PServer struct {
	// Add dependencies needed for P2P functionality
	peerManager p2p.P2PRPC
}

// NewP2PServer creates a new P2PServer instance
func NewP2PServer(peerManager p2p.P2PRPC) *P2PServer {
	return &P2PServer{
		peerManager: peerManager,
	}
}

// GetPeerInfo implements the GetPeerInfo RPC method
func (p *P2PServer) GetPeerInfo(
	ctx context.Context,
	req *connect.Request[emptypb.Empty],
) (*connect.Response[pb.GetPeerInfoResponse], error) {
	peers, err := p.peerManager.GetPeers()
	if err != nil {
		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to get peer info: %w", err))
	}

	// Convert to protobuf format
	pbPeers := make([]*pb.PeerInfo, len(peers))
	for i, peer := range peers {
		pbPeers[i] = &pb.PeerInfo{
			Id:      peer.ID.String(),
			Address: peer.String(),
		}
	}

	return connect.NewResponse(&pb.GetPeerInfoResponse{
		Peers: pbPeers,
	}), nil
}

// GetNetInfo implements the GetNetInfo RPC method
func (p *P2PServer) GetNetInfo(
	ctx context.Context,
	req *connect.Request[emptypb.Empty],
) (*connect.Response[pb.GetNetInfoResponse], error) {
	netInfo, err := p.peerManager.GetNetworkInfo()
	if err != nil {
		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to get network info: %w", err))
	}

	pbNetInfo := &pb.NetInfo{
		Id:              netInfo.ID,
		ListenAddresses: netInfo.ListenAddress,
	}

	return connect.NewResponse(&pb.GetNetInfoResponse{
		NetInfo: pbNetInfo,
	}), nil
}

// NewServiceHandler creates a new HTTP handler for Store, P2P and Config services
func NewServiceHandler(
	store store.Store,
	headerStore goheader.Store[*types.SignedHeader],
	dataStore goheader.Store[*types.Data],
	peerManager p2p.P2PRPC,
	proposerAddress []byte,
	logger zerolog.Logger,
	config config.Config,
	bestKnown BestKnownHeightProvider,
) (http.Handler, error) {
	var storeServer rpc.StoreServiceHandler = NewStoreServer(store, headerStore, dataStore, logger)
	var p2pServer rpc.P2PServiceHandler = NewP2PServer(peerManager)
	var configServer rpc.ConfigServiceHandler = NewConfigServer(config, proposerAddress, logger)

	if config.Instrumentation.IsTracingEnabled() {
		storeServer = WithTracingStoreServer(storeServer)
		p2pServer = WithTracingP2PServer(p2pServer)
		configServer = WithTracingConfigServer(configServer)
	}

	mux := http.NewServeMux()

	compress1KB := connect.WithCompressMinBytes(1024)
	reflector := grpcreflect.NewStaticReflector(
		rpc.StoreServiceName,
		rpc.P2PServiceName,
		rpc.ConfigServiceName,
	)
	mux.Handle(grpcreflect.NewHandlerV1(reflector, compress1KB))
	mux.Handle(grpcreflect.NewHandlerV1Alpha(reflector, compress1KB))

	// Register StoreService
	storePath, storeHandler := rpc.NewStoreServiceHandler(storeServer)
	mux.Handle(storePath, storeHandler)

	// Register P2PService
	p2pPath, p2pHandler := rpc.NewP2PServiceHandler(p2pServer)
	mux.Handle(p2pPath, p2pHandler)

	configPath, configHandler := rpc.NewConfigServiceHandler(configServer)
	mux.Handle(configPath, configHandler)

	// Register custom HTTP endpoints
	RegisterCustomHTTPEndpoints(mux, store, peerManager, config, bestKnown, logger)

	var handler http.Handler = mux
	if config.Instrumentation.IsTracingEnabled() {
		handler = telemetry.ExtractTraceContext(mux)
	}

	// Use h2c to support HTTP/2 without TLS
	return h2c.NewHandler(handler, &http2.Server{
		IdleTimeout:          120 * time.Second,
		MaxReadFrameSize:     1 << 24,
		MaxConcurrentStreams: 100,
		ReadIdleTimeout:      30 * time.Second,
		PingTimeout:          15 * time.Second,
	}), nil
}