fix(ledger): pit usage

components/ledger/internal/storage/ledgerstore/bucket.go

@@ -20,6 +20,9 @@ var initSchema string
 //go:embed migrations/1-fix-trigger.sql
 var fixTrigger string
 
+//go:embed migrations/2-fix-volumes-aggregation.sql
+var fixVolumesAggregation string
+
 type Bucket struct {
 	name string
 	db   *bun.DB
@@ -137,6 +140,13 @@ func registerMigrations(migrator *migrations.Migrator, name string) {
 				return err
 			},
 		},
+		migrations.Migration{
+			Name: "Fix volumes aggregation",
+			UpWithContext: func(ctx context.Context, tx bun.Tx) error {
+				_, err := tx.ExecContext(ctx, fixVolumesAggregation)
+				return err
+			},
+		},
 	)
 }
 

components/ledger/internal/storage/ledgerstore/migrations/2-fix-volumes-aggregation.sql

@@ -0,0 +1,23 @@
+create or replace function get_all_account_volumes(_ledger varchar, _account varchar, _before timestamp default null)
+    returns setof volumes_with_asset
+    language sql
+    stable
+as
+$$
+with all_assets as (select v.v as asset
+                    from get_all_assets(_ledger) v),
+     moves as (select m.*
+               from all_assets assets
+                        join lateral (
+                   select *
+                   from moves s
+                   where (_before is null or s.effective_date <= _before)
+                     and s.account_address = _account
+                     and s.asset = assets.asset
+                     and s.ledger = _ledger
+                   order by seq desc
+                   limit 1
+                   ) m on true)
+select moves.asset, moves.post_commit_volumes
+from moves
+$$;

releases/sdks/go/.speakeasy/gen.lock

@@ -1,5 +1,5 @@
 lockVersion: 2.0.0
-id: 9017c559-33fa-4ae1-be17-4f23d489a1ad
+id: 0fdfd940-dc71-4d72-b356-dc122a42aa76
 management:
   docChecksum: 5ecb893e1ddb7b683bdb143d145d47a0
   docVersion: INTERNAL

tests/integration/internal/test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/xo/dburl"
 	"net/http"
 	"net/http/httptest"
+	"os"
 )
 
 type routing struct {
@@ -83,8 +84,11 @@ func (test *Test) createDatabase(ctx context.Context, name string) error {
 }
 
 func (test *Test) dropDatabase(ctx context.Context, name string) error {
-	_, err := test.env.sqlConn.Exec(ctx, fmt.Sprintf(`DROP DATABASE "%s-%s";`, test.id, name))
-	return err
+	if os.Getenv("NO_CLEANUP") != "true" {
+		_, err := test.env.sqlConn.Exec(ctx, fmt.Sprintf(`DROP DATABASE "%s-%s";`, test.id, name))
+		return err
+	}
+	return nil
 }
 
 func (test *Test) registerServiceToRoute(name string, routing routing) {

tests/integration/suite/ledger-list-count-accounts.go

@@ -355,4 +355,51 @@ var _ = WithModules([]*Module{modules.Ledger}, func() {
 			})
 		})
 	})
+
+	When("Inserting one transaction in past and one in the future", func() {
+		now := time.Now()
+		BeforeEach(func() {
+			_, err := Client().Ledger.V2CreateTransaction(TestContext(), operations.V2CreateTransactionRequest{
+				V2PostTransaction: shared.V2PostTransaction{
+					Postings: []shared.V2Posting{{
+						Amount:      big.NewInt(100),
+						Asset:       "USD",
+						Destination: "foo",
+						Source:      "world",
+					}},
+					Timestamp: pointer.For(now.Add(-12 * time.Hour)),
+					Metadata:  map[string]string{},
+				},
+				Ledger: "default",
+			})
+			Expect(err).To(Succeed())
+
+			_, err = Client().Ledger.V2CreateTransaction(TestContext(), operations.V2CreateTransactionRequest{
+				V2PostTransaction: shared.V2PostTransaction{
+					Postings: []shared.V2Posting{{
+						Amount:      big.NewInt(100),
+						Asset:       "USD",
+						Destination: "foo",
+						Source:      "world",
+					}},
+					Timestamp: pointer.For(now.Add(12 * time.Hour)),
+					Metadata:  map[string]string{},
+				},
+				Ledger: "default",
+			})
+			Expect(err).To(Succeed())
+		})
+		When("getting account in the present", func() {
+			It("should ignore future transaction", func() {
+				accountResponse, err := Client().Ledger.V2GetAccount(TestContext(), operations.V2GetAccountRequest{
+					Address: "foo",
+					Expand:  pointer.For("volumes"),
+					Ledger:  "default",
+					Pit:     pointer.For(time.Now()),
+				})
+				Expect(err).To(Succeed())
+				Expect(accountResponse.V2AccountResponse.Data.Volumes["USD"].Balance).To(Equal(big.NewInt(100)))
+			})
+		})
+	})
 })