feat: add flush brc20 states

modules/brc20/database/postgresql/queries/data.sql

@@ -105,6 +105,9 @@ INSERT INTO "brc20_event_inscribe_transfers" ("inscription_id", "inscription_num
 -- name: CreateEventTransferTransfers :batchexec
 INSERT INTO "brc20_event_transfer_transfers" ("inscription_id", "inscription_number", "tick", "original_tick", "tx_hash", "block_height", "tx_index", "timestamp", "from_pkscript", "from_satpoint", "from_input_index", "to_pkscript", "to_satpoint", "to_output_index", "spent_as_fee", "amount") VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16);
 
+-- name: CreateBalances :batchexec
+INSERT INTO "brc20_balances" ("pkscript", "block_height", "tick", "overall_balance", "available_balance") VALUES ($1, $2, $3, $4, $5);
+
 -- name: DeleteIndexedBlocksSinceHeight :exec
 DELETE FROM "brc20_indexed_blocks" WHERE "height" >= $1;
 

modules/brc20/internal/datagateway/brc20.go

@@ -48,6 +48,7 @@ type BRC20WriterDataGateway interface {
 	CreateEventMints(ctx context.Context, events []*entity.EventMint) error
 	CreateEventInscribeTransfers(ctx context.Context, events []*entity.EventInscribeTransfer) error
 	CreateEventTransferTransfers(ctx context.Context, events []*entity.EventTransferTransfer) error
+	CreateBalances(ctx context.Context, balances []*entity.Balance) error
 
 	// used for revert data
 	DeleteIndexedBlocksSinceHeight(ctx context.Context, height uint64) error

modules/brc20/internal/repository/postgres/brc20.go

@@ -440,6 +440,23 @@ func (r *Repository) CreateEventTransferTransfers(ctx context.Context, events []
 	return nil
 }
 
+func (r *Repository) CreateBalances(ctx context.Context, balances []*entity.Balance) error {
+	params := lo.Map(balances, func(balance *entity.Balance, _ int) gen.CreateBalancesParams {
+		return mapBalanceTypeToParams(*balance)
+	})
+	results := r.queries.CreateBalances(ctx, params)
+	var execErrors []error
+	results.Exec(func(i int, err error) {
+		if err != nil {
+			execErrors = append(execErrors, err)
+		}
+	})
+	if len(execErrors) > 0 {
+		return errors.Wrap(errors.Join(execErrors...), "error during exec")
+	}
+	return nil
+}
+
 func (r *Repository) DeleteIndexedBlocksSinceHeight(ctx context.Context, height uint64) error {
 	if err := r.queries.DeleteIndexedBlocksSinceHeight(ctx, int32(height)); err != nil {
 		return errors.Wrap(err, "error during exec")

modules/brc20/internal/repository/postgres/gen/batch.go

@@ -17,6 +17,61 @@ var (
 	ErrBatchAlreadyClosed = errors.New("batch already closed")
 )
 
+const createBalances = `-- name: CreateBalances :batchexec
+INSERT INTO "brc20_balances" ("pkscript", "block_height", "tick", "overall_balance", "available_balance") VALUES ($1, $2, $3, $4, $5)
+`
+
+type CreateBalancesBatchResults struct {
+	br     pgx.BatchResults
+	tot    int
+	closed bool
+}
+
+type CreateBalancesParams struct {
+	Pkscript         string
+	BlockHeight      int32
+	Tick             string
+	OverallBalance   pgtype.Numeric
+	AvailableBalance pgtype.Numeric
+}
+
+func (q *Queries) CreateBalances(ctx context.Context, arg []CreateBalancesParams) *CreateBalancesBatchResults {
+	batch := &pgx.Batch{}
+	for _, a := range arg {
+		vals := []interface{}{
+			a.Pkscript,
+			a.BlockHeight,
+			a.Tick,
+			a.OverallBalance,
+			a.AvailableBalance,
+		}
+		batch.Queue(createBalances, vals...)
+	}
+	br := q.db.SendBatch(ctx, batch)
+	return &CreateBalancesBatchResults{br, len(arg), false}
+}
+
+func (b *CreateBalancesBatchResults) Exec(f func(int, error)) {
+	defer b.br.Close()
+	for t := 0; t < b.tot; t++ {
+		if b.closed {
+			if f != nil {
+				f(t, ErrBatchAlreadyClosed)
+			}
+			continue
+		}
+		_, err := b.br.Exec()
+		if f != nil {
+			f(t, err)
+		}
+	}
+}
+
+func (b *CreateBalancesBatchResults) Close() error {
+	b.closed = true
+	return b.br.Close()
+}
+
 const createEventDeploys = `-- name: CreateEventDeploys :batchexec
 INSERT INTO "brc20_event_deploys" ("inscription_id", "inscription_number", "tick", "original_tick", "tx_hash", "block_height", "tx_index", "timestamp", "pkscript", "satpoint", "total_supply", "decimals", "limit_per_mint", "is_self_mint") VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
 `

modules/brc20/internal/repository/postgres/mapper.go

@@ -602,3 +602,13 @@ func mapBalanceModelToType(src gen.Brc20Balance) (entity.Balance, error) {
 		AvailableBalance: decimalFromNumeric(src.AvailableBalance).Decimal,
 	}, nil
 }
+
+func mapBalanceTypeToParams(src entity.Balance) gen.CreateBalancesParams {
+	return gen.CreateBalancesParams{
+		Pkscript:         hex.EncodeToString(src.PkScript),
+		Tick:             src.Tick,
+		BlockHeight:      int32(src.BlockHeight),
+		OverallBalance:   numericFromDecimal(src.OverallBalance),
+		AvailableBalance: numericFromDecimal(src.AvailableBalance),
+	}
+}

modules/brc20/processor.go

@@ -51,7 +51,7 @@ type Processor struct {
 	newEventMints             []*entity.EventMint
 	newEventInscribeTransfers []*entity.EventInscribeTransfer
 	newEventTransferTransfers []*entity.EventTransferTransfer
-	newBalances               map[string]map[string]*entity.Balance
+	newBalances               map[string]map[string]*entity.Balance // pkscript -> tick -> balance
 
 	eventHashString string
 }

modules/brc20/processor_process.go

@@ -108,6 +108,7 @@ func (p *Processor) flushBlock(ctx context.Context, blockHeader types.BlockHeade
 		}); err != nil {
 			return errors.Wrap(err, "failed to create indexed block")
 		}
+		p.eventHashString = ""
 	}
 
 	// flush new inscription entries
