chore: add stream creation fees

extensions/tn_attestation/harness_integration_test.go

@@ -80,7 +80,7 @@ func TestSigningWorkflowWithHarness(t *testing.T) {
 
 	kwilTesting.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "tn_attestation_signing_harness",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		Owner:       ownerAddr.Address(),
 		FunctionTests: []kwilTesting.TestFunc{
 			func(ctx context.Context, platform *kwilTesting.Platform) error {

extensions/tn_cache/internal/engine_ops_integration_test.go

@@ -32,7 +32,7 @@ func generateTestStreamID(prefix string) util.StreamId {
 func TestEngineOperations_Integration(t *testing.T) {
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "engine_ops_integration_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testEngineOperationsIntegration(t),
 		},
@@ -331,7 +331,7 @@ func testEngineOperationsIntegration(t *testing.T) func(ctx context.Context, pla
 func TestEngineOperations_RealTimeData(t *testing.T) {
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "tn_ops_realtime_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testEngineOperationsRealTime(t),
 		},

extensions/tn_cache/internal/engine_ops_permissions_test.go

@@ -26,7 +26,7 @@ import (
 func TestExtensionAgentPermissions(t *testing.T) {
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "extension_agent_permissions_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testExtensionAgentAccess(t),
 		},

internal/benchmark/digest/digest_benchmark_test.go

@@ -198,7 +198,7 @@ func createBenchmarkSuiteFunc(t *testing.T, scale DigestBenchmarkScale) func(con
 func runDigestBenchmarkScale(t *testing.T, scale DigestBenchmarkScale) {
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        fmt.Sprintf("digest_benchmark_%s", scale.Name),
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			WithDigestBenchmarkSetup(createBenchmarkSuiteFunc(t, scale)),
 		},

internal/benchmark/load_test.go

@@ -241,9 +241,9 @@ func TestBench(t *testing.T) {
 		groupStartTime := time.Now()
 
 		schemaTest := kwilTesting.SchemaTest{
-			Name:          groupName,
-			FunctionTests: groupOfTests,
-			SeedScripts:   migrations.GetSeedScriptPaths(),
+			Name:           groupName,
+			FunctionTests:  groupOfTests,
+			SeedStatements: migrations.GetSeedScriptStatements(),
 		}
 
 		t.Run(schemaTest.Name, func(t *testing.T) {

internal/migrations/001-common-actions.sql

@@ -48,27 +48,46 @@ CREATE OR REPLACE ACTION create_stream(
 
 /**
  * create_streams: Creates multiple streams at once.
+ * Fee: 2 TRUF per stream created
+ * Exemption: system:network_writer role bypasses fee collection
  * Validates stream_id format, data provider address, and stream type.
  * Sets default metadata including type, owner, visibility, and readonly keys.
  */
 CREATE OR REPLACE ACTION create_streams(
     $stream_ids TEXT[],
     $stream_types TEXT[]
 ) PUBLIC {
+    -- ===== FEE COLLECTION WITH ROLE EXEMPTION =====
     $lower_caller TEXT := LOWER(@caller);
-    -- Permission Check: Ensure caller has the 'system:network_writer' role.
-    $has_permission BOOL := false;
-    for $row in are_members_of('system', 'network_writer', ARRAY[$lower_caller]) {
-        if $row.wallet = $lower_caller AND $row.is_member {
-            $has_permission := true;
-            break;
+
+    -- Check if caller is exempt (has system:network_writer role)
+    $is_exempt BOOL := FALSE;
+    FOR $row IN are_members_of('system', 'network_writer', ARRAY[$lower_caller]) {
+        IF $row.wallet = $lower_caller AND $row.is_member {
+            $is_exempt := TRUE;
+            BREAK;
         }
     }
-    if NOT $has_permission {
-        ERROR('Caller does not have the required system:network_writer role to create streams.');
+
+    -- Collect fee only from non-exempt wallets (2 TRUF per stream)
+    IF NOT $is_exempt {
+        $fee_per_stream := 2000000000000000000::NUMERIC(78, 0); -- 2 TRUF with 18 decimals
+        $num_streams := array_length($stream_ids);
+        $total_fee := $fee_per_stream * $num_streams::NUMERIC(78, 0);
+
+        $caller_balance := ethereum_bridge.balance(@caller);
+
+        IF $caller_balance < $total_fee {
+            ERROR('Insufficient balance for stream creation. Required: ' || ($num_streams * 2)::TEXT || ' TRUF for ' || $num_streams::TEXT || ' stream(s)');
+        }
+
+        $leader_addr TEXT := encode(@leader_sender, 'hex')::TEXT;
+        ethereum_bridge.transfer($leader_addr, $total_fee);
     }
+    -- ===== END FEE COLLECTION =====
 
-    -- Get caller's address (data provider) first
+    -- ===== STREAM CREATION LOGIC =====
+    -- Get caller's address (data provider)
     $data_provider TEXT := $lower_caller;
 
     -- Check if caller is a valid ethereum address

internal/migrations/migration.go

@@ -46,3 +46,64 @@ func GetSeedScriptPaths() []string {
 	sort.Strings(seedFilesPaths)
 	return seedFilesPaths
 }
+
+// GetSeedScriptStatements returns migration SQL statements with test-specific replacements.
+// This is used for test environments to replace production bridge namespace with test bridge.
+//
+// Replacement strategy:
+// - Production migrations use 'ethereum_bridge' (mainnet)
+// - Test environments need 'sepolia_bridge' (testnet)
+// - Runtime string replacement avoids duplicate migration files
+//
+// This implements the approach discussed to avoid maintaining separate test-only override files
+// that duplicate logic and can become a bug source if production and test logic diverges.
+func GetSeedScriptStatements() []string {
+	var statements []string
+
+	// Collect all SQL file paths in sorted order
+	var sqlPaths []string
+	err := fs.WalkDir(seedFiles, ".", func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			return nil
+		}
+		if !strings.HasSuffix(path, ".sql") {
+			return nil
+		}
+		if strings.HasSuffix(path, ".prod.sql") {
+			return nil
+		}
+		sqlPaths = append(sqlPaths, path)
+		return nil
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	if len(sqlPaths) == 0 {
+		panic("no SQL files found in embedded directory")
+	}
+
+	// Sort paths to ensure consistent loading order (0_test_only/ loads after production)
+	sort.Strings(sqlPaths)
+
+	// Read each file and apply test-specific replacements
+	for _, path := range sqlPaths {
+		content, err := seedFiles.ReadFile(path)
+		if err != nil {
+			panic("failed to read migration file " + path + ": " + err.Error())
+		}
+
+		sql := string(content)
+
+		// Replace production bridge namespace with test bridge namespace
+		// This allows tests to use the same migration logic without duplication
+		sql = strings.ReplaceAll(sql, "ethereum_bridge", "sepolia_bridge")
+
+		statements = append(statements, sql)
+	}
+
+	return statements
+}

tests/database_size/database_size_v2_test.go

@@ -35,7 +35,7 @@ import (
 func TestDatabaseSizeV2Actions(t *testing.T) {
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "database_size_v2_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testGetDatabaseSizeV2(t),
 			testGetDatabaseSizeV2Pretty(t),

tests/extensions/database-size/database_size_extension_test.go

@@ -27,7 +27,7 @@ func TestDatabaseSizeExtension(t *testing.T) {
 	// The database_size extension should be automatically available
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "database_size_extension_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testDatabaseSizeExtensionMethods(t),
 			testGetDatabaseSizeV2Actions(t),

tests/extensions/erc20/common_test.go

@@ -28,7 +28,7 @@ const (
 
 // seedAndRun is a helper that handles test execution with proper isolation
 func seedAndRun(t TestingT, name string, fn kwilTesting.TestFunc) {
-	seedScripts := migrations.GetSeedScriptPaths()
+	seedStatements := migrations.GetSeedScriptStatements()
 
 	// Wrap the test function to add singleton reset and cleanup
 	wrappedFn := func(ctx context.Context, platform *kwilTesting.Platform) error {
@@ -50,9 +50,9 @@ func seedAndRun(t TestingT, name string, fn kwilTesting.TestFunc) {
 	}
 
 	testutils.RunSchemaTest(t.(*testing.T), kwilTesting.SchemaTest{
-		Name:          name,
-		SeedScripts:   seedScripts,
-		FunctionTests: []kwilTesting.TestFunc{wrappedFn},
+		Name:           name,
+		SeedStatements: seedStatements,
+		FunctionTests:  []kwilTesting.TestFunc{wrappedFn},
 	}, &testutils.Options{Options: testutils.GetTestOptions()})
 }
 

tests/extensions/tn_cache/cache_height_tracking_test.go

@@ -32,7 +32,7 @@ func TestCacheHeightTracking(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "cache_height_tracking_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			func(ctx context.Context, platform *kwilTesting.Platform) error {
 				helper := cache.SetupCacheTest(ctx, platform, cacheConfig)

tests/extensions/tn_cache/cache_integration_test.go

@@ -36,7 +36,7 @@ func TestCacheIntegration(t *testing.T) {
 	// Run the test with cache enabled using the new wrapper
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "cache_integration_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testCacheBasicFunctionality(t, cacheConfig),
 		},
@@ -218,7 +218,7 @@ func TestCacheIncludeChildrenForNestedComposed(t *testing.T) {
 	// Run the test with cache enabled
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "cache_include_children_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testCacheIncludeChildren(t, cacheConfig),
 		},
@@ -240,7 +240,7 @@ func TestCacheBaseTimeVariants(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "cache_base_time_variants_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			testCacheBaseTimeVariants(t, cacheConfig, streamID, directiveBaseTime, lateFrom),
 		},

tests/extensions/tn_cache/cache_observability_test.go

@@ -31,7 +31,7 @@ func TestCacheObservability(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "cache_observability_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			func(ctx context.Context, platform *kwilTesting.Platform) error {
 				// Cache is already set up by the wrapper, but we need the helper for RefreshCache

tests/extensions/tn_cache/resolution_transaction_test.go

@@ -25,7 +25,7 @@ func TestResolutionInTransaction(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "resolution_transaction_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			func(ctx context.Context, platform *kwilTesting.Platform) error {
 				platform = procedure.WithSigner(platform, deployer.Bytes())

tests/streams/aggregation/aggr01_basic_aggregation_test.go

@@ -34,7 +34,7 @@ func TestAGGR01_BasicAggregation(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr01_basic_aggregation_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR01_BasicAggregation", testAGGR01_BasicAggregation),
 		},

tests/streams/aggregation/aggr02_weighted_contributions_test.go

@@ -34,7 +34,7 @@ func TestAGGR02_WeightedContributions(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr02_weighted_contributions_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR02_WeightedContributions", testAGGR02_WeightedContributions),
 		},

tests/streams/aggregation/aggr03_taxonomy_validity_periods_test.go

@@ -37,7 +37,7 @@ func TestAGGR03_TaxonomyValidityPeriods(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr03_taxonomy_validity_periods_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR03_TaxonomyValidityPeriods", testAGGR03_TaxonomyValidityPeriods),
 		},

tests/streams/aggregation/aggr04_missing_data_handling_test.go

@@ -36,7 +36,7 @@ func TestAGGR04_MissingDataHandling(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr04_missing_data_handling_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR04_MissingDataHandling", testAGGR04_MissingDataHandling),
 		},

tests/streams/aggregation/aggr05_no_duplicate_child_streams_test.go

@@ -30,7 +30,7 @@ func TestAGGR05_NoDuplicateChildStreams(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr05_no_duplicate_child_streams_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR05_NoDuplicateChildStreams", testAGGR05_NoDuplicateChildStreams),
 		},

tests/streams/aggregation/aggr06_single_active_taxonomy_test.go

@@ -33,7 +33,7 @@ func TestAGGR06_SingleActiveTaxonomy(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr06_single_active_taxonomy_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR06_SingleActiveTaxonomy", testAGGR06_SingleActiveTaxonomy),
 		},

tests/streams/aggregation/aggr07_inexistent_streams_test.go

@@ -29,7 +29,7 @@ func TestAGGR07_InexistentStreamsRejected(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr07_inexistent_streams_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR07_InexistentStreams", testAGGR07_InexistentStreams),
 		},

tests/streams/aggregation/aggr08_weight_change_test.go

@@ -41,7 +41,7 @@ func TestAGGR08_WeightChangeEventPoints(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr08_weight_change_event_points_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR08_WeightChangeEventPoints", testAGGR08_WeightChangeEventPoints),
 		},

tests/streams/aggregation/aggr09_duplicate_values_test.go

@@ -43,7 +43,7 @@ func TestAGGR09_DuplicateValues(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "aggr09_duplicate_values_test",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		FunctionTests: []kwilTesting.TestFunc{
 			wrapTestWithCacheModes(t, "AGGR09_DuplicateValues", testAGGR09_DuplicateValues),
 		},

tests/streams/attestation/attestation_request_test.go

@@ -25,7 +25,7 @@ func TestRequestAttestationInsertsCanonicalPayload(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "ATTESTATION01_RequestInsertion",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		Owner:       addrs.Owner.Address(),
 		FunctionTests: []kwilTesting.TestFunc{
 			func(ctx context.Context, platform *kwilTesting.Platform) error {

tests/streams/attestation/attestation_retrieval_test.go

@@ -20,7 +20,7 @@ func TestGetSignedAttestation(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "ATTESTATION02_GetSignedAttestation",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		Owner:       addrs.Owner.Address(),
 		FunctionTests: []kwilTesting.TestFunc{
 			func(ctx context.Context, platform *kwilTesting.Platform) error {
@@ -108,7 +108,7 @@ func TestListAttestations(t *testing.T) {
 
 	testutils.RunSchemaTest(t, kwilTesting.SchemaTest{
 		Name:        "ATTESTATION03_ListAttestations",
-		SeedScripts: migrations.GetSeedScriptPaths(),
+		SeedStatements: migrations.GetSeedScriptStatements(),
 		Owner:       addrs.Owner.Address(),
 		FunctionTests: []kwilTesting.TestFunc{
 			func(ctx context.Context, platform *kwilTesting.Platform) error {