exp/ingest: Ingest Session

exp/ingest/adapters/history_archive_adapter.go

@@ -1,4 +1,4 @@
-package ingestadapters
+package adapters
 
 import (
 	"fmt"

exp/ingest/adapters/history_archive_adapter_test.go

@@ -1,4 +1,4 @@
-package ingestadapters
+package adapters
 
 import (
 	"fmt"

exp/ingest/adapters/ledger_backend_adapter.go

@@ -1,4 +1,4 @@
-package ingestadapters
+package adapters
 
 import (
 	"errors"
@@ -28,8 +28,7 @@ func (lba *LedgerBackendAdapter) GetLedger(sequence uint32) (io.LedgerReadCloser
 		return nil, errors.New(noBackendErr)
 	}
 
-	dblrc := io.MakeLedgerReadCloser(sequence, lba.Backend)
-	return dblrc, nil
+	return io.MakeLedgerReadCloser(sequence, lba.Backend)
 }
 
 // Close shuts down the provided backend.

exp/ingest/adapters/ledger_backend_adapter_test.go

@@ -1,4 +1,4 @@
-package ingestadapters
+package adapters
 
 import (
 	"testing"

exp/ingest/cmd/main.go

@@ -42,10 +42,7 @@ func useAdapter() {
 	s := lrc.GetSequence()
 	fmt.Println("lrc sequence:", s)
 
-	h, err := lrc.GetHeader()
-	if err != nil {
-		log.Fatal(err)
-	}
+	h := lrc.GetHeader()
 	fmt.Println("lrc header:", h)
 
 	for {

exp/ingest/io/ledger_read_closer.go

@@ -9,15 +9,6 @@ import (
 	"github.com/stellar/go/xdr"
 )
 
-// LedgerTransaction represents the data for a single transaction within a ledger.
-type LedgerTransaction struct {
-	Index      uint32
-	Envelope   xdr.TransactionEnvelope
-	Result     xdr.TransactionResultPair
-	Meta       xdr.TransactionMeta
-	FeeChanges xdr.LedgerEntryChanges
-}
-
 // DBLedgerReadCloser is a database-backed implementation of the io.LedgerReadCloser interface.
 type DBLedgerReadCloser struct {
 	sequence     uint32
@@ -33,11 +24,19 @@ type DBLedgerReadCloser struct {
 var _ LedgerReadCloser = (*DBLedgerReadCloser)(nil)
 
 // MakeLedgerReadCloser is a factory method for LedgerReadCloser.
-func MakeLedgerReadCloser(sequence uint32, backend ledgerbackend.LedgerBackend) *DBLedgerReadCloser {
-	return &DBLedgerReadCloser{
+func MakeLedgerReadCloser(sequence uint32, backend ledgerbackend.LedgerBackend) (*DBLedgerReadCloser, error) {
+	reader := &DBLedgerReadCloser{
 		sequence: sequence,
 		backend:  backend,
 	}
+
+	var err error
+	reader.initOnce.Do(func() { err = reader.init() })
+	if err != nil {
+		return nil, err
+	}
+
+	return reader, nil
 }
 
 // GetSequence returns the sequence number of the ledger data stored by this object.
@@ -46,13 +45,15 @@ func (dblrc *DBLedgerReadCloser) GetSequence() uint32 {
 }
 
 // GetHeader returns the XDR Header data associated with the stored ledger.
-func (dblrc *DBLedgerReadCloser) GetHeader() (xdr.LedgerHeaderHistoryEntry, error) {
+func (dblrc *DBLedgerReadCloser) GetHeader() xdr.LedgerHeaderHistoryEntry {
 	var err error
 	dblrc.initOnce.Do(func() { err = dblrc.init() })
 	if err != nil {
-		return xdr.LedgerHeaderHistoryEntry{}, err
+		// TODO, object should be initialized in constructor.
+		// Not returning error here, makes this much simpler.
+		panic(err)
 	}
-	return dblrc.header, nil
+	return dblrc.header
 }
 
 // Read returns the next transaction in the ledger, ordered by tx number, each time it is called. When there
@@ -92,7 +93,7 @@ func (dblrc *DBLedgerReadCloser) init() error {
 		return errors.Wrap(err, "error reading ledger from backend")
 	}
 	if !exists {
-		return errors.Wrap(err, "ledger was not found")
+		return ErrNotFound
 	}
 
 	dblrc.header = ledgerCloseMeta.LedgerHeader

exp/ingest/io/ledger_transaction.go

@@ -0,0 +1,100 @@
+package io
+
+import (
+	"github.com/stellar/go/xdr"
+)
+
+// Change is a developer friendly representation of LedgerEntryChanges.
+// It also provides some helper functions to quickly check if a given
+// change has occured in an entry.
+//
+// If an entry is created: Pre is nil and Post is not nil.
+// If an entry is updated: Pre is not nil and Post is not nil.
+// If an entry is removed: Pre is not nil and Post is nil.
+type Change struct {
+	Type xdr.LedgerEntryType
+	Pre  *xdr.LedgerEntryData
+	Post *xdr.LedgerEntryData
+}
+
+func (c *Change) AccountSignersChanged() bool {
+	if c.Type != xdr.LedgerEntryTypeAccount {
+		panic("This should not be called on changes other than Account changes")
+	}
+
+	// Signers must be removed before merging an account and it's
+	// impossible to add signers during a creation of a new account.
+	if c.Pre == nil || c.Post == nil {
+		return false
+	}
+
+	if len(c.Pre.MustAccount().Signers) != len(c.Post.MustAccount().Signers) {
+		return true
+	}
+
+	signers := map[string]uint32{} // signer => weight
+
+	for _, signer := range c.Pre.MustAccount().Signers {
+		signers[signer.Key.Address()] = uint32(signer.Weight)
+	}
+
+	for _, signer := range c.Post.MustAccount().Signers {
+		weight, exist := signers[signer.Key.Address()]
+		if !exist {
+			return false
+		}
+
+		if weight != uint32(signer.Weight) {
+			return false
+		}
+	}
+
+	// TODO should it also change on master key weight changes?
+
+	return false
+}
+
+// GetChanges returns a developer friendly representation of LedgerEntryChanges.
+// Currently it results operations related LedgerEntryChanges only.
+// TODO this should include TransactionMetaV1.txChanges and fee changes too!
+func (t *LedgerTransaction) GetChanges() []Change {
+	changes := []Change{}
+
+	for _, operationMeta := range t.Meta.OperationsMeta() {
+		ledgerEntryChanges := operationMeta.Changes
+		for i := 0; i < len(ledgerEntryChanges); i++ {
+			entryChange := ledgerEntryChanges[i]
+
+			switch entryChange.Type {
+			case xdr.LedgerEntryChangeTypeLedgerEntryCreated:
+				created := entryChange.MustCreated()
+				changes = append(changes, Change{
+					Type: created.Data.Type,
+					Pre:  nil,
+					Post: &created.Data,
+				})
+			case xdr.LedgerEntryChangeTypeLedgerEntryUpdated:
+				state := ledgerEntryChanges[i-1].MustState()
+				updated := entryChange.MustUpdated()
+				changes = append(changes, Change{
+					Type: state.Data.Type,
+					Pre:  &state.Data,
+					Post: &updated.Data,
+				})
+			case xdr.LedgerEntryChangeTypeLedgerEntryRemoved:
+				state := ledgerEntryChanges[i-1].MustState()
+				changes = append(changes, Change{
+					Type: state.Data.Type,
+					Pre:  &state.Data,
+					Post: nil,
+				})
+			case xdr.LedgerEntryChangeTypeLedgerEntryState:
+				continue
+			default:
+				panic("Invalid LedgerEntryChangeType")
+			}
+		}
+	}
+
+	return changes
+}

exp/ingest/io/main.go

@@ -1,9 +1,12 @@
 package io
 
 import (
+	"github.com/stellar/go/support/errors"
 	"github.com/stellar/go/xdr"
 )
 
+var ErrNotFound = errors.New("not found")
+
 // StateReadCloser interface placeholder
 type StateReadCloser interface {
 	GetSequence() uint32
@@ -32,21 +35,35 @@ type StateWriteCloser interface {
 // LedgerReadCloser provides convenient, streaming access to the transactions within a ledger.
 type LedgerReadCloser interface {
 	GetSequence() uint32
-	GetHeader() (xdr.LedgerHeaderHistoryEntry, error)
+	GetHeader() xdr.LedgerHeaderHistoryEntry
 	// Read should return the next transaction. If there are no more
 	// transactions it should return `io.EOF` error.
 	Read() (LedgerTransaction, error)
 	// Close should be called when reading is finished. This is especially
-	// helpful when there are still some entries available so the reader can stop
+	// helpful when there are still some transactions available so reader can stop
 	// streaming them.
 	Close() error
 }
 
 // LedgerWriteCloser provides convenient, streaming access to the transactions within a ledger.
 type LedgerWriteCloser interface {
+	// Write is used to pass a transaction to the next processor. It can return
+	// `io.ErrClosedPipe` when the pipe between processors has been closed meaning
+	// that next processor does not need more data. In such situation the current
+	// processor can terminate as sending more transactions to a `LedgerWriteCloser`
+	// does not make sense (will not be read).
 	Write(LedgerTransaction) error
 	// Close should be called when reading is finished. This is especially
 	// helpful when there are still some transactions available so the reader can stop
 	// streaming them.
 	Close() error
 }
+
+// LedgerTransaction represents the data for a single transaction within a ledger.
+type LedgerTransaction struct {
+	Index      uint32
+	Envelope   xdr.TransactionEnvelope
+	Result     xdr.TransactionResultPair
+	Meta       xdr.TransactionMeta
+	FeeChanges xdr.LedgerEntryChanges
+}

exp/ingest/io/mock_ledger_read_closer.go

@@ -16,9 +16,9 @@ func (m *MockLedgerReadCloser) GetSequence() uint32 {
 	return args.Get(0).(uint32)
 }
 
-func (m *MockLedgerReadCloser) GetHeader() (xdr.LedgerHeaderHistoryEntry, error) {
+func (m *MockLedgerReadCloser) GetHeader() xdr.LedgerHeaderHistoryEntry {
 	args := m.Called()
-	return args.Get(0).(xdr.LedgerHeaderHistoryEntry), nil
+	return args.Get(0).(xdr.LedgerHeaderHistoryEntry)
 }
 
 func (m *MockLedgerReadCloser) Read() (LedgerTransaction, error) {