feat: Implemented nodesale indexer API

modules/nodesale/api/httphandler/handlers.go

@@ -1 +1,196 @@
 package httphandler
+
+import (
+	"fmt"
+
+	"github.com/gaze-network/indexer-network/modules/nodesale/protobuf"
+	repository "github.com/gaze-network/indexer-network/modules/nodesale/repository/postgres"
+	"github.com/gaze-network/indexer-network/modules/nodesale/repository/postgres/gen"
+	"github.com/gofiber/fiber/v2"
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+type handler struct {
+	repository *repository.Repository
+}
+
+func New(repo *repository.Repository) *handler {
+	h := handler{}
+	h.repository = repo
+	return &h
+}
+
+func (h *handler) infoHandler(ctx *fiber.Ctx) error {
+	block, err := h.repository.Queries.GetLastProcessedBlock(ctx.UserContext())
+	if err != nil {
+		return fmt.Errorf("Cannot get last processed block : %w", err)
+	}
+	err = ctx.JSON(infoResponse{
+		IndexedBlockHeight: block.BlockHeight,
+		IndexedBlockHash:   block.BlockHash,
+	})
+	if err != nil {
+		return fmt.Errorf("Go fiber cannot parse JSON: %w", err)
+	}
+	return nil
+}
+
+func (h *handler) deployHandler(ctx *fiber.Ctx) error {
+	deployId := ctx.Params("deployId")
+	if deployId == "" {
+		err := ctx.SendStatus(404)
+		if err != nil {
+			return fmt.Errorf("Go fiber cannot send status: %w", err)
+		}
+		return nil
+	}
+	var blockHeight, txIndex int32
+	count, err := fmt.Sscanf(deployId, "%d-%d", &blockHeight, &txIndex)
+	if count != 2 || err != nil {
+		err := ctx.SendStatus(404)
+		if err != nil {
+			return fmt.Errorf("Go fiber cannot send status: %w", err)
+		}
+		return nil
+	}
+	deploys, err := h.repository.Queries.GetNodesale(ctx.UserContext(), gen.GetNodesaleParams{
+		BlockHeight: blockHeight,
+		TxIndex:     txIndex,
+	})
+	if err != nil {
+		return fmt.Errorf("Cannot get nodesale from db: %w", err)
+	}
+	if len(deploys) < 1 {
+		err := ctx.SendStatus(404)
+		if err != nil {
+			return fmt.Errorf("Go fiber cannot send status: %w", err)
+		}
+		return nil
+	}
+
+	deploy := deploys[0]
+
+	nodeCount, err := h.repository.Queries.GetNodeCountByTierIndex(ctx.UserContext(), gen.GetNodeCountByTierIndexParams{
+		SaleBlock:   deploy.BlockHeight,
+		SaleTxIndex: deploy.TxIndex,
+		FromTier:    0,
+		ToTier:      int32(len(deploy.Tiers) - 1),
+	})
+	if err != nil {
+		return fmt.Errorf("Cannot get node count from db : %w", err)
+	}
+
+	tiers := make([]protobuf.Tier, len(deploy.Tiers))
+	tierResponses := make([]tierResponse, len(deploy.Tiers))
+	for i, tierJson := range deploy.Tiers {
+		tier := &tiers[i]
+		err := protojson.Unmarshal(tierJson, tier)
+		if err != nil {
+			return fmt.Errorf("Failed to decode tiers json : %w", err)
+		}
+		tierResponses[i].Limit = tiers[i].Limit
+		tierResponses[i].MaxPerAddress = tiers[i].MaxPerAddress
+		tierResponses[i].PriceSat = tiers[i].PriceSat
+		tierResponses[i].Sold = nodeCount[i].Count
+	}
+
+	err = ctx.JSON(&deployResponse{
+		Id:              deployId,
+		Name:            deploy.Name,
+		StartAt:         deploy.StartsAt.Time.UTC(),
+		EndAt:           deploy.EndsAt.Time.UTC(),
+		Tiers:           tierResponses,
+		SellerPublicKey: deploy.SellerPublicKey,
+		MaxPerAddress:   deploy.MaxPerAddress,
+		DeployTxHash:    deploy.DeployTxHash,
+	})
+	if err != nil {
+		return fmt.Errorf("Go fiber cannot parse JSON: %w", err)
+	}
+	return nil
+}
+
+func (h *handler) nodesHandler(ctx *fiber.Ctx) error {
+	deployId := ctx.Query("deployId")
+	if deployId == "" {
+		err := ctx.SendStatus(404)
+		if err != nil {
+			return fmt.Errorf("Go fiber cannot send status: %w", err)
+		}
+		return nil
+	}
+	ownerPublicKey := ctx.Query("ownerPublicKey")
+	delegateePublicKey := ctx.Query("delegateePublicKey")
+
+	var blockHeight, txIndex int32
+	count, err := fmt.Sscanf(deployId, "%d-%d", &blockHeight, &txIndex)
+	if count != 2 || err != nil {
+		err := ctx.SendStatus(404)
+		if err != nil {
+			return fmt.Errorf("Go fiber cannot send status: %w", err)
+		}
+		return nil
+	}
+
+	nodes, err := h.repository.Queries.GetNodesByPubkey(ctx.UserContext(), gen.GetNodesByPubkeyParams{
+		SaleBlock:      blockHeight,
+		SaleTxIndex:    txIndex,
+		OwnerPublicKey: ownerPublicKey,
+		DelegatedTo:    delegateePublicKey,
+	})
+	if err != nil {
+		err := ctx.SendStatus(404)
+		if err != nil {
+			return fmt.Errorf("Can't get nodes from db: %w", err)
+		}
+		return nil
+	}
+	responses := make([]nodeResponse, len(nodes))
+	for i, node := range nodes {
+		responses[i].DeployId = deployId
+		responses[i].NodeId = node.NodeID
+		responses[i].TierIndex = node.TierIndex
+		responses[i].DelegatedTo = node.DelegatedTo
+		responses[i].OwnerPublicKey = node.OwnerPublicKey
+		responses[i].PurchaseTxHash = node.PurchaseTxHash
+		responses[i].DelegateTxHash = node.DelegateTxHash
+		responses[i].PurchaseBlockHeight = node.TxIndex
+	}
+
+	err = ctx.JSON(responses)
+	if err != nil {
+		return fmt.Errorf("Go fiber cannot parse JSON: %w", err)
+	}
+	return nil
+}
+
+func (h *handler) eventsHandler(ctx *fiber.Ctx) error {
+	walletAddress := ctx.Query("walletAddress")
+
+	events, err := h.repository.Queries.GetEventsByWallet(ctx.UserContext(), walletAddress)
+	if err != nil {
+		err := ctx.SendStatus(404)
+		if err != nil {
+			return fmt.Errorf("Can't get events from db: %w", err)
+		}
+		return nil
+	}
+
+	responses := make([]eventResposne, len(events))
+	for i, event := range events {
+		responses[i].TxHash = event.TxHash
+		responses[i].BlockHeight = event.BlockHeight
+		responses[i].TxIndex = event.TxIndex
+		responses[i].WalletAddress = event.WalletAddress
+		responses[i].Action = protobuf.Action_name[event.Action]
+		responses[i].ParsedMessage = event.ParsedMessage
+		responses[i].BlockTimestamp = event.BlockTimestamp.Time
+		responses[i].BlockHash = event.BlockHash
+	}
+
+	err = ctx.JSON(responses)
+	if err != nil {
+		return fmt.Errorf("Go fiber cannot parse JSON: %w", err)
+	}
+	return nil
+}

modules/nodesale/api/httphandler/responses.go

@@ -0,0 +1,51 @@
+package httphandler
+
+import (
+	"encoding/json"
+	"time"
+)
+
+type infoResponse struct {
+	IndexedBlockHeight int32  `json:"indexedBlockHeight"`
+	IndexedBlockHash   string `json:"indexedBlockHash"`
+}
+
+type deployResponse struct {
+	Id              string         `json:"id"`
+	Name            string         `json:"name"`
+	StartAt         time.Time      `json:"startAt"`
+	EndAt           time.Time      `json:"EndAt"`
+	Tiers           []tierResponse `json:"tiers"`
+	SellerPublicKey string         `json:"sellerPublicKey"`
+	MaxPerAddress   int32          `json:"maxPerAddress"`
+	DeployTxHash    string         `json:"deployTxHash"`
+}
+
+type tierResponse struct {
+	PriceSat      uint32 `json:"priceSat"`
+	Limit         uint32 `json:"limit"`
+	MaxPerAddress uint32 `json:"maxPerAddress"`
+	Sold          int64  `json:"sold"`
+}
+
+type nodeResponse struct {
+	DeployId            string `json:"deployId"`
+	NodeId              int32  `json:"nodeId"`
+	TierIndex           int32  `json:"tierIndex"`
+	DelegatedTo         string `json:"delegatedTo"`
+	OwnerPublicKey      string `json:"ownerPublicKey"`
+	PurchaseTxHash      string `json:"purchaseTxHash"`
+	DelegateTxHash      string `json:"delegateTxHash"`
+	PurchaseBlockHeight int32  `json:"purchaseBlockHeight"`
+}
+
+type eventResposne struct {
+	TxHash         string          `json:"txHash"`
+	BlockHeight    int32           `json:"blockHeight"`
+	TxIndex        int32           `json:"txIndex"`
+	WalletAddress  string          `json:"walletAddress"`
+	Action         string          `json:"action"`
+	ParsedMessage  json.RawMessage `json:"parsedMessage"`
+	BlockTimestamp time.Time       `json:"blockTimestamp"`
+	BlockHash      string          `json:"blockHash"`
+}

modules/nodesale/api/httphandler/routes.go

@@ -1 +1,16 @@
 package httphandler
+
+import (
+	"github.com/gofiber/fiber/v2"
+)
+
+func (h *handler) Mount(router fiber.Router) error {
+	r := router.Group("/nodesale/v1")
+
+	r.Get("/info", h.infoHandler)
+	r.Get("/deploy/:deployId", h.deployHandler)
+	r.Get("/nodes", h.nodesHandler)
+	r.Get("/events", h.eventsHandler)
+
+	return nil
+}

modules/nodesale/database/postgresql/migrations/000001_initialize.up.sql

@@ -21,6 +21,15 @@ CREATE TABLE IF NOT EXISTS events (
     "metadata" JSONB NOT NULL
 );
 
+INSERT INTO events("tx_hash", "block_height", "tx_index",
+                 "wallet_address", "valid", "action", 
+                "raw_message", "parsed_message", "block_timestamp",
+                "block_hash", "metadata")
+VALUES ('', -1, -1,
+        '', false, -1,
+        '', '{}', NOW(),
+        '', '{}');
+
 CREATE TABLE IF NOT EXISTS node_sales (
     "block_height" INTEGER NOT NULL,
     "tx_index" INTEGER NOT NULL,
@@ -40,10 +49,10 @@ CREATE TABLE IF NOT EXISTS nodes (
     "sale_tx_index" INTEGER NOT NULL,
     "node_id" INTEGER NOT NULL,
     "tier_index" INTEGER NOT NULL,
-    "delegated_to" TEXT,
+    "delegated_to" TEXT NOT NULL DEFAULT '',
     "owner_public_key" TEXT NOT NULL,
     "purchase_tx_hash" TEXT NOT NULL REFERENCES events(tx_hash) ON DELETE CASCADE,
-    "delegate_tx_hash" TEXT REFERENCES events(tx_hash) ON DELETE SET NULL,
+    "delegate_tx_hash" TEXT NOT NULL DEFAULT '' REFERENCES events(tx_hash) ON DELETE SET DEFAULT,
     PRIMARY KEY("sale_block", "sale_tx_index", "node_id"),
     FOREIGN KEY("sale_block", "sale_tx_index") REFERENCES node_sales("block_height", "tx_index")
 );

modules/nodesale/database/postgresql/queries/events.sql

@@ -5,4 +5,9 @@ WHERE "block_height" >= @from_block;
 -- name: AddEvent :exec
 INSERT INTO events("tx_hash", "block_height", "tx_index", "wallet_address", "valid", "action", 
                     "raw_message", "parsed_message", "block_timestamp", "block_hash", "metadata")
-VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11);
+VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11);
+
+-- name: GetEventsByWallet :many
+SELECT *
+FROM events
+WHERE wallet_address = $1;

