Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 70 additions & 0 deletions internal/cdptest/ledgerlib.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
package cdptest

import (
"fmt"
"time"

"github.com/stellar/go/xdr"
"github.com/stellar/stellar-etl/internal/utils"
)

func LedgerSequence(lcm xdr.LedgerCloseMeta) (*uint32, error) {
ledgerSequence := lcm.LedgerSequence()
return &ledgerSequence, nil
}

func CloseTime(lcm xdr.LedgerCloseMeta) (*time.Time, error) {
ledgerHeader := lcm.LedgerHeaderHistoryEntry()
closeTime, err := utils.TimePointToUTCTimeStamp(ledgerHeader.Header.ScpValue.CloseTime)
if err != nil {
return nil, err
}

return &closeTime, nil
}

func BaseFee(lcm xdr.LedgerCloseMeta) (*uint32, error) {
ledgerHeader := lcm.LedgerHeaderHistoryEntry()
baseFee := uint32(ledgerHeader.Header.BaseFee)
return &baseFee, nil
}

func BaseReserve(lcm xdr.LedgerCloseMeta) (*uint32, error) {
ledgerHeader := lcm.LedgerHeaderHistoryEntry()
baseReserve := uint32(ledgerHeader.Header.BaseReserve)
return &baseReserve, nil
}

func SorobanFeeWrite1Kb(lcm xdr.LedgerCloseMeta) (*int64, error) {
switch lcm.V {
case 0:
return nil, nil
case 1:
lcmV1Ext := lcm.MustV1().Ext
switch lcmV1Ext.V {
case 0:
return nil, nil
case 1:
ext := lcmV1Ext.MustV1()
sorobanFreeWrite1Kb := int64(ext.SorobanFeeWrite1Kb)
return &sorobanFreeWrite1Kb, nil
default:
panic(fmt.Errorf("unsupported LedgerCloseMeta.V1.Ext.V: %d", lcmV1Ext.V))
}
default:
panic(fmt.Errorf("unsupported LedgerCloseMeta.V: %d", lcm.V))
}
}

func TotalByteSizeOfBucketList(lcm xdr.LedgerCloseMeta) (*uint64, error) {
switch lcm.V {
case 0:
return nil, nil
case 1:
lcmV1 := lcm.MustV1()
totalByteSizeOfBucketList := uint64(lcmV1.TotalByteSizeOfBucketList)
return &totalByteSizeOfBucketList, nil
default:
panic(fmt.Sprintf("Unsupported LedgerCloseMeta.V: %d", lcm.V))
}
}
3 changes: 3 additions & 0 deletions internal/cdptest/ledgersprocessor.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
{
"functions": ["LedgerSequence", "CloseTime", "BaseFee", "SorobanFeeWrite1Kb"]
}
31 changes: 31 additions & 0 deletions internal/cdptest/transactionlib.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
package cdptest

import (
"github.com/stellar/go/ingest"
"github.com/stellar/go/xdr"
"github.com/stellar/stellar-etl/internal/utils"
)

func TransactionHash(tx ingest.LedgerTransaction) (*string, error) {
transactionHash := utils.HashToHexString(tx.Result.TransactionHash)
return &transactionHash, nil
}

func Account(tx ingest.LedgerTransaction) (*string, error) {
account, err := utils.GetAccountAddressFromMuxedAccount(tx.Envelope.SourceAccount())
if err != nil {
return nil, err
}

return &account, nil

}

func TransactionEnvelope(tx ingest.LedgerTransaction) (*string, error) {
transactionEnvelope, err := xdr.MarshalBase64(tx.Envelope)
if err != nil {
return nil, err
}

return &transactionEnvelope, nil
}
3 changes: 3 additions & 0 deletions internal/cdptest/transactionprocessor.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
{
"functions": ["LedgerSequence", "CloseTime", "TransactionHash", "Account", "TransactionEnvelope"]
}
81 changes: 81 additions & 0 deletions internal/cdptestprocessor/genericprocessor.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
package cdptest

import (
"io"
"maps"

"github.com/stellar/go/ingest"
"github.com/stellar/go/xdr"
"github.com/stellar/stellar-etl/internal/cdptest"
)

var ledgerFunctionMap = map[string]interface{}{
"LedgerSequence": cdptest.LedgerSequence,
"CloseTime": cdptest.CloseTime,
"BaseFee": cdptest.BaseFee,
"BaseReserve": cdptest.BaseReserve,
"SorobanFeeWrite1Kb": cdptest.SorobanFeeWrite1Kb,
"TotalByteSizeOfBucketList": cdptest.TotalByteSizeOfBucketList,
}

var transactionFunctionMap = map[string]interface{}{
"TransactionHash": cdptest.TransactionHash,
"Account": cdptest.Account,
"TransactionEnvelope": cdptest.TransactionEnvelope,
}

func GenericProcessor(lcm xdr.LedgerCloseMeta, processorJSON string) ([]map[string]interface{}, error) {
// result can be any number of key:val
var ledgerResult map[string]interface{}

// Get the functions from the *processor.json file
funcs := make(map[string][]string)

// LedgerFunctions are single value
for _, funcName := range funcs["functions"] {
fn, ok := ledgerFunctionMap[funcName]
if ok {
f := fn.(func(xdr.LedgerCloseMeta) interface{})
ledgerResult[funcName] = f(lcm)
}
}

hasTx := false
for _, key := range funcs["functions"] {
_, ok := transactionFunctionMap[key]
if ok {
hasTx = true
}
}

if !hasTx {
var results []map[string]interface{}
results = append(results, ledgerResult)
return results, nil
}

// There can be multiple tx in LCM
var transactionResult []map[string]interface{}

txReader, _ := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta("password", lcm)
for {
tx, err := txReader.Read()
if err == io.EOF {
break
}

var result map[string]interface{}
for _, funcName := range funcs["functions"] {

fn, ok := transactionFunctionMap[funcName]
if ok {
f := fn.(func(ingest.LedgerTransaction) interface{})
result[funcName] = f(tx)
}
}
maps.Copy(result, ledgerResult)
transactionResult = append(transactionResult, result)
}

return transactionResult, nil
}