@@ -148,6 +149,65 @@ func (p *Processor) flushBlock(ctx context.Context, blockHeader types.BlockHeade
 			return errors.Wrap(err, "failed to create processor stats")
 		}
 	}
+	// newTickEntries            map[string]*entity.TickEntry
+	// newTickEntryStates        map[string]*entity.TickEntry
+	// newEventDeploys           []*entity.EventDeploy
+	// newEventMints             []*entity.EventMint
+	// newEventInscribeTransfers []*entity.EventInscribeTransfer
+	// newEventTransferTransfers []*entity.EventTransferTransfer
+	// newBalances               map[string]map[string]*entity.Balance
+
+	// flush new tick entries
+	{
+		newTickEntries := lo.Values(p.newTickEntries)
+		if err := brc20DgTx.CreateTickEntries(ctx, blockHeight, newTickEntries); err != nil {
+			return errors.Wrap(err, "failed to create tick entries")
+		}
+		p.newTickEntries = make(map[string]*entity.TickEntry)
+	}
+
+	// flush new tick entry states
+	{
+		newTickEntryStates := lo.Values(p.newTickEntryStates)
+		if err := brc20DgTx.CreateTickEntryStates(ctx, blockHeight, newTickEntryStates); err != nil {
+			return errors.Wrap(err, "failed to create tick entry states")
+		}
+		p.newTickEntryStates = make(map[string]*entity.TickEntry)
+	}
+
+	// flush new events
+	{
+		if err := brc20DgTx.CreateEventDeploys(ctx, p.newEventDeploys); err != nil {
+			return errors.Wrap(err, "failed to create event deploys")
+		}
+		if err := brc20DgTx.CreateEventMints(ctx, p.newEventMints); err != nil {
+			return errors.Wrap(err, "failed to create event mints")
+		}
+		if err := brc20DgTx.CreateEventInscribeTransfers(ctx, p.newEventInscribeTransfers); err != nil {
+			return errors.Wrap(err, "failed to create event inscribe transfers")
+		}
+		if err := brc20DgTx.CreateEventTransferTransfers(ctx, p.newEventTransferTransfers); err != nil {
+			return errors.Wrap(err, "failed to create event transfer transfers")
+		}
+		p.newEventDeploys = make([]*entity.EventDeploy, 0)
+		p.newEventMints = make([]*entity.EventMint, 0)
+		p.newEventInscribeTransfers = make([]*entity.EventInscribeTransfer, 0)
+		p.newEventTransferTransfers = make([]*entity.EventTransferTransfer, 0)
+	}
+
+	// flush new balances
+	{
+		newBalances := make([]*entity.Balance, 0)
+		for _, tickBalances := range p.newBalances {
+			for _, balance := range tickBalances {
+				newBalances = append(newBalances, balance)
+			}
+		}
+		if err := brc20DgTx.CreateBalances(ctx, newBalances); err != nil {
+			return errors.Wrap(err, "failed to create balances")
+		}
+		p.newBalances = make(map[string]map[string]*entity.Balance)
+	}
 
 	if err := brc20DgTx.Commit(ctx); err != nil {
 		return errors.Wrap(err, "failed to commit transaction")