modules/nodesale/database/postgresql/queries/nodes.sql

@@ -1,6 +1,6 @@
 -- name: ClearDelegate :execrows
 UPDATE nodes
-SET "delegated_to" = NULL
+SET "delegated_to" = ''
 WHERE "delegate_tx_hash" = NULL;
 
 -- name: SetDelegates :execrows
@@ -26,6 +26,26 @@ WHERE sale_block = $1 AND
     owner_public_key = $3
 ORDER BY tier_index;
 
+-- name: GetNodesByPubkey :many
+SELECT *
+FROM nodes JOIN events ON nodes.purchase_tx_hash = events.tx_hash
+WHERE sale_block = $1 AND
+    sale_tx_index = $2 AND
+    owner_public_key = $3 AND
+    delegated_to = $4;
+
 -- name: AddNode :exec
 INSERT INTO nodes(sale_block, sale_tx_index, node_id, tier_index, delegated_to, owner_public_key, purchase_tx_hash, delegate_tx_hash)
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8);
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8);
+
+-- name: GetNodeCountByTierIndex :many
+SELECT tiers.tier_index as tier_index, count(nodes.tier_index)
+FROM generate_series(@from_tier::int,@to_tier::int) as tiers(tier_index)
+LEFT JOIN 
+	(select * 
+	from nodes 
+	where sale_block = $1 and 
+		sale_tx_index= $2) 
+	as nodes on tiers.tier_index = nodes.tier_index 
+group by tiers.tier_index
+ORDER BY tiers.tier_index;

