feat: adapt for v2.2 merge

formancehq · Nov 22, 2024 · ac50890 · ac50890
1 parent 1351a12
commit ac50890
Show file tree

Hide file tree

Showing 10 changed files with 95 additions and 31 deletions.
diff --git a/internal/storage/bucket/bucket.go b/internal/storage/bucket/bucket.go
@@ -13,6 +13,7 @@ type Bucket interface {
 	Migrate(ctx context.Context, minimalVersionReached chan struct{}, opts ...migrations.Option) error
 	AddLedger(ctx context.Context, ledger ledger.Ledger) error
 	HasMinimalVersion(ctx context.Context) (bool, error)
+	IsUpToDate(ctx context.Context) (bool, error)
 	GetMigrationsInfo(ctx context.Context) ([]migrations.Info, error)
 }
 

diff --git a/internal/storage/bucket/default_bucket.go b/internal/storage/bucket/default_bucket.go
@@ -22,20 +22,27 @@ type DefaultBucket struct {
 	tracer trace.Tracer
 }
 
+func (b *DefaultBucket) IsUpToDate(ctx context.Context) (bool, error) {
+	return GetMigrator(b.db, b.name).IsUpToDate(ctx)
+}
+
 func (b *DefaultBucket) Migrate(ctx context.Context, minimalVersionReached chan struct{}, options ...migrations.Option) error {
 	return migrate(ctx, b.tracer, b.db, b.name, minimalVersionReached, options...)
 }
 
 func (b *DefaultBucket) HasMinimalVersion(ctx context.Context) (bool, error) {
-	migrator := GetMigrator(b.db, b.name)
-	lastVersion, err := migrator.GetLastVersion(ctx)
+	lastVersion, err := b.GetLastVersion(ctx)
 	if err != nil {
 		return false, err
 	}
 
 	return lastVersion >= MinimalSchemaVersion, nil
 }
 
+func (b *DefaultBucket) GetLastVersion(ctx context.Context) (int, error) {
+	return GetMigrator(b.db, b.name).GetLastVersion(ctx)
+}
+
 func (b *DefaultBucket) GetMigrationsInfo(ctx context.Context) ([]migrations.Info, error) {
 	return GetMigrator(b.db, b.name).GetMigrations(ctx)
 }

diff --git a/internal/storage/bucket/migrations.go b/internal/storage/bucket/migrations.go
@@ -15,11 +15,11 @@ var MigrationsFS embed.FS
 func GetMigrator(db *bun.DB, name string, options ...migrations.Option) *migrations.Migrator {
 	options = append(options, migrations.WithSchema(name))
 	migrator := migrations.NewMigrator(db, options...)
-	_migrations, err := migrations.CollectMigrations(MigrationsFS, name)
+	migrations, err := migrations.CollectMigrations(MigrationsFS, name)
 	if err != nil {
 		panic(err)
 	}
-	migrator.RegisterMigrations(_migrations...)
+	migrator.RegisterMigrations(migrations...)
 
 	return migrator
 }
@@ -59,4 +59,4 @@ func migrate(ctx context.Context, tracer trace.Tracer, db *bun.DB, name string,
 			}
 		}
 	}
