chore: add stream info on attestation table

internal/migrations/024-attestation-actions.sql

@@ -148,11 +148,11 @@ CREATE OR REPLACE ACTION request_attestation(
 
     -- Store unsigned attestation
     INSERT INTO attestations (
-        request_tx_id, attestation_hash, requester, result_canonical, encrypt_sig, 
-        created_height, signature, validator_pubkey, signed_height
+        request_tx_id, attestation_hash, requester, data_provider, stream_id,
+        result_canonical, encrypt_sig, created_height, signature, validator_pubkey, signed_height
     ) VALUES (
-        $request_tx_id, $attestation_hash, $caller_bytes, $result_canonical, $encrypt_sig, 
-        $created_height, NULL, NULL, NULL
+        $request_tx_id, $attestation_hash, $caller_bytes, $data_provider, $stream_id,
+        $result_canonical, $encrypt_sig, $created_height, NULL, NULL, NULL
     );
 
     -- Queue for signing (no-op on non-leader validators; handled by precompile)
@@ -284,6 +284,8 @@ CREATE OR REPLACE ACTION list_attestations(
     request_tx_id TEXT,
     attestation_hash BYTEA,
     requester BYTEA,
+    data_provider TEXT,
+    stream_id TEXT,
     created_height INT8,
     signed_height INT8,
     encrypt_sig BOOLEAN
@@ -310,21 +312,21 @@ CREATE OR REPLACE ACTION list_attestations(
     -- Build query with optional requester and request_tx_id filters
     IF $request_tx_id IS NOT NULL {
         -- Filter by specific request_tx_id (ignores other filters for exact match)
-        RETURN SELECT request_tx_id, attestation_hash, requester,
+        RETURN SELECT request_tx_id, attestation_hash, requester, data_provider, stream_id,
                       created_height, signed_height, encrypt_sig
                FROM attestations
                WHERE request_tx_id = $request_tx_id
                LIMIT 1;
     } ELSEIF $requester IS NULL {
         -- Show all attestations (analytics/auditing)
         IF $order_desc {
-            RETURN SELECT request_tx_id, attestation_hash, requester,
+            RETURN SELECT request_tx_id, attestation_hash, requester, data_provider, stream_id,
                           created_height, signed_height, encrypt_sig
                    FROM attestations
                    ORDER BY created_height DESC, request_tx_id ASC
                    LIMIT $limit OFFSET $offset;
         } ELSE {
-            RETURN SELECT request_tx_id, attestation_hash, requester,
+            RETURN SELECT request_tx_id, attestation_hash, requester, data_provider, stream_id,
                           created_height, signed_height, encrypt_sig
                    FROM attestations
                    ORDER BY created_height ASC, request_tx_id ASC
@@ -333,14 +335,14 @@ CREATE OR REPLACE ACTION list_attestations(
     } ELSE {
         -- Filter by requester
         IF $order_desc {
-            RETURN SELECT request_tx_id, attestation_hash, requester,
+            RETURN SELECT request_tx_id, attestation_hash, requester, data_provider, stream_id,
                           created_height, signed_height, encrypt_sig
                    FROM attestations
                    WHERE requester = $requester
                    ORDER BY created_height DESC, request_tx_id ASC
                    LIMIT $limit OFFSET $offset;
         } ELSE {
-            RETURN SELECT request_tx_id, attestation_hash, requester,
+            RETURN SELECT request_tx_id, attestation_hash, requester, data_provider, stream_id,
                           created_height, signed_height, encrypt_sig
                    FROM attestations
                    WHERE requester = $requester

internal/migrations/028-attestation-add-provider-stream.sql

@@ -0,0 +1,19 @@
+-- Add data_provider column to attestations table
+ALTER TABLE attestations
+ADD COLUMN IF NOT EXISTS data_provider TEXT;
+
+-- Add stream_id column to attestations table
+ALTER TABLE attestations
+ADD COLUMN IF NOT EXISTS stream_id TEXT;
+
+-- Create index on data_provider for efficient querying
+CREATE INDEX IF NOT EXISTS ix_att_data_provider
+    ON attestations(data_provider);
+
+-- Create index on stream_id for efficient querying
+CREATE INDEX IF NOT EXISTS ix_att_stream_id
+    ON attestations(stream_id);
+
+-- Create composite index for data_provider + stream_id queries
+CREATE INDEX IF NOT EXISTS ix_att_provider_stream
+    ON attestations(data_provider, stream_id);

tests/streams/attestation/attestation_request_test.go

@@ -85,6 +85,8 @@ func runAttestationHappyPath(helper *AttestationTestHelper, actionName string, a
 
 	stored := fetchAttestationRow(helper, attestationHash)
 	require.Equal(helper.t, requestTxID, stored.requestTxID, "request_tx_id should be captured and stored")
+	require.Equal(helper.t, dataProviderHex, stored.dataProvider, "data_provider should be stored")
+	require.Equal(helper.t, streamID, stored.streamID, "stream_id should be stored")
 
 	// Rebuild expected canonical payload
 	valueDecimal := types.MustParseDecimal(fmt.Sprintf("%d.%018d", attestedValue, 0))
@@ -124,6 +126,8 @@ func runAttestationHappyPath(helper *AttestationTestHelper, actionName string, a
 type attestationRow struct {
 	requestTxID     string
 	requester       []byte
+	dataProvider    string
+	streamID        string
 	attestationHash []byte
 	resultCanonical []byte
 	encryptSig      bool
@@ -178,26 +182,28 @@ func fetchAttestationRow(helper *AttestationTestHelper, hash []byte) attestation
 
 	var rowData attestationRow
 	err := helper.platform.Engine.Execute(engineCtx, helper.platform.DB, `
-SELECT request_tx_id, requester, attestation_hash, result_canonical, encrypt_sig, signature, validator_pubkey, signed_height, created_height
+SELECT request_tx_id, requester, data_provider, stream_id, attestation_hash, result_canonical, encrypt_sig, signature, validator_pubkey, signed_height, created_height
 FROM attestations
 WHERE attestation_hash = $hash;
 `, map[string]any{"hash": hash}, func(row *common.Row) error {
 		rowData.requestTxID = row.Values[0].(string)
 		rowData.requester = append([]byte(nil), row.Values[1].([]byte)...)
-		rowData.attestationHash = append([]byte(nil), row.Values[2].([]byte)...)
-		rowData.resultCanonical = append([]byte(nil), row.Values[3].([]byte)...)
-		rowData.encryptSig = row.Values[4].(bool)
-		if row.Values[5] != nil {
-			rowData.signature = append([]byte(nil), row.Values[5].([]byte)...)
+		rowData.dataProvider = row.Values[2].(string)
+		rowData.streamID = row.Values[3].(string)
+		rowData.attestationHash = append([]byte(nil), row.Values[4].([]byte)...)
+		rowData.resultCanonical = append([]byte(nil), row.Values[5].([]byte)...)
+		rowData.encryptSig = row.Values[6].(bool)
+		if row.Values[7] != nil {
+			rowData.signature = append([]byte(nil), row.Values[7].([]byte)...)
 		}
-		if row.Values[6] != nil {
-			rowData.validatorPubKey = append([]byte(nil), row.Values[6].([]byte)...)
+		if row.Values[8] != nil {
+			rowData.validatorPubKey = append([]byte(nil), row.Values[8].([]byte)...)
 		}
-		if row.Values[7] != nil {
-			height := row.Values[7].(int64)
+		if row.Values[9] != nil {
+			height := row.Values[9].(int64)
 			rowData.signedHeight = &height
 		}
-		rowData.createdHeight = row.Values[8].(int64)
+		rowData.createdHeight = row.Values[10].(int64)
 		return nil
 	})
 	require.NoError(helper.t, err)

tests/streams/attestation/attestation_retrieval_test.go

@@ -160,7 +160,15 @@ func testListNoFilter(h *AttestationTestHelper, actionName string, addrs *TestAd
 	count := 0
 	h.CallAction("list_attestations", []any{nil, nil, 100, 0, nil}, func(row *common.Row) error {
 		count++
-		require.Len(h.t, row.Values, 6, "should return 6 columns")
+		require.Len(h.t, row.Values, 8, "should return 8 columns: request_tx_id, attestation_hash, requester, data_provider, stream_id, created_height, signed_height, encrypt_sig")
+
+		// Verify data_provider and stream_id are present
+		dataProvider := row.Values[3].(string)
+		streamID := row.Values[4].(string)
+		require.NotEmpty(h.t, dataProvider, "data_provider should not be empty")
+		require.NotEmpty(h.t, streamID, "stream_id should not be empty")
+		require.Equal(h.t, TestDataProviderHex, dataProvider, "data_provider should match test value")
+		require.Equal(h.t, TestStreamID, streamID, "stream_id should match test value")
 		return nil
 	})
 	require.Equal(h.t, 3, count, "should return all 3 attestations")