Generate payer reports (#646)

## tl;dr

- Defines structs for the off chain representation of Payer Reports
- Actually generates Payer Reports
- Alters the `unsettled_usage` table to keep track of the last known sequence ID for each originator/payer combo in a given minute. This makes it cheap to look up when generating a Payer Report

## Notes

### Minimum Report Size

- To generate a Payer Report you need a minimum of two messages, originated in two distinct minutes. This resolves #645

## Tickets

Implements part of #514

<!-- This is an auto-generated comment: release notes by coderabbit.ai -->
## Summary by CodeRabbit

- **New Features**
  - Introduced new queries to aggregate spending data and retrieve envelope details, enhancing data reporting.
  - Added a comprehensive mechanism for generating and verifying payer reports, improving overall reporting accuracy.
  - Improved envelope creation by integrating timestamp support for more precise data handling.
  - Added new queries for enhanced data retrieval capabilities, including payer reports and gateway envelope details.
  - Introduced a new field for tracking last sequence IDs in unsettled usage records, enhancing data integrity.

- **Bug Fixes**
  - Adjusted existing queries to return additional data fields, ensuring accurate reporting.

- **Tests**
  - Expanded test coverage to validate various report generation scenarios, ensuring reliable performance.
  - Added tests for handling out-of-order sequence IDs during database operations, improving robustness.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: neekolas

pkg/db/gatewayEnvelope.go

@@ -33,6 +33,10 @@ func InsertGatewayEnvelopeAndIncrementUnsettledUsage(
 
 			var wg sync.WaitGroup
 			var incrementErr, congestionErr error
+			// Use the sequence ID from the envelope to set the last sequence ID value
+			if incrementParams.SequenceID == 0 {
+				incrementParams.SequenceID = insertParams.OriginatorSequenceID
+			}
 
 			wg.Add(2)
 

pkg/db/gatewayEnvelope_test.go

@@ -64,7 +64,8 @@ func TestInsertAndIncrement(t *testing.T) {
 		queries.GetPayerUnsettledUsageParams{PayerID: payerID},
 	)
 	require.NoError(t, err)
-	require.Equal(t, payerSpend, int64(100))
+	require.Equal(t, payerSpend.TotalSpendPicodollars, int64(100))
+	require.Equal(t, payerSpend.LastSequenceID, sequenceID)
 
 	originatorCongestion, err := querier.SumOriginatorCongestion(
 		ctx,
@@ -138,7 +139,8 @@ func TestInsertAndIncrementParallel(t *testing.T) {
 		queries.GetPayerUnsettledUsageParams{PayerID: payerID},
 	)
 	require.NoError(t, err)
-	require.Equal(t, payerSpend, int64(100))
+	require.Equal(t, payerSpend.TotalSpendPicodollars, int64(100))
+	require.Equal(t, payerSpend.LastSequenceID, sequenceID)
 
 	originatorCongestion, err := querier.SumOriginatorCongestion(
 		ctx,
@@ -147,3 +149,44 @@ func TestInsertAndIncrementParallel(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, originatorCongestion, int64(1))
 }
+
+func TestInsertAndIncrementWithOutOfOrderSequenceID(t *testing.T) {
+	ctx := context.Background()
+	db, _, cleanup := testutils.NewDB(t, ctx)
+	defer cleanup()
+
+	querier := queries.New(db)
+
+	payerID := testutils.CreatePayer(t, db, testutils.RandomAddress().Hex())
+	originatorID := testutils.RandomInt32()
+	sequenceID := int64(10)
+
+	insertParams, incrementParams := buildParams(payerID, originatorID, sequenceID, 100)
+
+	_, err := xmtpd_db.InsertGatewayEnvelopeAndIncrementUnsettledUsage(
+		ctx,
+		db,
+		insertParams,
+		incrementParams,
+	)
+	require.NoError(t, err)
+
+	lowerSequenceID := int64(5)
+
+	insertParams, incrementParams = buildParams(payerID, originatorID, lowerSequenceID, 100)
+
+	_, err = xmtpd_db.InsertGatewayEnvelopeAndIncrementUnsettledUsage(
+		ctx,
+		db,
+		insertParams,
+		incrementParams,
+	)
+	require.NoError(t, err)
+
+	payerSpend, err := querier.GetPayerUnsettledUsage(
+		ctx,
+		queries.GetPayerUnsettledUsageParams{PayerID: payerID},
+	)
+	require.NoError(t, err)
+	require.Equal(t, payerSpend.LastSequenceID, sequenceID)
+}

pkg/db/queries.sql

@@ -181,15 +181,17 @@ ON CONFLICT (address)
 		id;
 
 -- name: IncrementUnsettledUsage :exec
-INSERT INTO unsettled_usage(payer_id, originator_id, minutes_since_epoch, spend_picodollars)
-	VALUES (@payer_id, @originator_id, @minutes_since_epoch, @spend_picodollars)
+INSERT INTO unsettled_usage(payer_id, originator_id, minutes_since_epoch, spend_picodollars, last_sequence_id)
+	VALUES (@payer_id, @originator_id, @minutes_since_epoch, @spend_picodollars, @sequence_id)
 ON CONFLICT (payer_id, originator_id, minutes_since_epoch)
 	DO UPDATE SET
-		spend_picodollars = unsettled_usage.spend_picodollars + @spend_picodollars;
+		spend_picodollars = unsettled_usage.spend_picodollars + @spend_picodollars,
+		last_sequence_id = GREATEST(unsettled_usage.last_sequence_id, @sequence_id);
 
 -- name: GetPayerUnsettledUsage :one
 SELECT
-	COALESCE(SUM(spend_picodollars), 0)::BIGINT AS total_spend_picodollars
+	COALESCE(SUM(spend_picodollars), 0)::BIGINT AS total_spend_picodollars,
+	COALESCE(MAX(last_sequence_id), 0)::BIGINT AS last_sequence_id
 FROM
 	unsettled_usage
 WHERE
@@ -261,3 +263,42 @@ WHERE
 		OR minutes_since_epoch > @minutes_since_epoch_gt::BIGINT)
 	AND (@minutes_since_epoch_lt::BIGINT = 0
 		OR minutes_since_epoch < @minutes_since_epoch_lt::BIGINT);
+
+-- name: BuildPayerReport :many
+SELECT
+	payers.address as payer_address,
+	SUM(spend_picodollars)::BIGINT AS total_spend_picodollars
+FROM
+	unsettled_usage
+JOIN payers on payers.id = unsettled_usage.payer_id
+WHERE
+	originator_id = @originator_id
+	AND minutes_since_epoch > @start_minutes_since_epoch
+	AND minutes_since_epoch <= @end_minutes_since_epoch
+GROUP BY
+	payers.address;
+
+-- name: GetGatewayEnvelopeByID :one
+SELECT * FROM gateway_envelopes
+WHERE originator_sequence_id = @originator_sequence_id
+-- Include the node ID to take advantage of the primary key index
+AND originator_node_id = @originator_node_id;
+
+-- name: GetSecondNewestMinute :one
+WITH second_newest_minute
+AS
+  (
+           SELECT minutes_since_epoch
+           FROM     unsettled_usage
+           WHERE    originator_id = @originator_id
+           AND      unsettled_usage.minutes_since_epoch > @minimum_minutes_since_epoch
+           GROUP BY unsettled_usage.minutes_since_epoch
+           ORDER BY unsettled_usage.minutes_since_epoch DESC
+           LIMIT    1
+           OFFSET   1)
+  SELECT coalesce(max(last_sequence_id), 0)::BIGINT as max_sequence_id,
+         coalesce(max(unsettled_usage.minutes_since_epoch), 0)::INT as minutes_since_epoch
+  FROM   unsettled_usage
+  JOIN   second_newest_minute
+  ON     second_newest_minute.minutes_since_epoch = unsettled_usage.minutes_since_epoch
+  WHERE  unsettled_usage.originator_id = @originator_id;

pkg/db/queries/models.go

@@ -34,7 +34,7 @@ func TestIncrementUnsettledUsage(t *testing.T) {
 		},
 	)
 	require.NoError(t, err)
-	require.Equal(t, unsettledUsage, int64(100))
+	require.Equal(t, unsettledUsage.TotalSpendPicodollars, int64(100))
 
 	require.NoError(t, querier.IncrementUnsettledUsage(ctx, queries.IncrementUnsettledUsageParams{
 		PayerID:           payerId,
@@ -50,7 +50,7 @@ func TestIncrementUnsettledUsage(t *testing.T) {
 		},
 	)
 	require.NoError(t, err)
-	require.Equal(t, unsettledUsage, int64(200))
+	require.Equal(t, unsettledUsage.TotalSpendPicodollars, int64(200))
 }
 
 func TestGetUnsettledUsage(t *testing.T) {
@@ -86,7 +86,7 @@ func TestGetUnsettledUsage(t *testing.T) {
 		},
 	)
 	require.NoError(t, err)
-	require.Equal(t, unsettledUsage, int64(300))
+	require.Equal(t, unsettledUsage.TotalSpendPicodollars, int64(300))
 
 	unsettledUsage, err = querier.GetPayerUnsettledUsage(
 		ctx,
@@ -96,5 +96,5 @@ func TestGetUnsettledUsage(t *testing.T) {
 		},
 	)
 	require.NoError(t, err)
-	require.Equal(t, unsettledUsage, int64(500))
+	require.Equal(t, unsettledUsage.TotalSpendPicodollars, int64(500))
 }

pkg/db/queries/queries.sql.go

@@ -0,0 +1 @@
+ALTER TABLE unsettled_usage DROP COLUMN last_sequence_id;

pkg/db/unsettledUsage_test.go

@@ -0,0 +1,14 @@
+-- Dropping the table to add a new column with a not null constraint
+DROP TABLE unsettled_usage;
+
+CREATE TABLE unsettled_usage(
+	payer_id INTEGER NOT NULL,
+	originator_id INTEGER NOT NULL,
+	minutes_since_epoch INTEGER NOT NULL,
+	spend_picodollars BIGINT NOT NULL,
+	last_sequence_id BIGINT NOT NULL,
+	PRIMARY KEY (payer_id, originator_id, minutes_since_epoch)
+);
+
+CREATE INDEX idx_unsettled_usage_originator_id_minutes_since_epoch
+    ON unsettled_usage(originator_id, minutes_since_epoch DESC);