-}
+}
diff --git a/internal/storage/driver/adapters.go b/internal/storage/driver/adapters.go
@@ -2,6 +2,7 @@ package driver
 
 import (
 	"context"
+	"fmt"
 	ledgerstore "github.com/formancehq/ledger/internal/storage/ledger/legacy"
 
 	ledger "github.com/formancehq/ledger/internal"
@@ -19,7 +20,12 @@ func (d *DefaultStorageDriverAdapter) OpenLedger(ctx context.Context, name strin
 		return nil, nil, err
 	}
 
-	return ledgerstore.NewDefaultStoreAdapter(store), l, nil
+	isUpToDate, err := store.GetBucket().IsUpToDate(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("checking if bucket is up to date: %w", err)
+	}
+
+	return ledgerstore.NewDefaultStoreAdapter(isUpToDate, store), l, nil
 }
 
 func (d *DefaultStorageDriverAdapter) CreateLedger(ctx context.Context, l *ledger.Ledger) error {
@@ -31,4 +37,4 @@ func NewControllerStorageDriverAdapter(d *Driver) *DefaultStorageDriverAdapter {
 	return &DefaultStorageDriverAdapter{Driver: d}
 }
 
-var _ systemcontroller.Store = (*DefaultStorageDriverAdapter)(nil)
+var _ systemcontroller.Store = (*DefaultStorageDriverAdapter)(nil)
diff --git a/internal/storage/driver/buckets_generated_test.go b/internal/storage/driver/buckets_generated_test.go
diff --git a/internal/storage/ledger/legacy/adapters.go b/internal/storage/ledger/legacy/adapters.go
@@ -14,8 +14,9 @@ import (
 )
 
 type DefaultStoreAdapter struct {
-	newStore    *ledgerstore.Store
-	legacyStore *Store
+	newStore       *ledgerstore.Store
+	legacyStore    *Store
+	isFullUpToDate bool
 }
 
 func (d *DefaultStoreAdapter) GetDB() bun.IDB {
@@ -63,43 +64,79 @@ func (d *DefaultStoreAdapter) LockLedger(ctx context.Context) error {
 }
 
 func (d *DefaultStoreAdapter) ListLogs(ctx context.Context, q ledgercontroller.GetLogsQuery) (*bunpaginate.Cursor[ledger.Log], error) {
-	return d.legacyStore.GetLogs(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.GetLogs(ctx, q)
+	}
+
+	return d.newStore.ListLogs(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) ReadLogWithIdempotencyKey(ctx context.Context, ik string) (*ledger.Log, error) {
 	return d.newStore.ReadLogWithIdempotencyKey(ctx, ik)
 }
 
 func (d *DefaultStoreAdapter) ListTransactions(ctx context.Context, q ledgercontroller.ListTransactionsQuery) (*bunpaginate.Cursor[ledger.Transaction], error) {
-	return d.legacyStore.GetTransactions(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.GetTransactions(ctx, q)
+	}
+
+	return d.newStore.ListTransactions(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) CountTransactions(ctx context.Context, q ledgercontroller.ListTransactionsQuery) (int, error) {
-	return d.legacyStore.CountTransactions(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.CountTransactions(ctx, q)
+	}
+
+	return d.newStore.CountTransactions(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) GetTransaction(ctx context.Context, query ledgercontroller.GetTransactionQuery) (*ledger.Transaction, error) {
-	return d.legacyStore.GetTransactionWithVolumes(ctx, query)
+	if !d.isFullUpToDate {
+		return d.legacyStore.GetTransactionWithVolumes(ctx, query)
+	}
+
+	return d.newStore.GetTransaction(ctx, query)
 }
 
 func (d *DefaultStoreAdapter) CountAccounts(ctx context.Context, q ledgercontroller.ListAccountsQuery) (int, error) {
-	return d.legacyStore.CountAccounts(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.CountAccounts(ctx, q)
+	}
+
+	return d.newStore.CountAccounts(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) ListAccounts(ctx context.Context, q ledgercontroller.ListAccountsQuery) (*bunpaginate.Cursor[ledger.Account], error) {
-	return d.legacyStore.GetAccountsWithVolumes(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.GetAccountsWithVolumes(ctx, q)
+	}
+
+	return d.newStore.ListAccounts(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) GetAccount(ctx context.Context, q ledgercontroller.GetAccountQuery) (*ledger.Account, error) {
-	return d.legacyStore.GetAccountWithVolumes(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.GetAccountWithVolumes(ctx, q)
+	}
+
+	return d.newStore.GetAccount(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) GetAggregatedBalances(ctx context.Context, q ledgercontroller.GetAggregatedBalanceQuery) (ledger.BalancesByAssets, error) {
-	return d.legacyStore.GetAggregatedBalances(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.GetAggregatedBalances(ctx, q)
+	}
+
+	return d.newStore.GetAggregatedBalances(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) GetVolumesWithBalances(ctx context.Context, q ledgercontroller.GetVolumesWithBalancesQuery) (*bunpaginate.Cursor[ledger.VolumesWithBalanceByAssetByAccount], error) {
-	return d.legacyStore.GetVolumesWithBalances(ctx, q)
+	if !d.isFullUpToDate {
+		return d.legacyStore.GetVolumesWithBalances(ctx, q)
+	}
+
+	return d.newStore.GetVolumesWithBalances(ctx, q)
 }
 
 func (d *DefaultStoreAdapter) IsUpToDate(ctx context.Context) (bool, error) {
@@ -132,10 +169,11 @@ func (d *DefaultStoreAdapter) Rollback() error {
 	return d.newStore.Rollback()
 }
 
-func NewDefaultStoreAdapter(store *ledgerstore.Store) *DefaultStoreAdapter {
+func NewDefaultStoreAdapter(isFullUpToDate bool, store *ledgerstore.Store) *DefaultStoreAdapter {
 	return &DefaultStoreAdapter{
-		newStore:    store,
-		legacyStore: New(store.GetDB(), store.GetLedger().Bucket, store.GetLedger().Name),
+		isFullUpToDate: isFullUpToDate,
+		newStore:       store,
+		legacyStore:    New(store.GetDB(), store.GetLedger().Bucket, store.GetLedger().Name),
 	}
 }
 

diff --git a/internal/storage/ledger/store.go b/internal/storage/ledger/store.go
@@ -86,6 +86,10 @@ func (s *Store) GetDB() bun.IDB {
 	return s.db
 }
 
+func (s *Store) GetBucket() bucket.Bucket {
+	return s.bucket
+}
+
 func (s *Store) GetPrefixedRelationName(v string) string {
 	return fmt.Sprintf(`"%s".%s`, s.ledger.Bucket, v)
 }

diff --git a/internal/storage/ledger/transactions_test.go b/internal/storage/ledger/transactions_test.go
@@ -7,13 +7,6 @@ import (
 	"database/sql"
 	"fmt"
 	"github.com/alitto/pond"
-	"github.com/formancehq/go-libs/v2/bun/bunconnect"
-	"github.com/formancehq/ledger/internal/storage/bucket"
-	driver "github.com/formancehq/ledger/internal/storage/driver"
-	ledgerstore "github.com/formancehq/ledger/internal/storage/ledger"
-	systemstore "github.com/formancehq/ledger/internal/storage/system"
-	"github.com/google/uuid"
-	"go.opentelemetry.io/otel/trace/noop"
 	"math/big"
 	"slices"
 	"testing"

diff --git a/test/e2e/api_accounts_list_test.go b/test/e2e/api_accounts_list_test.go
@@ -110,7 +110,7 @@ var _ = Context("Ledger accounts list API tests", func() {
 				},
 			)
 			Expect(err).To(HaveOccurred())
-			Expect(err).To(HaveErrorCode(string(components.V2ErrorsEnumInternal)))
+			Expect(err).To(HaveErrorCode(string(components.V2ErrorsEnumValidation)))
 		})
 		It("should be countable on api", func() {
 			response, err := CountAccounts(

diff --git a/test/e2e/api_transactions_list_test.go b/test/e2e/api_transactions_list_test.go
@@ -331,8 +331,8 @@ var _ = Context("Ledger transactions list API tests", func() {
 				)
 				Expect(err).To(HaveOccurred())
 			})
-			It("Should fail with "+string(components.V2ErrorsEnumInternal)+" error code", func() {
-				Expect(err).To(HaveErrorCode(string(components.V2ErrorsEnumInternal)))
+			It("Should fail with "+string(components.V2ErrorsEnumValidation)+" error code", func() {
+				Expect(err).To(HaveErrorCode(string(components.V2ErrorsEnumValidation)))
 			})
 		})
 	})