loopdb+sweepbatcher: add GetParentBatch and TotalSwept calls

loopdb/sqlc/batch.sql.go

@@ -144,6 +144,54 @@ func (q *Queries) GetBatchSweeps(ctx context.Context, batchID int32) ([]GetBatch
 	return items, nil
 }
 
+const getBatchSweptAmount = `-- name: GetBatchSweptAmount :one
+SELECT
+        SUM(amt) AS total
+FROM
+        sweeps
+WHERE
+        batch_id = $1
+AND
+        completed = TRUE
+`
+
+func (q *Queries) GetBatchSweptAmount(ctx context.Context, batchID int32) (int64, error) {
+	row := q.db.QueryRowContext(ctx, getBatchSweptAmount, batchID)
+	var total int64
+	err := row.Scan(&total)
+	return total, err
+}
+
+const getParentBatch = `-- name: GetParentBatch :one
+SELECT
+        sweep_batches.id, sweep_batches.confirmed, sweep_batches.batch_tx_id, sweep_batches.batch_pk_script, sweep_batches.last_rbf_height, sweep_batches.last_rbf_sat_per_kw, sweep_batches.max_timeout_distance
+FROM
+        sweep_batches
+JOIN
+        sweeps ON sweep_batches.id = sweeps.batch_id
+WHERE
+        sweeps.swap_hash = $1
+AND
+        sweeps.completed = TRUE
+AND   
+        sweep_batches.confirmed = TRUE
+`
+
+func (q *Queries) GetParentBatch(ctx context.Context, swapHash []byte) (SweepBatch, error) {
+	row := q.db.QueryRowContext(ctx, getParentBatch, swapHash)
+	var i SweepBatch
+	err := row.Scan(
+		&i.ID,
+		&i.Confirmed,
+		&i.BatchTxID,
+		&i.BatchPkScript,
+		&i.LastRbfHeight,
+		&i.LastRbfSatPerKw,
+		&i.MaxTimeoutDistance,
+	)
+	return i, err
+}
+
 const getSweepStatus = `-- name: GetSweepStatus :one
 SELECT
     COALESCE(s.completed, f.false_value) AS completed

loopdb/sqlc/querier.go

@@ -13,6 +13,7 @@ type Querier interface {
 	CreateReservation(ctx context.Context, arg CreateReservationParams) error
 	FetchLiquidityParams(ctx context.Context) ([]byte, error)
 	GetBatchSweeps(ctx context.Context, batchID int32) ([]GetBatchSweepsRow, error)
+	GetBatchSweptAmount(ctx context.Context, batchID int32) (int64, error)
 	GetInstantOutSwap(ctx context.Context, swapHash []byte) (GetInstantOutSwapRow, error)
 	GetInstantOutSwapUpdates(ctx context.Context, swapHash []byte) ([]InstantoutUpdate, error)
 	GetInstantOutSwaps(ctx context.Context) ([]GetInstantOutSwapsRow, error)
@@ -20,6 +21,7 @@ type Querier interface {
 	GetLoopInSwaps(ctx context.Context) ([]GetLoopInSwapsRow, error)
 	GetLoopOutSwap(ctx context.Context, swapHash []byte) (GetLoopOutSwapRow, error)
 	GetLoopOutSwaps(ctx context.Context) ([]GetLoopOutSwapsRow, error)
+	GetParentBatch(ctx context.Context, swapHash []byte) (SweepBatch, error)
 	GetReservation(ctx context.Context, reservationID []byte) (Reservation, error)
 	GetReservationUpdates(ctx context.Context, reservationID []byte) ([]ReservationUpdate, error)
 	GetReservations(ctx context.Context) ([]Reservation, error)

loopdb/sqlc/queries/batch.sql

@@ -62,6 +62,29 @@ INSERT INTO sweeps (
         amt = $5,
         completed = $6;
 
+-- name: GetParentBatch :one
+SELECT
+        sweep_batches.*
+FROM
+        sweep_batches
+JOIN
+        sweeps ON sweep_batches.id = sweeps.batch_id
+WHERE
+        sweeps.swap_hash = $1
+AND
+        sweeps.completed = TRUE
+AND   
+        sweep_batches.confirmed = TRUE;
+
+-- name: GetBatchSweptAmount :one
+SELECT
+        SUM(amt) AS total
+FROM
+        sweeps
+WHERE
+        batch_id = $1
+AND
+        completed = TRUE;
 
 -- name: GetBatchSweeps :many
 SELECT

sweepbatcher/store.go

@@ -22,9 +22,17 @@ type BaseDB interface {
 	GetBatchSweeps(ctx context.Context, batchID int32) (
 		[]sqlc.GetBatchSweepsRow, error)
 
+	// GetBatchSweptAmount returns the total amount of sats swept by a
+	// (confirmed) batch.
+	GetBatchSweptAmount(ctx context.Context, batchID int32) (int64, error)
+
 	// GetSweepStatus returns true if the sweep has been completed.
 	GetSweepStatus(ctx context.Context, swapHash []byte) (bool, error)
 
+	// GetParentBatch fetches the parent batch of a completed sweep.
+	GetParentBatch(ctx context.Context, swapHash []byte) (sqlc.SweepBatch,
+		error)
+
 	// GetSwapUpdates fetches all the updates for a swap.
 	GetSwapUpdates(ctx context.Context, swapHash []byte) (
 		[]sqlc.SwapUpdate, error)
@@ -148,6 +156,34 @@ func (s *SQLStore) FetchBatchSweeps(ctx context.Context, id int32) (
 	return sweeps, nil
 }
 
+// TotalSweptAmount returns the total amount swept by a (confirmed) batch.
+func (s *SQLStore) TotalSweptAmount(ctx context.Context, id int32) (
+	btcutil.Amount, error) {
+
+	amt, err := s.baseDb.GetBatchSweptAmount(ctx, id)
+	if err != nil {
+		return 0, err
+	}
+
+	return btcutil.Amount(amt), nil
+}
+
+// GetParentBatch fetches the parent batch of a completed sweep.
+func (s *SQLStore) GetParentBatch(ctx context.Context, swapHash lntypes.Hash) (
+	*dbBatch, error) {
+
+	batch, err := s.baseDb.GetParentBatch(ctx, swapHash[:])
+	if err != nil {
+		return nil, err
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return convertBatchRow(batch), nil
+}
+
 // UpsertSweep inserts a sweep into the database, or updates an existing sweep
 // if it already exists.
 func (s *SQLStore) UpsertSweep(ctx context.Context, sweep *dbSweep) error {

sweepbatcher/store_mock.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"sort"
 
+	"github.com/btcsuite/btcd/btcutil"
 	"github.com/lightningnetwork/lnd/lntypes"
 )
 
@@ -123,3 +124,44 @@ func (s *StoreMock) AssertSweepStored(id lntypes.Hash) bool {
 	_, ok := s.sweeps[id]
 	return ok
 }
+
+// GetParentBatch returns the parent batch of a swap.
+func (s *StoreMock) GetParentBatch(ctx context.Context, swapHash lntypes.Hash) (
+	*dbBatch, error) {
+
+	for _, sweep := range s.sweeps {
+		if sweep.SwapHash == swapHash {
+			batch, ok := s.batches[sweep.BatchID]
+			if !ok {
+				return nil, errors.New("batch not found")
+			}
+			return &batch, nil
+		}
+	}
+
+	return nil, errors.New("batch not found")
+}
+
+// TotalSweptAmount returns the total amount of BTC that has been swept from a
+// batch.
+func (s *StoreMock) TotalSweptAmount(ctx context.Context, batchID int32) (
+	btcutil.Amount, error) {
+
+	batch, ok := s.batches[batchID]
+	if !ok {
+		return 0, errors.New("batch not found")
+	}
+
+	if batch.State != batchConfirmed && batch.State != batchClosed {
+		return 0, nil
+	}
+
+	var total btcutil.Amount
+	for _, sweep := range s.sweeps {
+		if sweep.BatchID == batchID {
+			total += sweep.Amount
+		}
+	}
+
+	return 0, nil
+}

sweepbatcher/sweep_batcher.go

@@ -72,6 +72,14 @@ type BatcherStore interface {
 	// GetSweepStatus returns the completed status of the sweep.
 	GetSweepStatus(ctx context.Context, swapHash lntypes.Hash) (
 		bool, error)
+
+	// GetParentBatch returns the parent batch of a (completed) sweep.
+	GetParentBatch(ctx context.Context, swapHash lntypes.Hash) (
+		*dbBatch, error)
+
+	// TotalSweptAmount returns the total amount swept by a (confirmed)
+	// batch.
+	TotalSweptAmount(ctx context.Context, id int32) (btcutil.Amount, error)
 }
 
 // MuSig2SignSweep is a function that can be used to sign a sweep transaction