modules/nodesale/database/postgresql/queries/test.sql

@@ -1,2 +1,4 @@
 -- name: ClearEvents :exec
-DELETE FROM events;
+DELETE FROM events
+WHERE tx_hash <> '';
+;

modules/nodesale/delegate.go

@@ -33,17 +33,17 @@ func (p *Processor) processDelegate(ctx context.Context, qtx gen.Querier, block
 
 	if valid {
 		for _, node := range nodes {
-			decoded, err := hex.DecodeString(node.OwnerPublicKey)
+			OwnerPublicKeyBytes, err := hex.DecodeString(node.OwnerPublicKey)
 			if err != nil {
 				valid = false
 				break
 			}
-			pubkey, err := btcec.ParsePubKey(decoded)
+			OwnerPublicKey, err := btcec.ParsePubKey(OwnerPublicKeyBytes)
 			if err != nil {
 				valid = false
 				break
 			}
-			if !event.txPubkey.IsEqual(pubkey) {
+			if !event.txPubkey.IsEqual(OwnerPublicKey) {
 				valid = false
 				break
 			}
@@ -71,11 +71,8 @@ func (p *Processor) processDelegate(ctx context.Context, qtx gen.Querier, block
 		_, err = qtx.SetDelegates(ctx, gen.SetDelegatesParams{
 			SaleBlock:   int32(delegate.DeployID.Block),
 			SaleTxIndex: int32(delegate.DeployID.TxIndex),
-			Delegatee: pgtype.Text{
-				String: delegate.DelegateePublicKey,
-				Valid:  true,
-			},
-			NodeIds: nodeIds,
+			Delegatee:   delegate.DelegateePublicKey,
+			NodeIds:     nodeIds,
 		})
 		if err != nil {
 			return fmt.Errorf("Failed to set delegate : %w", err)

modules/nodesale/delegate_test.go

@@ -116,10 +116,9 @@ func TestDelegate(t *testing.T) {
 	require.Len(t, nodes, 3)
 	for _, node := range nodes {
 		if node.NodeID == 9 || node.NodeID == 10 {
-			require.True(t, node.DelegatedTo.Valid)
-			require.Equal(t, delegateePubkeyHex, node.DelegatedTo.String)
+			require.NotEmpty(t, node.DelegatedTo)
 		} else if node.NodeID == 11 {
-			require.False(t, node.DelegatedTo.Valid)
+			require.Empty(t, node.DelegatedTo)
 		} else {
 			require.Fail(t, "Unhandled")
 		}

modules/nodesale/deploy.go

@@ -25,17 +25,17 @@ func (p *Processor) processDeploy(ctx context.Context, qtx gen.Querier, block *t
 		) {
 			valid = false
 		}*/
-	decoded, err := hex.DecodeString(deploy.SellerPublicKey)
+	sellerPubKeyBytes, err := hex.DecodeString(deploy.SellerPublicKey)
 	if err != nil {
 		valid = false
 	}
 
 	if valid {
-		pubkey, err := btcec.ParsePubKey(decoded)
+		sellerPubKey, err := btcec.ParsePubKey(sellerPubKeyBytes)
 		if err != nil {
 			valid = false
 		}
-		if valid && !event.txPubkey.IsEqual(pubkey) {
+		if valid && !event.txPubkey.IsEqual(sellerPubKey) {
 			valid = false
 		}
 	}

modules/nodesale/nodesale.go

@@ -9,8 +9,10 @@ import (
 	"github.com/gaze-network/indexer-network/core/indexer"
 	"github.com/gaze-network/indexer-network/internal/config"
 	"github.com/gaze-network/indexer-network/internal/postgres"
+	"github.com/gaze-network/indexer-network/modules/nodesale/api/httphandler"
 	repository "github.com/gaze-network/indexer-network/modules/nodesale/repository/postgres"
 	"github.com/gaze-network/indexer-network/pkg/logger"
+	"github.com/gofiber/fiber/v2"
 	"github.com/samber/do/v2"
 )
 
@@ -45,6 +47,14 @@ func New(injector do.Injector) (indexer.IndexerWorker, error) {
 		network:      conf.Network,
 		cleanupFuncs: cleanupFuncs,
 	}
+
+	httpServer := do.MustInvoke[*fiber.App](injector)
+	nodesaleHandler := httphandler.New(repository)
+	if err := nodesaleHandler.Mount(httpServer); err != nil {
+		return nil, fmt.Errorf("Can't mount nodesale API : %w", err)
+	}
+	logger.InfoContext(ctx, "Mounted nodesale HTTP handler")
+
 	indexer := indexer.New(processor, datasource)
 	logger.InfoContext(ctx, "Nodesale module started.")
 	return indexer, nil

modules/nodesale/purchase.go

@@ -31,17 +31,17 @@ func (p *Processor) processPurchase(ctx context.Context, qtx gen.Querier, block
 	purchase := event.eventMessage.Purchase
 	payload := purchase.Payload
 
-	decoded, err := hex.DecodeString(payload.BuyerPublicKey)
+	buyerPubkeyBytes, err := hex.DecodeString(payload.BuyerPublicKey)
 	if err != nil {
 		valid = false
 	}
 
 	if valid {
-		pubkey, err := btcec.ParsePubKey(decoded)
+		buyerPubkey, err := btcec.ParsePubKey(buyerPubkeyBytes)
 		if err != nil {
 			valid = false
 		}
-		if valid && !event.txPubkey.IsEqual(pubkey) {
+		if valid && !event.txPubkey.IsEqual(buyerPubkey) {
 			valid = false
 		}
 	}
@@ -256,10 +256,10 @@ func (p *Processor) processPurchase(ctx context.Context, qtx gen.Querier, block
 				SaleTxIndex:    deploy.TxIndex,
 				NodeID:         int32(nodeId),
 				TierIndex:      nodeIdToTier[nodeId],
-				DelegatedTo:    pgtype.Text{Valid: false},
+				DelegatedTo:    "",
 				OwnerPublicKey: payload.BuyerPublicKey,
 				PurchaseTxHash: event.transaction.TxHash.String(),
-				DelegateTxHash: pgtype.Text{Valid: false},
+				DelegateTxHash: "",
 			})
 			if err != nil {
 				return fmt.Errorf("Failed to insert node : %w", err)

modules/nodesale/repository/postgres/gen/events.sql.go

@@ -48,6 +48,44 @@ func (q *Queries) AddEvent(ctx context.Context, arg AddEventParams) error {
 	return err
 }
 
+const getEventsByWallet = `-- name: GetEventsByWallet :many
+SELECT tx_hash, block_height, tx_index, wallet_address, valid, action, raw_message, parsed_message, block_timestamp, block_hash, metadata
+FROM events
+WHERE wallet_address = $1
+`
+
+func (q *Queries) GetEventsByWallet(ctx context.Context, walletAddress string) ([]Event, error) {
+	rows, err := q.db.Query(ctx, getEventsByWallet, walletAddress)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []Event
+	for rows.Next() {
+		var i Event
+		if err := rows.Scan(
+			&i.TxHash,
+			&i.BlockHeight,
+			&i.TxIndex,
+			&i.WalletAddress,
+			&i.Valid,
+			&i.Action,
+			&i.RawMessage,
+			&i.ParsedMessage,
+			&i.BlockTimestamp,
+			&i.BlockHash,
+			&i.Metadata,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const removeEventsFromBlock = `-- name: RemoveEventsFromBlock :execrows
 DELETE FROM events
 WHERE "block_height" >= $1

modules/nodesale/repository/postgres/gen/models.go

@@ -33,10 +33,10 @@ type Node struct {
 	SaleTxIndex    int32
 	NodeID         int32
 	TierIndex      int32
-	DelegatedTo    pgtype.Text
+	DelegatedTo    string
 	OwnerPublicKey string
 	PurchaseTxHash string
-	DelegateTxHash pgtype.Text
+	DelegateTxHash string
 }
 
 type NodeSale struct {

modules/nodesale/repository/postgres/gen/nodes.sql.go

@@ -21,10 +21,10 @@ type AddNodeParams struct {
 	SaleTxIndex    int32
 	NodeID         int32
 	TierIndex      int32
-	DelegatedTo    pgtype.Text
+	DelegatedTo    string
 	OwnerPublicKey string
 	PurchaseTxHash string
-	DelegateTxHash pgtype.Text
+	DelegateTxHash string
 }
 
 func (q *Queries) AddNode(ctx context.Context, arg AddNodeParams) error {
@@ -43,7 +43,7 @@ func (q *Queries) AddNode(ctx context.Context, arg AddNodeParams) error {
 
 const clearDelegate = `-- name: ClearDelegate :execrows
 UPDATE nodes
-SET "delegated_to" = NULL
+SET "delegated_to" = ''
 WHERE "delegate_tx_hash" = NULL
 `
 
@@ -55,6 +55,56 @@ func (q *Queries) ClearDelegate(ctx context.Context) (int64, error) {
 	return result.RowsAffected(), nil
 }
 
+const getNodeCountByTierIndex = `-- name: GetNodeCountByTierIndex :many
+SELECT tiers.tier_index as tier_index, count(nodes.tier_index)
+FROM generate_series($3::int,$4::int) as tiers(tier_index)
+LEFT JOIN 
+	(select sale_block, sale_tx_index, node_id, tier_index, delegated_to, owner_public_key, purchase_tx_hash, delegate_tx_hash 
+	from nodes 
+	where sale_block = $1 and 
+		sale_tx_index= $2) 
+	as nodes on tiers.tier_index = nodes.tier_index 
+group by tiers.tier_index
+ORDER BY tiers.tier_index
+`
+
+type GetNodeCountByTierIndexParams struct {
+	SaleBlock   int32
+	SaleTxIndex int32
+	FromTier    int32
+	ToTier      int32
+}
+
+type GetNodeCountByTierIndexRow struct {
+	TierIndex interface{}
+	Count     int64
+}
+
+func (q *Queries) GetNodeCountByTierIndex(ctx context.Context, arg GetNodeCountByTierIndexParams) ([]GetNodeCountByTierIndexRow, error) {
+	rows, err := q.db.Query(ctx, getNodeCountByTierIndex,
+		arg.SaleBlock,
+		arg.SaleTxIndex,
+		arg.FromTier,
+		arg.ToTier,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetNodeCountByTierIndexRow
+	for rows.Next() {
+		var i GetNodeCountByTierIndexRow
+		if err := rows.Scan(&i.TierIndex, &i.Count); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getNodes = `-- name: GetNodes :many
 SELECT sale_block, sale_tx_index, node_id, tier_index, delegated_to, owner_public_key, purchase_tx_hash, delegate_tx_hash
 FROM nodes
@@ -142,6 +192,89 @@ func (q *Queries) GetNodesByOwner(ctx context.Context, arg GetNodesByOwnerParams
 	return items, nil
 }
 
+const getNodesByPubkey = `-- name: GetNodesByPubkey :many
+SELECT sale_block, sale_tx_index, node_id, tier_index, delegated_to, owner_public_key, purchase_tx_hash, delegate_tx_hash, tx_hash, block_height, tx_index, wallet_address, valid, action, raw_message, parsed_message, block_timestamp, block_hash, metadata
+FROM nodes JOIN events ON nodes.purchase_tx_hash = events.tx_hash
+WHERE sale_block = $1 AND
+    sale_tx_index = $2 AND
+    owner_public_key = $3 AND
+    delegated_to = $4
+`
+
+type GetNodesByPubkeyParams struct {
+	SaleBlock      int32
+	SaleTxIndex    int32
+	OwnerPublicKey string
+	DelegatedTo    string
+}
+
+type GetNodesByPubkeyRow struct {
+	SaleBlock      int32
+	SaleTxIndex    int32
+	NodeID         int32
+	TierIndex      int32
+	DelegatedTo    string
+	OwnerPublicKey string
+	PurchaseTxHash string
+	DelegateTxHash string
+	TxHash         string
+	BlockHeight    int32
+	TxIndex        int32
+	WalletAddress  string
+	Valid          bool
+	Action         int32
+	RawMessage     []byte
+	ParsedMessage  []byte
+	BlockTimestamp pgtype.Timestamp
+	BlockHash      string
+	Metadata       []byte
+}
+
+func (q *Queries) GetNodesByPubkey(ctx context.Context, arg GetNodesByPubkeyParams) ([]GetNodesByPubkeyRow, error) {
+	rows, err := q.db.Query(ctx, getNodesByPubkey,
+		arg.SaleBlock,
+		arg.SaleTxIndex,
+		arg.OwnerPublicKey,
+		arg.DelegatedTo,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetNodesByPubkeyRow
+	for rows.Next() {
+		var i GetNodesByPubkeyRow
+		if err := rows.Scan(
+			&i.SaleBlock,
+			&i.SaleTxIndex,
+			&i.NodeID,
+			&i.TierIndex,
+			&i.DelegatedTo,
+			&i.OwnerPublicKey,
+			&i.PurchaseTxHash,
+			&i.DelegateTxHash,
+			&i.TxHash,
+			&i.BlockHeight,
+			&i.TxIndex,
+			&i.WalletAddress,
+			&i.Valid,
+			&i.Action,
+			&i.RawMessage,
+			&i.ParsedMessage,
+			&i.BlockTimestamp,
+			&i.BlockHash,
+			&i.Metadata,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const setDelegates = `-- name: SetDelegates :execrows
 UPDATE nodes
 SET delegated_to = $3
@@ -153,7 +286,7 @@ WHERE sale_block = $1 AND
 type SetDelegatesParams struct {
 	SaleBlock   int32
 	SaleTxIndex int32
-	Delegatee   pgtype.Text
+	Delegatee   string
 	NodeIds     []int32
 }
 

modules/nodesale/repository/postgres/gen/querier.go

@@ -16,9 +16,12 @@ type Querier interface {
 	ClearDelegate(ctx context.Context) (int64, error)
 	ClearEvents(ctx context.Context) error
 	GetBlock(ctx context.Context, blockHeight int32) (Block, error)
+	GetEventsByWallet(ctx context.Context, walletAddress string) ([]Event, error)
 	GetLastProcessedBlock(ctx context.Context) (Block, error)
+	GetNodeCountByTierIndex(ctx context.Context, arg GetNodeCountByTierIndexParams) ([]GetNodeCountByTierIndexRow, error)
 	GetNodes(ctx context.Context, arg GetNodesParams) ([]Node, error)
 	GetNodesByOwner(ctx context.Context, arg GetNodesByOwnerParams) ([]Node, error)
+	GetNodesByPubkey(ctx context.Context, arg GetNodesByPubkeyParams) ([]GetNodesByPubkeyRow, error)
 	GetNodesale(ctx context.Context, arg GetNodesaleParams) ([]NodeSale, error)
 	RemoveBlockFrom(ctx context.Context, fromBlock int32) (int64, error)
 	RemoveEventsFromBlock(ctx context.Context, fromBlock int32) (int64, error)

modules/nodesale/repository/postgres/gen/test.sql.go

@@ -11,6 +11,7 @@ import (
 
 const clearEvents = `-- name: ClearEvents :exec
 DELETE FROM events
+WHERE tx_hash <> ''
 `
 
 func (q *Queries) ClearEvents(ctx context.Context) error {