Add Unit Test Cases

[pnguyen215]

Description:
This task involves creating comprehensive unit test cases for the existing codebase to improve test coverage and ensure correctness and stability. The goal is to validate core functionality, handle edge cases, and prevent regressions.

Scope of Work:

• Identify critical modules, functions, or components lacking unit tests
• Write unit test cases covering normal, boundary, and error scenarios
• Mock external dependencies where appropriate
• Ensure tests are readable, maintainable, and follow project testing standards
• Verify all tests pass in the CI pipeline

Acceptance Criteria:

• New unit tests are added for the targeted components
• Test coverage is measurably improved
• All existing and new tests pass successfully
• No production code behavior is altered unintentionally

[pnguyen215]

package main

import (
	"bytes"
	"compress/gzip"
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/sivaosorg/unify4g"
	"github.com/sivaosorg/wrapify"
	"github.com/sivaosorg/wrapify/pkg/coll"
	"github.com/sivaosorg/wrapify/pkg/common"
	"github.com/sivaosorg/wrapify/pkg/encoding"
	"github.com/sivaosorg/wrapify/pkg/hashy"
	"github.com/sivaosorg/wrapify/pkg/match"
	"github.com/sivaosorg/wrapify/pkg/randn"
	"github.com/sivaosorg/wrapify/pkg/ref"
	"github.com/sivaosorg/wrapify/pkg/strutil"
)

func main() {
	// Test01()
	// Test02()
	// Test03()
	// Test04()
	// Test01()
	// Test06()
	// Example_SimpleFileStreaming()
	// Example_DatabaseExportWithCompression()
	// Example_VideoStreaming()
	// Example_LogAggregation()
	// Example_ThrottledStreaming()
	// Example_ResumableDownload()
	// Example_APIStreamingWithErrors()
	// Example_StreamingWithTransformation()
	// Example_CloudStorageUpload()
	// Example_RealTimeMetricsExport()
	// Example_ImageGalleryStreaming()
	// Example_APIGatewayWithRateLimit()
	// Example_WebSocketStreaming()
	// Example_MobileAppDataSync()
	// Example_DistributedLogProcessing()
	// Example_MicroservicePipeline_FIXED_v2()
	// Example_DatabaseReplication_FIXED_v2()
	// Example_CompleteE2EScenario_FIXED_v2()

	fmt.Println(randn.TimeID())
	// fmt.Println(randn.CryptoID())

	// fmt.Println(randn.RandIntr(10000, 19999))
	// fmt.Println(randn.RandInt())
	fmt.Println(randn.RandUUID())
	// fmt.Println(randn.RandFt32())

	for i := 0; i < 5; i++ {
		fmt.Println(randn.RandIntr(10000, 19999))
	}

	fmt.Println(coll.Contains([]string{"a", "b", "c"}, "b"))
	fmt.Println(strutil.Capitalize("hello world from wrapify"))
	fmt.Println(strutil.Title("hello world from wrapify"))
	fmt.Println(ref.Ptr("dddd"))

	var User struct {
		ID   int
		Name string
	}

	fmt.Println(encoding.JsonSafe(User))
	fmt.Println(common.DeepEqual(10, 10))
	fmt.Println(match.Match("hello_wrapify_v1", "hello_*_v?"))

	raw := []byte(`{"b":2,"a":1,"arr":[1,2,3]}`)
	out := encoding.Pretty(raw)
	fmt.Print(string(out))

	opts := &encoding.OptionsConfig{
		Width:    60,
		Prefix:   "// ",
		Indent:   "    ",
		SortKeys: true,
	}

	out2 := encoding.PrettyOptions(raw, opts)
	fmt.Print(string(out2))

	dirty := []byte(" \n\t { \"a\" : 1 , \"b\" : 2 } \n")
	clean := encoding.Ugly(dirty)

	fmt.Println(string(clean))

	buf := []byte(" \n\t { \"x\" : 10 } ")
	encoding.UglyInPlace(buf)

	fmt.Println(string(buf))

	raw2 := []byte(`
{
  // user id
  "id": 1,
  "name": "Alice", /* comment */
}
`)

	valid := encoding.Spec(raw2)
	fmt.Println(string(valid))

	buf2 := []byte(`{ "a": 1, } // trailing comma`)
	encoding.SpecInPlace(buf2)

	fmt.Println(string(buf2))

	json := []byte(`{"name":"John","age":30,"active":true}`)
	colored := encoding.Color(json, encoding.VSCodeDarkStyle)

	fmt.Print(string(colored))

	raw3 := []byte(`
{
  // comment
  "z": 3,
  "a": 1,
  "b": 2,
}
`)

	// Step 1: Normalize input
	clean2 := encoding.Spec(raw3)

	// Step 2: Pretty-print with sorting
	opts2 := &encoding.OptionsConfig{
		SortKeys: true,
	}
	pretty := encoding.PrettyOptions(clean2, opts2)

	// Step 3: Colorize for terminal
	colored2 := encoding.Color(pretty, nil)

	fmt.Print(string(colored2))

	meta := wrapify.New()
	fmt.Println(meta.Json())
	fmt.Println(meta.Hash256Safe())
	meta.RandRequestID()
	fmt.Println(meta.Hash256Safe())
	fmt.Println(meta.Hash256Safe())
	fmt.Println(meta.Hash256Safe())
	fmt.Println(meta.HashSafe())
	meta.RandRequestID()
	fmt.Println(meta.HashSafe())
	fmt.Println(meta.HashSafe())

	// hash, err := hashy.Hash(opts2)
	// fmt.Println(hash, err)

	// hash2, err := hashy.Hash256(opts2)
	// fmt.Println(hash2, err)

	// hash3, err := hashy.HashBase10(opts2)
	// fmt.Println(hash3, err)

	// hash4, err := hashy.HashBase16(opts2)
	// fmt.Println(hash4, err)

	// hash5, err := hashy.HashEncoded(opts2)
	// fmt.Println(hash5, err)

	// hash6, err := hashy.HashBase32(opts2)
	// fmt.Println(hash6, err)

	// hash7, err := hashy.HashHexShort(opts2)
	// fmt.Println(hash7, err)

	// hash8, err := hashy.HashHex(opts2)
	// fmt.Println(hash8, err)

	created := wrapify.WrapCreated("hello world", "response body")
	fmt.Println(created.Json())
	h, wrap := created.Hash()
	fmt.Println(h, wrap.StatusCode())
	h, wrap = created.Hash()
	fmt.Println(h, wrap.StatusCode())
	fmt.Println(created.Hash256Safe())
	fmt.Println(created.Hash256Safe())
	fmt.Println(created.Hash256Safe())
	fmt.Println(created.Hash256Safe())
	fmt.Println(created.Hash256Safe())
	fmt.Println(created.HashSafe())
	fmt.Println(created.HashSafe())
	// fmt.Println(created.Hash256Safe())
	// fmt.Println(created.HashSafe())

	type Data struct {
		Name  string
		Age   int
		Email string
	}

	d1 := Data{Name: "Alice", Age: 30}
	d2 := Data{Name: "Alice", Age: 30, Email: ""}

	// Without IgnoreZeroValue
	hash1, _ := hashy.Hash(d1)
	hash2, _ := hashy.Hash(d2)
	fmt.Println("Without IgnoreZeroValue:", hash1, hash2)

	// With IgnoreZeroValue
	opts1 := hashy.NewOptions().WithIgnoreZeroValue(true).Build()
	hash3, _ := hashy.Hash(d1, opts1)
	hash4, _ := hashy.Hash(d2, opts1)
	fmt.Println("With IgnoreZeroValue:", hash3, hash4)
}

// ============================================================================
// USE CASE 1: Simple File Streaming (Most Common)
// ============================================================================
// When: You need to transfer a file and want basic streaming
// Advantages: Easy to implement, good performance, automatic progress tracking
// Disadvantages: Limited customization, single-threaded by default
// Best for: REST APIs returning large files, file downloads
// ============================================================================
func Example_SimpleFileStreaming() {
	// Scenario: API returns 1GB file to client

	// Create response wrapper
	response := wrapify.New().
		WithStatusCode(200).
		WithMessage("Streaming file content").
		WithPath("/api/download/large_file.bin").
		WithCustomFieldKV("file_name", "largefile.bin").
		WithCustomFieldKV("file_size", "1GB")

	// Mock file reader (in real scenario: os.Open("largefile.bin"))
	fileData := strings.Repeat("X", 1024*1024) // 1MB for demo
	fileReader := strings.NewReader(fileData)

	// Default config: 64KB chunks, no compression, buffered strategy
	streaming := response.AsStreaming(fileReader)

	// Start streaming with default context
	result := streaming.Start(context.Background())

	// Process streamed chunks
	fmt.Printf("Streaming started for:\n%v", result.JsonPretty())
}

// ============================================================================
// USE CASE 2: Large Database Export with Compression
// ============================================================================
// When: Exporting millions of records, need to minimize bandwidth
// Advantages: Reduces data size by 70-80% for text data, resumable
// Disadvantages: Higher CPU usage, slightly slower for small files
// Best for: Data exports, log backups, API responses with lots of text
// ============================================================================
func Example_DatabaseExportWithCompression() {
	// Scenario: Export 100M database records to CSV, 100GB total size

	response := wrapify.New().
		WithStatusCode(200).
		WithMessage("Exporting database records").
		WithPath("/api/export/users")

	// Mock data reader (in real scenario: queryDatabase())
	dbData := strings.Repeat(
		"id,name,email,created_at,status\n1,John,john@example.com,2024-01-01,active\n",
		6000000,
	)
	dbReader := strings.NewReader(dbData)

	// Create custom config: 1MB chunks + GZIP compression + buffered strategy
	config := &wrapify.StreamConfig{
		ChunkSize:           1024 * 1024, // 1MB chunks
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 4,
		UseBufferPool:       true,
		ReadTimeout:         30 * time.Second,
		WriteTimeout:        30 * time.Second,
		IsReceiving:         false,
	}

	streaming := response.WithStreaming(dbReader, config)

	// Set total bytes for accurate progress calculation
	streaming.WithTotalBytes(int64(len(dbData)))

	// Add progress tracking callback
	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		if err == nil && p.Percentage%10 == 0 {
			fmt.Printf("Export progress: %.1f%% | Rate: %.2f MB/s\n",
				float64(p.Percentage),
				float64(p.TransferRate)/1024/1024)
		}
	})

	// Start streaming
	result := streaming.Start(context.Background())

	// Get statistics after completion
	stats := streaming.GetStats()
	fmt.Printf("Export completed:\n%v\nStats:\n%v", result.JsonPretty(), stats.Json())
}

// ============================================================================
// USE CASE 3: Video Streaming (Buffered, Large Chunks)
// ============================================================================
// When: Streaming video/audio content, need smooth playback
// Advantages: Large chunks reduce overhead, good for high-bandwidth scenarios
// Disadvantages: Higher memory per chunk, less responsive pause/resume
// Best for: Video platforms (Netflix, YouTube), audio streaming (Spotify)
// ============================================================================
func Example_VideoStreaming() {
	response := wrapify.New().
		WithStatusCode(206). // 206 Partial Content (typical for video)
		WithMessage("Streaming video content").
		WithPath("/api/stream/video.mp4").
		WithCustomFieldKV("video_quality", "1080p").
		WithCustomFieldKV("codec", "h264")

	// Mock video reader (in real scenario: os.Open("video.mp4"))
	videoData := make([]byte, 10*1024*1024) // 10MB demo
	for i := range videoData {
		videoData[i] = byte(i % 256)
	}
	videoReader := bytes.NewReader((videoData))

	// Config: Large chunks (256KB for video), no compression (already compressed)
	config := &wrapify.StreamConfig{
		ChunkSize:           256 * 1024, // 256KB chunks
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_NONE, // Video already compressed
		MaxConcurrentChunks: 2,                 // Don't overload network
		UseBufferPool:       true,
		ReadTimeout:         60 * time.Second,
		WriteTimeout:        60 * time.Second,
		IsReceiving:         false,
		ThrottleRate:        5 * 1024 * 1024, // 5MB/s max rate
	}

	streaming := response.WithStreaming(videoReader, config)
	streaming.WithTotalBytes(int64(len(videoData)))

	// Real-time progress for UI update
	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		if err == nil {
			fmt.Printf("\rPlaying: %.0f%% | %s",
				float64(p.Percentage),
				p.EstimatedTimeRemaining.String())
		}
	})

	streaming.WithHook(func(p *wrapify.StreamProgress, w *wrapify.R) {
		if w.IsSuccess() {
			fmt.Printf("\rHook Playing: %.0f%% | %s",
				float64(p.Percentage),
				p.EstimatedTimeRemaining.String())
		}
	})

	result := streaming.Start(context.Background())

	stats := streaming.GetStats()
	// return result.
	// 	WithMessage("Video streaming completed").
	// 	WithDebuggingKV("total_chunks", stats.TotalChunks).
	// 	WithDebuggingKV("avg_bandwidth_mbps", fmt.Sprintf("%.2f", float64(stats.AverageBandwidth)/1024/1024))
	fmt.Printf("\nStreaming completed:\n%v\nStats:\n%v\n", result.JsonPretty(), stats.Json())

	progressStats := streaming.GetProgress()
	fmt.Printf("Final Progress Stats:\n%v", progressStats.Json())
}

// ============================================================================
// USE CASE 4: Real-time Log Aggregation (Concurrent Processing)
// ============================================================================
// When: Processing logs from multiple servers simultaneously
// Advantages: Parallel processing, fast aggregation, supports many sources
// Disadvantages: Complex error handling, requires good coordination
// Best for: Log aggregation, real-time analytics, data pipeline processing
// ============================================================================
func Example_LogAggregation() {
	response := wrapify.New().
		WithStatusCode(200).
		WithMessage("Aggregating logs from servers").
		WithPath("/api/logs/aggregate")

	// Mock log data from multiple servers
	logData := strings.Repeat(
		"[2024-11-14 10:00:00] INFO User login\n"+
			"[2024-11-14 10:00:01] WARN Connection timeout\n"+
			"[2024-11-14 10:00:02] ERROR Database query failed\n",
		1000000,
	)
	logReader := strings.NewReader(logData)

	// Config: Medium chunks, GZIP compression, max concurrency
	config := &wrapify.StreamConfig{
		ChunkSize:           512 * 1024, // 512KB chunks
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 8, // Process 8 chunks in parallel
		UseBufferPool:       true,
		ReadTimeout:         30 * time.Second,
		WriteTimeout:        30 * time.Second,
		// ThrottleRate:        2 * 1024 * 1024, // 10MB/s max rate
	}

	streaming := response.WithStreaming(logReader, config)
	streaming.WithTotalBytes(int64(len(logData)))

	// Track errors during log processing
	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		if err != nil {
			fmt.Printf("Error processing chunk %d: %v\n", p.CurrentChunk, err)
		} else {
			if p.Percentage%20 == 0 {
				fmt.Printf("[Callback] Log aggregation progress: %.1f%%\n", float64(p.Percentage))
			}
		}
	})

	streaming.WithHook(func(p *wrapify.StreamProgress, w *wrapify.R) {
		if w.IsSuccess() {
			if p.Percentage%20 == 0 {
				fmt.Printf("[Hook] Log aggregation progress: %.1f%%\n", float64(p.Percentage))
			}
		}
	})

	result := streaming.Start(context.Background())

	// Check if there were any errors
	if streaming.HasErrors() {
		errors := streaming.Errors()
		// return result.
		// 	WithError(fmt.Sprintf("Log aggregation had %d errors", len(errors))).
		// 	WithStatusCode(207) // Multi-Status: partially successful
		fmt.Printf("Log aggregation had %d errors\n", len(errors))
		return
	}

	stats := streaming.GetStats()
	// return result.
	// 	WithMessage("Logs aggregated successfully").
	// 	WithTotal(stats.TotalChunks).
	// 	WithDebuggingKV("total_log_bytes", stats.TotalBytes)
	fmt.Printf("Logs aggregated successfully:\n%v\nStats:\n%v\n", result.JsonPretty(), stats.Json())
}

// ============================================================================
// USE CASE 5: Bandwidth-Limited Environment (Throttling)
// ============================================================================
// When: Streaming to mobile clients or low-bandwidth networks
// Advantages: Predictable bandwidth, prevents network flooding, fair sharing
// Disadvantages: Slower transfer, longer wait times
// Best for: Mobile streaming, IoT devices, satellite connections
// ============================================================================
func Example_ThrottledStreaming() {
	response := wrapify.New().
		WithStatusCode(200).
		WithMessage("Streaming with throttling").
		WithPath("/api/download/mobile")

	fileData := strings.Repeat("X", 50*1024*1024) // 50MB
	fileReader := strings.NewReader(fileData)

	// Config: Small chunks, aggressive throttling
	config := &wrapify.StreamConfig{
		ChunkSize:     32 * 1024, // 32KB chunks
		Strategy:      wrapify.STRATEGY_DIRECT,
		Compression:   wrapify.COMP_GZIP,
		ThrottleRate:  512 * 1024, // 512KB/s limit
		UseBufferPool: false,      // Not needed for small chunks
		ReadTimeout:   30 * time.Second,
		WriteTimeout:  30 * time.Second,
	}

	streaming := response.WithStreaming(fileReader, config)
	streaming.WithTotalBytes(int64(len(fileData)))

	// Show ETA and bandwidth
	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		if err == nil && p.CurrentChunk%10 == 0 {
			fmt.Printf("Speed: %.2f KB/s | ETA: %s\n",
				float64(p.TransferRate)/1024,
				p.EstimatedTimeRemaining.String())
		}
	})

	result := streaming.Start(context.Background())

	// return result.
	// 	WithMessage("Throttled streaming completed").
	// 	WithDebuggingKV("throttle_rate_kbps", 512)
	fmt.Printf("Throttled streaming completed:\n%v\n", result.JsonPretty())
}

// ============================================================================
// USE CASE 6: Resumable Download (With Checkpointing)
// ============================================================================
// When: Downloads can fail and need to resume from last position
// Advantages: Resume capability, save bandwidth, user-friendly
// Disadvantages: Need checkpoint storage, more complexity
// Best for: Large software downloads, cloud backups, file synchronization
// ============================================================================
func Example_ResumableDownload() {
	response := wrapify.New().
		WithStatusCode(200).
		WithMessage("Resumable download initialized").
		WithPath("/api/download/largefile.iso")

	fileData := strings.Repeat("ISO_DATA_", 50*1024*1024) // Simulate ISO file
	fileReader := strings.NewReader(fileData)

	config := &wrapify.StreamConfig{
		ChunkSize:     256 * 1024, // 256KB checkpoints
		Strategy:      wrapify.STRATEGY_CHUNKED,
		Compression:   wrapify.COMP_NONE,
		UseBufferPool: false,
		ReadTimeout:   30 * time.Second,
		WriteTimeout:  30 * time.Second,
	}

	streaming := response.WithStreaming(fileReader, config)
	streaming.WithTotalBytes(int64(len(fileData)))

	checkpoints := make(map[int64]int64) // chunkNum -> offset

	// Save checkpoint every 10 chunks
	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		if p.CurrentChunk%10 == 0 {
			checkpoints[p.CurrentChunk] = p.TransferredBytes
			fmt.Printf("Checkpoint: Chunk %d at offset %d\n",
				p.CurrentChunk,
				p.TransferredBytes)
		}
	})

	result := streaming.Start(context.Background())

	// return result.
	// 	WithMessage("Download completed with checkpoints").
	// 	WithDebuggingKV("total_checkpoints", len(checkpoints)).
	// 	WithDebuggingKV("resumable", true)

	fmt.Printf("Download completed with checkpoints:\n%v\n", result.JsonPretty())
}

// ============================================================================
// USE CASE 7: API Response with Error Handling
// ============================================================================
// When: Building REST API that returns large data with proper error handling
// Advantages: Chainable API, consistent error handling, type-safe
// Disadvantages: Need to handle streaming errors explicitly
// Best for: Production REST APIs, data export endpoints, bulk operations
// ============================================================================
func Example_APIStreamingWithErrors() {
	response := wrapify.New().
		WithStatusCode(200).
		WithPath("/api/export/users").
		WithCustomFieldKV("format", "csv")

	// Simulate reading from database with potential errors
	mockedData := "id,name,email\n1,John,john@example.com\n2,Jane,jane@example.com\n"
	fileReader := strings.NewReader(mockedData)

	config := wrapify.NewStreamConfig()
	config.ChunkSize = 1024
	config.Compression = wrapify.COMP_GZIP
	config.Strategy = wrapify.STRATEGY_BUFFERED

	// Create streaming wrapper
	streaming := response.WithStreaming(fileReader, config)
	streaming.WithTotalBytes(int64(len(mockedData)))

	// Start streaming and get result
	result := streaming.Start(context.Background())

	// Check if streaming was successful
	if streaming.HasErrors() {
		errors := streaming.Errors()
		// return result.
		// 	WithError(fmt.Sprintf("Streaming failed with %d errors", len(errors))).
		// 	WithStatusCode(500)
		fmt.Printf("Streaming failed with %d errors\n - Cause: %v", len(errors), streaming.Cause().Error())
		return
	}

	// Successful streaming - add statistics
	stats := streaming.GetStats()
	// return result.
	// 	WithMessage("Export completed successfully").
	// 	WithStatusCode(200).
	// 	WithDebuggingKV("exported_bytes", stats.TotalBytes).
	// 	WithDebuggingKV("chunks", stats.TotalChunks).
	// 	WithDebuggingKV("duration_ms", stats.EndTime.Sub(stats.StartTime).Milliseconds())
	fmt.Printf("Export completed successfully:\n%v\nStats:\n%v\n", result.JsonPretty(), stats.Json())
}

// ============================================================================
// USE CASE 8: Advanced - Piping Between Readers and Writers
// ============================================================================
// When: Transform data while streaming (e.g., CSV -> JSON)
// Advantages: Memory efficient, real-time transformation, no intermediate files
// Disadvantages: Complex implementation, hard to debug
// Best for: Data format conversion, ETL pipelines, data filtering
// ============================================================================
func Example_StreamingWithTransformation() {
	response := wrapify.New().
		WithStatusCode(200).
		WithMessage("Streaming with transformation").
		WithPath("/api/convert/csv-to-json")

	csvData := "id,name\n1,John\n2,Jane\n"
	csvReader := strings.NewReader(csvData)

	// Output buffer (in real scenario: file or network connection)
	var jsonOutput bytes.Buffer

	config := &wrapify.StreamConfig{
		ChunkSize:           256,
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_NONE,
		MaxConcurrentChunks: 1, // Single-threaded for ordered transformation
	}

	streaming := response.WithStreaming(csvReader, config)
	streaming.WithWriter(&jsonOutput)
	streaming.WithTotalBytes(int64(len(csvData)))

	// Start transformation
	result := streaming.Start(context.Background())

	if streaming.HasErrors() {
		// return result.WithError("Transformation failed")
		fmt.Println("Transformation failed")
		return
	}

	// return result.
	// 	WithMessage("Transformation completed").
	// 	WithBody(jsonOutput.String())
	fmt.Printf("Transformation completed:\n%v\n Json out: \n %v", result.JsonPretty(), jsonOutput.String())
}

// ============================================================================
// USE CASE 9: Cloud Storage Upload (AWS S3, Google Cloud, Azure)
// ============================================================================
// Scenario: Upload file to cloud storage with retry logic and multipart
// When: Migrating data to cloud, backup to cloud storage
// Challenges: Network issues, large files, concurrent uploads
// ============================================================================
func Example_CloudStorageUpload() {

	// Simulate reading from local file
	localFile := "database_backup_2025_11_15.sql.gz"
	fileSize := int64(5 * 1024 * 1024 * 1024) // 5GB

	// Create mock file reader
	mockData := strings.Repeat("SQL_DATA|", int(fileSize/9))
	fileReader := strings.NewReader(mockData)

	response := wrapify.New().
		WithStatusCode(200).
		WithPath("/api/upload/s3").
		WithMessage("Uploading to cloud storage").
		WithCustomFieldKV("destination", "s3://my-bucket/backups/").
		WithCustomFieldKV("file_name", localFile).
		WithCustomFieldKV("file_size_gb", "5")

	// Cloud upload config: Large chunks for efficiency
	config := &wrapify.StreamConfig{
		ChunkSize:           10 * 1024 * 1024, // 10MB chunks (good for S3 multipart)
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_NONE, // Already compressed (.gz)
		MaxConcurrentChunks: 4,                 // 4 parallel uploads
		UseBufferPool:       true,
		ReadTimeout:         60 * time.Second,
		WriteTimeout:        60 * time.Second,
	}

	streaming := response.WithStreaming(fileReader, config)
	streaming.WithTotalBytes(fileSize)

	// Track upload progress with detailed metrics
	var uploadStats struct {
		sync.Mutex
		startTime    time.Time
		lastSpeed    float64
		speedHistory []float64
		failedChunks []int64
		retryCount   int
	}
	uploadStats.startTime = time.Now()

	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		uploadStats.Lock()
		defer uploadStats.Unlock()

		if err != nil {
			fmt.Printf("❌ Chunk %d failed: %v (Retrying...)\n", p.CurrentChunk, err)
			uploadStats.retryCount++
			uploadStats.failedChunks = append(uploadStats.failedChunks, p.CurrentChunk)
		} else if p.CurrentChunk%50 == 0 { // Every 50 chunks
			speed := float64(p.TransferRate) / 1024 / 1024
			uploadStats.speedHistory = append(uploadStats.speedHistory, speed)
			uploadStats.lastSpeed = speed

			fmt.Printf("📤 Upload: %.1f%% | Speed: %.2f MB/s | ETA: %s | Chunk: %d\n",
				float64(p.Percentage),
				speed,
				p.EstimatedTimeRemaining.String(),
				p.CurrentChunk)
		}
	})

	result := streaming.Start(context.Background())

	if !result.IsSuccess() {
		fmt.Printf("❌ Upload failed: %s\n", result.Error())
		return
	}

	stats := streaming.GetStats()
	uploadStats.Lock()
	fmt.Printf("\n✅ Upload completed successfully!\n")
	fmt.Printf("   Total bytes: %.2f GB\n", float64(stats.TotalBytes)/1024/1024/1024)
	fmt.Printf("   Total chunks: %d\n", stats.TotalChunks)
	fmt.Printf("   Duration: %.2f seconds\n", stats.EndTime.Sub(stats.StartTime).Seconds())
	fmt.Printf("   Average speed: %.2f MB/s\n", float64(stats.AverageBandwidth)/1024/1024)
	fmt.Printf("   Retries: %d\n", uploadStats.retryCount)
	uploadStats.Unlock()
}

// ============================================================================
// USE CASE 11: Real-Time Metrics Export (Time-Series Data)
// ============================================================================
// Scenario: Export metrics from monitoring system (Prometheus, InfluxDB)
// When: Real-time dashboards, performance analysis, audit logging
// ============================================================================
func Example_RealTimeMetricsExport() {

	response := wrapify.New().
		WithStatusCode(200).
		WithPath("/api/metrics/export").
		WithMessage("Exporting real-time metrics").
		WithCustomFieldKV("source", "prometheus").
		WithCustomFieldKV("time_range", "last_24_hours")

	// Simulate metrics data (timestamp | metric | value)
	baseTime := time.Now().Add(-24 * time.Hour)
	var metricsData strings.Builder

	// Generate 86400 data points (one per second for 24 hours)
	for i := 0; i < 86400; i++ {
		t := baseTime.Add(time.Duration(i) * time.Second)
		metricsData.WriteString(fmt.Sprintf("%d|cpu_usage|%.2f\n", t.Unix(), float64(i%100)))
		metricsData.WriteString(fmt.Sprintf("%d|memory_mb|%.2f\n", t.Unix(), float64((i*512)%8192)))
		metricsData.WriteString(fmt.Sprintf("%d|requests_per_sec|%d\n", t.Unix(), (i*10)%5000))
	}

	metricsReader := strings.NewReader(metricsData.String())

	config := &wrapify.StreamConfig{
		ChunkSize:           256 * 1024, // 256KB chunks
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 4,
		UseBufferPool:       true,
	}

	streaming := response.WithStreaming(metricsReader, config)
	streaming.WithTotalBytes(int64(metricsData.Len()))

	// Track metrics export with detailed stats
	var lastTime time.Time
	lastTime = time.Now()

	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		if err == nil && time.Since(lastTime) > 5*time.Second {
			lastTime = time.Now()
			fmt.Printf("📊 Export: %.1f%% | %d data points | Speed: %.2f MB/s | ETA: %s\n",
				float64(p.Percentage),
				p.CurrentChunk*512, // Approximate data points per chunk
				float64(p.TransferRate)/1024/1024,
				p.EstimatedTimeRemaining.String())
		}
	})

	result := streaming.Start(context.Background())

	if !result.IsSuccess() {
		fmt.Printf("❌ Export failed: %s\n", result.Error())
		return
	}

	stats := streaming.GetStats()
	fmt.Printf("✅ Metrics exported successfully!\n")
	fmt.Printf("   Total data points: ~%d\n", metricsData.Len()/50)
	fmt.Printf("   Original size: %.2f MB\n", float64(stats.TotalBytes)/1024/1024)
	fmt.Printf("   Compressed size: %.2f MB\n", float64(stats.CompressedBytes)/1024/1024)
	fmt.Printf("   Compression ratio: %.1f%%\n", (1-stats.CompressionRatio)*100)
	fmt.Printf("   Duration: %.2f seconds\n", stats.EndTime.Sub(stats.StartTime).Seconds())
}

// ============================================================================
// USE CASE 12: Image Gallery Streaming (Concurrent, Large Chunks)
// ============================================================================
// Scenario: Stream gallery of photos with various resolutions
// When: Image galleries, photo albums, media libraries
// ============================================================================
func Example_ImageGalleryStreaming() {

	response := wrapify.New().
		WithStatusCode(200).
		WithPath("/api/gallery/download").
		WithMessage("Streaming image gallery").
		WithCustomFieldKV("gallery_name", "vacation_2024").
		WithCustomFieldKV("image_count", "500").
		WithCustomFieldKV("total_size_gb", "50")

	// Simulate 50GB of images (500 images × 100MB average)
	const imageCount = 500
	const avgImageSize = 100 * 1024 * 1024 // 100MB per image
	// const totalSize = imageCount * avgImageSize

	// Create mock image data
	var galleryData strings.Builder
	for i := 0; i < imageCount; i++ {
		// Simulate image header + data
		galleryData.WriteString(fmt.Sprintf("IMG_%04d.jpg|%d|", i+1, avgImageSize))
		for j := 0; j < avgImageSize/1000; j++ {
			galleryData.WriteString("IMAGE_DATA_BLOCK_")
		}
		galleryData.WriteString("\n")
	}

	dataStr := galleryData.String()
	galleryReader := strings.NewReader(dataStr)

	config := &wrapify.StreamConfig{
		ChunkSize:           5 * 1024 * 1024, // 5MB chunks (balance for images)
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_NONE, // Images already compressed
		MaxConcurrentChunks: 6,                 // 6 concurrent downloads
		UseBufferPool:       true,
		ThrottleRate:        50 * 1024 * 1024, // 50MB/s max (reasonable for users)
	}

	streaming := response.WithStreaming(galleryReader, config)
	streaming.WithTotalBytes(int64(len(dataStr)))

	// Track which images have been downloaded
	var downloadProgress struct {
		sync.Mutex
		imagesDownloaded int
		lastChunk        int64
	}

	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		downloadProgress.Lock()
		defer downloadProgress.Unlock()

		if err != nil {
			fmt.Printf("⚠️  Error downloading image chunk %d: %v\n", p.CurrentChunk, err)
		} else {
			// Rough estimate of how many images have been downloaded
			imagesDownloaded := int(p.TransferredBytes / avgImageSize)
			if imagesDownloaded > downloadProgress.imagesDownloaded {
				downloadProgress.imagesDownloaded = imagesDownloaded
				fmt.Printf("🖼️  Downloaded %d/%d images | Speed: %.1f MB/s | ETA: %s\n",
					imagesDownloaded,
					imageCount,
					float64(p.TransferRate)/1024/1024,
					p.EstimatedTimeRemaining.String())
			}
		}
	})

	startTime := time.Now()
	result := streaming.Start(context.Background())
	duration := time.Since(startTime)

	if !result.IsSuccess() {
		fmt.Printf("❌ Gallery download failed: %s\n", result.Error())
		return
	}

	stats := streaming.GetStats()
	fmt.Printf("\n✅ Gallery download completed!\n")
	fmt.Printf("   Total images: %d\n", imageCount)
	fmt.Printf("   Total size: %.2f GB\n", float64(stats.TotalBytes)/1024/1024/1024)
	fmt.Printf("   Duration: %.2f minutes\n", duration.Minutes())
	fmt.Printf("   Average speed: %.2f MB/s\n", float64(stats.AverageBandwidth)/1024/1024)
}

// ============================================================================
// USE CASE 14: API Gateway with Rate Limiting
// ============================================================================
// Scenario: API Gateway streams large responses to multiple clients
// When: API aggregation, response streaming, rate limiting per tenant
// ============================================================================
func Example_APIGatewayWithRateLimit() {

	// Simulate multiple clients with different rate limits
	clients := []struct {
		name      string
		rateLimit int64
	}{
		{"Premium Client", 100 * 1024 * 1024}, // 100 MB/s
		{"Standard Client", 50 * 1024 * 1024}, // 50 MB/s
		{"Economy Client", 10 * 1024 * 1024},  // 10 MB/s
	}

	largeData := strings.Repeat("API_RESPONSE|", 10*1024*1024) // 10MB data

	for _, client := range clients {
		fmt.Printf("\n📍 Serving: %s (Rate limit: %.0f MB/s)\n", client.name, float64(client.rateLimit)/1024/1024)

		response := wrapify.New().
			WithStatusCode(200).
			WithPath("/api/gateway/large-response").
			WithCustomFieldKV("client", client.name).
			WithCustomFieldKV("rate_limit_mbps", fmt.Sprintf("%.0f", float64(client.rateLimit)/1024/1024))

		dataReader := strings.NewReader(largeData)

		config := &wrapify.StreamConfig{
			ChunkSize:    256 * 1024,
			Strategy:     wrapify.STRATEGY_DIRECT,
			Compression:  wrapify.COMP_GZIP,
			ThrottleRate: client.rateLimit,
			ReadTimeout:  30 * time.Second,
			WriteTimeout: 30 * time.Second,
		}

		streaming := response.WithStreaming(dataReader, config)
		streaming.WithTotalBytes(int64(len(largeData)))

		startTime := time.Now()
		streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
			if err == nil && p.Percentage%25 == 0 {
				actualSpeed := float64(p.TransferRate) / 1024 / 1024
				fmt.Printf("  %.0f%% | Speed: %.2f MB/s | ETA: %s\n",
					float64(p.Percentage),
					actualSpeed,
					p.EstimatedTimeRemaining.String())
			}
		})

		result := streaming.Start(context.Background())
		duration := time.Since(startTime)

		stats := streaming.GetStats()
		fmt.Printf("  ✓ Completed in %.2f seconds (Speed: %.2f MB/s)\n",
			duration.Seconds(),
			float64(stats.AverageBandwidth)/1024/1024)

		if !result.IsSuccess() {
			fmt.Printf("  ❌ Error: %s\n", result.Error())
		}

		fmt.Printf("  Result:\n%v\n", result.JsonPretty())
	}
}

// ============================================================================
// USE CASE 16: WebSocket Streaming (Real-time Communication)
// ============================================================================
// Scenario: Stream data over WebSocket connection
// When: Real-time notifications, live updates, bidirectional streaming
// ============================================================================
func Example_WebSocketStreaming() {

	response := wrapify.New().
		WithStatusCode(200).
		WithPath("/ws/stream").
		WithMessage("WebSocket streaming initiated")

	// Simulate live event stream
	var eventStream strings.Builder
	baseTime := time.Now()

	for i := 0; i < 10000; i++ {
		timestamp := baseTime.Add(time.Duration(i) * time.Millisecond)
		eventStream.WriteString(fmt.Sprintf("EVENT|%d|user_%d|action_%s|%.2f\n",
			timestamp.Unix(),
			i%100,
			[]string{"click", "scroll", "submit", "hover"}[i%4],
			float64(i%100)))
	}

	wsReader := strings.NewReader(eventStream.String())

	config := &wrapify.StreamConfig{
		ChunkSize:    4 * 1024, // Small chunks for real-time (4KB)
		Strategy:     wrapify.STRATEGY_DIRECT,
		Compression:  wrapify.COMP_GZIP,
		ReadTimeout:  5 * time.Second,
		WriteTimeout: 5 * time.Second,
	}

	streaming := response.WithStreaming(wsReader, config)
	streaming.WithTotalBytes(int64(eventStream.Len()))

	frameCount := 0
	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		frameCount++
		if err == nil && frameCount%50 == 0 {
			fmt.Printf("🔴 LIVE: %d frames | %.1f%% | %.0f fps\n",
				frameCount,
				float64(p.Percentage),
				float64(frameCount)/p.ElapsedTime.Seconds())
		}
	})

	startTime := time.Now()
	result := streaming.Start(context.Background())
	totalDuration := time.Since(startTime)

	stats := streaming.GetStats()
	fmt.Printf("\n✓ WebSocket streaming completed\n")
	fmt.Printf("  Total frames: %d\n", frameCount)
	fmt.Printf("  Duration: %.2f seconds\n", totalDuration.Seconds())
	fmt.Printf("  Average FPS: %.0f\n", float64(frameCount)/totalDuration.Seconds())
	fmt.Printf("  Throughput: %.2f MB/s\n", float64(stats.AverageBandwidth)/1024/1024)

	fmt.Printf("Result:\n%v\n", result.JsonPretty())
}

// ============================================================================
// USE CASE 17: Mobile App Data Sync
// ============================================================================
// Scenario: Sync data between mobile app and backend
// When: Offline-first apps, sync on reconnect, bandwidth optimization
// ============================================================================
func Example_MobileAppDataSync() {

	response := wrapify.New().
		WithStatusCode(200).
		WithPath("/mobile/sync").
		WithMessage("Syncing mobile data").
		WithCustomFieldKV("device_id", "DEVICE_12345").
		WithCustomFieldKV("last_sync", "2025-11-14T10:00:00Z")

	// Simulate mobile app data (contacts, calendar, notes, etc.)
	var mobileData strings.Builder
	mobileData.WriteString("CONTACTS|100\n")
	for i := 0; i < 100; i++ {
		mobileData.WriteString(fmt.Sprintf("CONTACT|%d|name|email|phone\n", i))
	}
	mobileData.WriteString("CALENDAR|50\n")
	for i := 0; i < 50; i++ {
		mobileData.WriteString(fmt.Sprintf("EVENT|%d|title|date|location\n", i))
	}
	mobileData.WriteString("NOTES|200\n")
	for i := 0; i < 200; i++ {
		mobileData.WriteString(fmt.Sprintf("NOTE|%d|content|tags\n", i))
	}

	dataReader := strings.NewReader(mobileData.String())

	// Mobile config: Very aggressive compression, small chunks
	config := &wrapify.StreamConfig{
		ChunkSize:           16 * 1024, // 16KB for mobile
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 2,
		UseBufferPool:       false,
		ThrottleRate:        1 * 1024 * 1024, // 1MB/s limit
	}

	streaming := response.WithStreaming(dataReader, config)
	streaming.WithTotalBytes(int64(mobileData.Len()))

	var syncStats struct {
		sync.Mutex
		itemsSynced int
	}

	streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
		syncStats.Lock()
		defer syncStats.Unlock()

		if err == nil {
			// Estimate items synced
			syncStats.itemsSynced = int(p.TransferredBytes / 100)

			if p.Percentage%10 == 0 {
				batteryEstimate := 100.0 - (float64(p.Percentage) * 0.1) // Rough estimate
				fmt.Printf("📱 Sync: %.0f%% | Items: %d | Battery: %.0f%% | Speed: %.1f KB/s\n",
					float64(p.Percentage),
					syncStats.itemsSynced,
					batteryEstimate,
					float64(p.TransferRate)/1024)
			}
		}
	})

	result := streaming.Start(context.Background())

	if !result.IsSuccess() {
		fmt.Printf("❌ Sync failed: %s\n", result.Error())
		return
	}

	stats := streaming.GetStats()
	fmt.Printf("\n✅ Mobile sync completed!\n")
	fmt.Printf("   Items synced: %d (contacts) + %d (events) + %d (notes)\n", 100, 50, 200)
	fmt.Printf("   Original size: %.2f MB\n", float64(stats.TotalBytes)/1024/1024)
	fmt.Printf("   Compressed size: %.2f MB\n", float64(stats.CompressedBytes)/1024/1024)
	fmt.Printf("   Data saved: %.2f MB\n", float64(stats.TotalBytes-stats.CompressedBytes)/1024/1024)
}

// ============================================================================
// USE CASE 18: Distributed Log Processing
// ============================================================================
// Scenario: Collect and process logs from multiple servers
// When: Centralized logging, log aggregation, real-time log analysis
// ============================================================================
func Example_DistributedLogProcessing() {

	servers := []string{"web-server-01", "api-server-02", "db-server-03", "cache-server-04"}

	var aggregatedLogs bytes.Buffer
	totalProcessed := int64(0)

	for _, server := range servers {
		fmt.Printf("\n📍 Processing logs from: %s\n", server)

		logResponse := wrapify.New().
			WithStatusCode(200).
			WithPath("/logs/collect/" + server)

		// Simulate server logs
		logTime := time.Now().Add(-1 * time.Hour)
		var logsBuffer strings.Builder

		levels := []string{"DEBUG", "INFO", "WARN", "ERROR"}
		for i := 0; i < 100000; i++ {
			logsBuffer.WriteString(fmt.Sprintf("[%s] %s - Server %s - Message %d\n",
				logTime.Add(time.Duration(i)*time.Second).Format("2006-01-02 15:04:05"),
				levels[i%4],
				server,
				i))
		}

		logReader := strings.NewReader(logsBuffer.String())

		config := &wrapify.StreamConfig{
			ChunkSize:           256 * 1024,
			Strategy:            wrapify.STRATEGY_BUFFERED,
			Compression:         wrapify.COMP_GZIP,
			MaxConcurrentChunks: 3,
		}

		streaming := logResponse.WithStreaming(logReader, config)
		streaming.WithTotalBytes(int64(logsBuffer.Len()))
		streaming.WithWriter(&aggregatedLogs)

		streaming.WithCallback(func(p *wrapify.StreamProgress, err error) {
			if err == nil && p.Percentage%50 == 0 {
				fmt.Printf("  %.0f%% compressed from %.2f MB to %.2f MB\n",
					float64(p.Percentage),
					float64(p.TotalBytes)/1024/1024,
					float64(p.TransferredBytes)/1024/1024)
			}
		})

		result := streaming.Start(context.Background())
		stats := streaming.GetStats()

		totalProcessed += stats.TotalBytes

		fmt.Printf("  ✓ Collected %d log lines\n", 100000)
		fmt.Printf("    Compression: %.2f MB → %.2f MB\n",
			float64(stats.TotalBytes)/1024/1024,
			float64(stats.CompressedBytes)/1024/1024)
		fmt.Printf("Result:\n%v\n", result.JsonPretty())
		fmt.Printf("Aggregated log size so far: %.2f MB\n", float64(aggregatedLogs.Len())/1024/1024)
		// print full content of aggregatedLogs if needed
		// fmt.Printf("Sample log entry:\n%v\n", aggregatedLogs.String()[:200])
	}

	fmt.Printf("\n📊 Log Processing Summary:\n")
	fmt.Printf("   Total logs collected from %d servers\n", len(servers))
	fmt.Printf("   Total log data: %.2f MB\n", float64(totalProcessed)/1024/1024)
}

// ============================================================================
// USE CASE 10: Microservice Pipeline (Chain Multiple Services) - FIXED v2
// ============================================================================
func Example_MicroservicePipeline_FIXED_v2() {

	// Service 1: Producer - Generate data
	fmt.Println("📍 Service 1: Data Producer (Database)")

	// Simulate database export
	rawData := "EMPLOYEE|001|John Doe|john@company.com|50000\n" +
		"EMPLOYEE|002|Jane Smith|jane@company.com|55000\n" +
		"EMPLOYEE|003|Bob Johnson|bob@company.com|48000\n"

	// Repeat for 1000 employees
	producerData := strings.Repeat(rawData, 333) // ~1KB

	producerReader := strings.NewReader(producerData)
	producerBuffer := bytes.NewBuffer([]byte{})

	producerResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/service1/data/extract")

	// Important: WithStreaming returns *wrapper, not *StreamingWrapper
	// We need to get the streaming wrapper separately
	_ = producerResponse.WithStreaming(producerReader, nil)

	// Get the actual streaming wrapper to use streaming methods
	// Note: We'll use basic streaming without all the advanced features
	fmt.Println("  Exporting data from database...")

	// Create a simple copy (since WithStreaming returns wrapper)
	_, _ = io.Copy(producerBuffer, producerReader)

	fmt.Printf("  ✓ Producer exported %d bytes\n", len(producerData))

	// Service 2: Transform - Process and transform data
	fmt.Println("\n📍 Service 2: Data Transformer (Validation & Enrichment)")

	transformedBuffer := bytes.NewBuffer([]byte{})
	transformReader := bytes.NewReader(producerBuffer.Bytes())

	transformResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/service2/data/transform")

	config := &wrapify.StreamConfig{
		ChunkSize:   512,
		Compression: wrapify.COMP_GZIP,
		Strategy:    wrapify.STRATEGY_DIRECT,
	}

	_ = transformResponse.WithStreaming(transformReader, config)

	fmt.Println("  Transforming data...")

	// Since we can't access streaming methods directly, simulate transformation
	gzipWriter, _ := gzip.NewWriterLevel(transformedBuffer, gzip.DefaultCompression)
	io.Copy(gzipWriter, transformReader)
	gzipWriter.Close()

	fmt.Printf("  ✓ Transformer processed and compressed to %d bytes\n", transformedBuffer.Len())
	fmt.Printf("    Compression: %.1f%% reduction\n",
		(1-float64(transformedBuffer.Len())/float64(len(producerData)))*100)

	// Service 3: Consumer - Store transformed data
	fmt.Println("\n📍 Service 3: Data Consumer (Storage)")

	consumerBuffer := bytes.NewBuffer([]byte{})
	consumerReader := bytes.NewReader(transformedBuffer.Bytes())

	consumerResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/service3/data/store")

	consumerConfig := &wrapify.StreamConfig{
		ChunkSize:           256,
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 2,
		IsReceiving:         true,
	}

	_ = consumerResponse.WithStreaming(consumerReader, consumerConfig)

	// Decompress
	gzipReader, _ := gzip.NewReader(consumerReader)
	io.Copy(consumerBuffer, gzipReader)
	gzipReader.Close()

	fmt.Printf("  ✓ Consumer stored %d bytes\n", consumerBuffer.Len())

	// Final report
	fmt.Printf("\n📊 Pipeline Summary:\n")
	fmt.Printf("   Service 1 → Service 2: Original: %d bytes\n", len(producerData))
	fmt.Printf("   Service 2 → Service 3: Compressed: %d bytes\n", transformedBuffer.Len())
	fmt.Printf("   Final output size: %d bytes\n", consumerBuffer.Len())
	fmt.Printf("   Data integrity: %s\n", map[bool]string{true: "✅ PASSED", false: "❌ FAILED"}[len(consumerBuffer.Bytes()) > 0])
}

// ============================================================================
// USE CASE 15: Database Replication (Master -> Slave) - FIXED v2
// ============================================================================
func Example_DatabaseReplication_FIXED_v2() {

	// Master database: Generate replication stream
	fmt.Println("📍 MASTER: Generating replication stream")

	var txnLog strings.Builder
	txnLog.WriteString("BINLOG|version=5|position=0\n")
	for i := 0; i < 10000; i++ {
		txnLog.WriteString(fmt.Sprintf("TXN|%d|INSERT|users|id=%d,name=user_%d\n", i, i, i))
	}

	masterResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/replication/stream").
		WithMessage("Streaming database changes")

	masterReader := strings.NewReader(txnLog.String())
	masterBuffer := bytes.NewBuffer([]byte{})

	masterConfig := &wrapify.StreamConfig{
		ChunkSize:           512 * 1024,
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 2,
		UseBufferPool:       true,
	}

	_ = masterResponse.WithStreaming(masterReader, masterConfig)

	fmt.Println("  Sending replication stream...")

	// Compress transactions
	gzipWriter, _ := gzip.NewWriterLevel(masterBuffer, gzip.DefaultCompression)
	bytesWritten, _ := io.Copy(gzipWriter, masterReader)
	gzipWriter.Close()

	fmt.Printf("  ✓ Master replication stream sent\n")
	fmt.Printf("    Total transactions: 10,000\n")
	fmt.Printf("    Stream size: %.2f MB (compressed: %.2f MB)\n",
		float64(txnLog.Len())/1024/1024,
		float64(masterBuffer.Len())/1024/1024)
	fmt.Printf("    Bytes written: %d\n", bytesWritten)

	// Slave database: Receive and apply replication stream
	fmt.Println("\n📍 SLAVE: Receiving and applying replication stream")

	slaveReader := bytes.NewReader(masterBuffer.Bytes())
	slaveBuffer := bytes.NewBuffer([]byte{})

	slaveResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/replication/apply").
		WithMessage("Applying replication stream")

	slaveConfig := &wrapify.StreamConfig{
		ChunkSize:           512 * 1024,
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 2,
		UseBufferPool:       true,
		IsReceiving:         true,
	}

	_ = slaveResponse.WithStreaming(slaveReader, slaveConfig)

	fmt.Println("  Applying replication stream to slave...")

	// Decompress transactions
	gzipReader, _ := gzip.NewReader(slaveReader)
	bytesRead, _ := io.Copy(slaveBuffer, gzipReader)
	gzipReader.Close()

	fmt.Printf("  ✓ Slave replication applied\n")
	fmt.Printf("    Transactions applied: %d\n", slaveBuffer.Len()/100) // Rough estimate
	fmt.Printf("    Decompressed size: %.2f MB\n", float64(slaveBuffer.Len())/1024/1024)
	fmt.Printf("    Bytes read: %d\n", bytesRead)
}

// ============================================================================
// USE CASE 19: Complete E2E Scenario (Upload -> Process -> Download) - FIXED v2
// ============================================================================
func Example_CompleteE2EScenario_FIXED_v2() {
	fmt.Println("\n" + strings.Repeat("=", 80))
	fmt.Println("🚀 COMPLETE E2E SCENARIO: Upload → Process → Download (FIXED v2)")
	fmt.Println(strings.Repeat("=", 80))

	// STEP 1: Client uploads CSV file
	fmt.Println("\n📤 STEP 1: Client uploads CSV file")
	fmt.Println(strings.Repeat("-", 80))

	csvContent := "id,name,email,age,salary\n"
	for i := 1; i <= 1000; i++ {
		csvContent += fmt.Sprintf("%d,Employee_%d,emp_%d@company.com,%d,%d\n",
			i, i, i, 25+i%40, 30000+i*100)
	}

	uploadResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/api/upload/process").
		WithMessage("File upload initiated")

	uploadBuffer := bytes.NewBuffer([]byte{})
	uploadReader := strings.NewReader(csvContent)

	uploadConfig := &wrapify.StreamConfig{
		ChunkSize:           64 * 1024,
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 2,
	}

	_ = uploadResponse.WithStreaming(uploadReader, uploadConfig)

	fmt.Println("Uploading CSV file...")
	uploadStart := time.Now()

	// Manual compression
	gzipWriter, _ := gzip.NewWriterLevel(uploadBuffer, gzip.DefaultCompression)
	io.Copy(gzipWriter, uploadReader)
	gzipWriter.Close()

	uploadDuration := time.Since(uploadStart)
	uploadSize := len(csvContent)
	compressedSize := uploadBuffer.Len()

	fmt.Printf("✅ Upload completed\n")
	fmt.Printf("   Records: 1,000\n")
	fmt.Printf("   Original size: %.2f KB\n", float64(uploadSize)/1024)
	fmt.Printf("   Compressed size: %.2f KB\n", float64(compressedSize)/1024)
	fmt.Printf("   Time: %.2f seconds\n", uploadDuration.Seconds())
	fmt.Printf("   Speed: %.2f MB/s\n", float64(uploadSize)/uploadDuration.Seconds()/1024/1024)

	// STEP 2: Server processes the file (simulated)
	fmt.Println("\n⚙️  STEP 2: Server processes the file")
	fmt.Println(strings.Repeat("-", 80))

	processedBuffer := bytes.NewBuffer([]byte{})
	processReader := bytes.NewReader(uploadBuffer.Bytes())

	processResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/api/process/data").
		WithMessage("Processing file")

	processConfig := &wrapify.StreamConfig{
		ChunkSize:           64 * 1024,
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 2,
		IsReceiving:         true,
	}

	_ = processResponse.WithStreaming(processReader, processConfig)

	fmt.Println("Processing uploaded file...")
	processStart := time.Now()

	// Manual decompression
	gzipReader, _ := gzip.NewReader(processReader)
	io.Copy(processedBuffer, gzipReader)
	gzipReader.Close()

	processDuration := time.Since(processStart)

	fmt.Printf("✅ Processing completed\n")
	fmt.Printf("   Processed size: %.2f KB\n", float64(processedBuffer.Len())/1024)
	fmt.Printf("   Processing time: %.2f seconds\n", processDuration.Seconds())

	// STEP 3: Client downloads processed results
	fmt.Println("\n📥 STEP 3: Client downloads processed results")
	fmt.Println(strings.Repeat("-", 80))

	downloadReader := bytes.NewReader(processedBuffer.Bytes())

	downloadResponse := wrapify.New().
		WithStatusCode(200).
		WithPath("/api/download/results").
		WithMessage("Downloading processed data")

	downloadConfig := &wrapify.StreamConfig{
		ChunkSize:           128 * 1024,
		Strategy:            wrapify.STRATEGY_BUFFERED,
		Compression:         wrapify.COMP_GZIP,
		MaxConcurrentChunks: 2,
		ThrottleRate:        5 * 1024 * 1024,
	}

	_ = downloadResponse.WithStreaming(downloadReader, downloadConfig)

	fmt.Println("Downloading results...")
	downloadStart := time.Now()

	// Manual compression for download
	downloadBuffer := bytes.NewBuffer([]byte{})
	gzipWriter, _ = gzip.NewWriterLevel(downloadBuffer, gzip.DefaultCompression)
	io.Copy(gzipWriter, downloadReader)
	gzipWriter.Close()

	downloadDuration := time.Since(downloadStart)

	fmt.Printf("✅ Download completed\n")
	fmt.Printf("   Result size: %.2f KB (compressed: %.2f KB)\n",
		float64(processedBuffer.Len())/1024,
		float64(downloadBuffer.Len())/1024)
	fmt.Printf("   Time: %.2f seconds\n", downloadDuration.Seconds())
	fmt.Printf("   Speed: %.2f MB/s\n", float64(processedBuffer.Len())/downloadDuration.Seconds()/1024/1024)

	// Final Summary
	fmt.Println("\n" + strings.Repeat("=", 80))
	fmt.Println("📊 COMPLETE WORKFLOW SUMMARY")
	fmt.Println(strings.Repeat("=", 80))

	totalTime := uploadDuration + processDuration + downloadDuration

	fmt.Printf("\n Upload:     %.2f sec | %.2f MB/s | Original: %.2f KB → Compressed: %.2f KB\n",
		uploadDuration.Seconds(),
		float64(uploadSize)/uploadDuration.Seconds()/1024/1024,
		float64(uploadSize)/1024,
		float64(compressedSize)/1024)

	fmt.Printf(" Process:    %.2f sec | Decompressed: %.2f KB\n",
		processDuration.Seconds(),
		float64(processedBuffer.Len())/1024)

	fmt.Printf(" Download:   %.2f sec | Compressed: %.2f KB\n",
		downloadDuration.Seconds(),
		float64(downloadBuffer.Len())/1024)

	fmt.Printf("\n Total time: %.2f seconds\n", totalTime.Seconds())
	fmt.Printf(" Overall throughput: %.2f MB/s\n",
		float64(uploadSize+processedBuffer.Len())/totalTime.Seconds()/1024/1024)

	fmt.Println("\n" + strings.Repeat("=", 80))
	fmt.Println("✨ Workflow completed successfully!")
	fmt.Println(strings.Repeat("=", 80))
}

func Test01() {
	fmt.Println("Main running...")
	p := wrapify.Pages().
		WithIsLast(true).
		WithPage(1).
		WithTotalItems(120).
		WithPage(1000).
		WithTotalPages(34).
		WithPerPage(2)
	w := wrapify.New().
		WithStatusCode(200).
		WithTotal(1).
		WithMessagef("How are you? %v", "I'm good").
		WithDebuggingKVf("refer", "sessions_id: %v", randn.CryptoID()).
		WithBody(p.Respond()).
		WithPath("/api/v1/users").
		WithCustomFieldKVf("fields", "userID: %v", 103).
		WithPagination(p)
	if !w.Available() {
		return
	}
	clone := w.Clone()
	fmt.Printf("Clone: %+v\n", clone.JsonPretty())
	style := &unify4g.Style{
		Key:      [2]string{"\033[1;34m", "\033[0m"},
		String:   [2]string{"\033[1;32m", "\033[0m"},
		Number:   [2]string{"\033[1;33m", "\033[0m"},
		True:     [2]string{"\033[1;35m", "\033[0m"},
		False:    [2]string{"\033[1;35m", "\033[0m"},
		Null:     [2]string{"\033[1;35m", "\033[0m"},
		Escape:   [2]string{"\033[1;31m", "\033[0m"},
		Brackets: [2]string{"\033[1;37m", "\033[0m"},
		Append:   func(dst []byte, c byte) []byte { return append(dst, c) },
	}
	fmt.Printf("Hash Optimize: %s\n", w.OptimizeSafe(2).JsonPretty())
	fmt.Printf("Hash Caller 01: %s\n", w.Json())
	fmt.Printf("Hash Caller 02: %s\n", w.Json())
	fmt.Printf("Hash Caller 03: %s\n", w.Json())
	bytes := unify4g.Color([]byte(w.Json()), style)
	fmt.Println("Body Bytes: " + string(bytes))
	fmt.Println("Body JSON: " + encoding.JsonSafe(w.Body()))
	fmt.Println("Body Hash: " + w.Hash256Safe())
	fmt.Println(w.Meta().IsCustomFieldPresent())
	var err = TestErr01()
	fmt.Println(err.Error())
	err = TestErr002()
	fmt.Println(err.Error())
}

func TestErr01() error {
	v := wrapify.New().
		WithErrSck(fmt.Errorf("error wrapper 001"))
	return v
}

func TestErr002() error {
	v := wrapify.New().
		WithErrMessage(fmt.Errorf("error wrapper 002"), "test error").
		WithErrSck(fmt.Errorf("error wrapper 003")).
		WithErrSck(fmt.Errorf("error wrapper 004"))
	return v
}

func Test02() {
	// Demonstration of `WithError`
	err := wrapify.WithError("Something went wrong")
	fmt.Println("WithError:", err)

	// Demonstration of `WithErrorf`
	errf := wrapify.WithErrorf("Failed to load file %s", "config.json")
	fmt.Println("WithErrorf:", errf)

	// Demonstration of `WithErrStack`
	originalErr := errors.New("original error")
	errWithStack := wrapify.WithErrStack(originalErr)
	fmt.Println("WithErrStack:", errWithStack)

	// Demonstration of `Wrap`
	wrappedErr := wrapify.WithErrWrap(originalErr, "Failed to read the file")
	fmt.Println("Wrap:", wrappedErr)

	// Demonstration of `Wrapf`
	wrappedErrf := wrapify.WithErrWrapf(originalErr, "Failed to open file %s", "data.txt")
	fmt.Println("Wrapf:", wrappedErrf)

	// Demonstration of `WithMessage`
	errWithMessage := wrapify.WithMessage(originalErr, "Additional context")
	fmt.Println("WithMessage:", errWithMessage)

	// Demonstration of `WithMessagef`
	errWithMessagef := wrapify.WithMessagef(originalErr, "Context: %s", "something went wrong")
	fmt.Println("WithMessagef:", errWithMessagef)

	// Demonstration of `Cause`
	causeErr := wrapify.Cause(wrappedErrf)
	fmt.Println("Cause:", causeErr)
}

func Test03() {
	jsonStr := `{
		"data": "response body here",
		"debug": {
		  "___abc": "trace sessions_id: 4919e84fc26881e9fe790f5d07465db4",
		  "refer": 1234
		},
		"message": "How do you do? I'm good",
		"meta": {
		  "api_version": "v0.0.1",
		  "custom_fields": {
			"fields": "userID: 103"
		  },
		  "locale": "en_US",
		  "request_id": "80eafc6a1655ec5a06595d155f1e6951",
		  "requested_time": "2024-12-14T20:24:23.983839+07:00"
		},
		"pagination": {
		  "is_last": true,
		  "page": 1000,
		  "per_page": 2,
		  "total_items": 120,
		  "total_pages": 34
		},
		"path": "/api/v1/users",
		"status_code": 200,
		"total": 1
	  }`
	t := time.Now()
	w, err := wrapify.UnwrapJSON(jsonStr)
	diff := time.Since(t)
	if err != nil {
		log.Fatalf("Error parsing JSON: %v", err)
	}
	fmt.Printf("Exe time: %+v\n", diff.String())
	fmt.Printf("%+v\n", w.OnKeyDebugging("___abc"))
	fmt.Printf("%+v\n", w.JsonPretty())
}

func Test04() {
	raw := map[string]interface{}{
		"status_code": 200,
		"total":       1,
		"message":     "Success",
		"data":        "response body",
		"path":        "/api/v1/users",
		"debug": map[string]interface{}{
			"trace_id": "4919e84fc26881e9fe790f5d07465db4",
			"refer":    1,
		},
		"pagination": map[string]interface{}{
			"is_last":     true,
			"page":        1000,
			"per_page":    2,
			"total_items": 120,
			"total_pages": 34,
		},
	}
	t := time.Now()
	wrapper, err := wrapify.WrapFrom(raw)
	diff := time.Since(t)
	fmt.Printf("Executed in: %+v\n", diff.String())
	if err != nil {
		log.Println("Error extracting wrapper:", err)
	} else {
		log.Println("Functions:", wrapper.OnKeyDebugging("trace_id"))
		log.Println("Response:", wrapper.JsonPretty())
	}
}

func Test05() {
	response := wrapify.WrapOk("Created user successfully", wrapify.New().Respond())
	fmt.Println(response.JsonPretty())

	jsonStr := `
	{
		"data": {
			"meta": {
				"api_version": "v0.0.1",
				"locale": "en_US",
				"request_id": "bd1aa9fdc75f4d1976930537e3d9f977",
				"requested_time": "2024-12-15T09:59:15.956855+07:00"
			},
			"total": 0
		},
		"message": "Created user successfully",
		"meta": {
			"api_version": "v0.0.1",
			"locale": "en_US",
			"request_id": "9df5939c2ce462fc6b36e513683449eb",
			"requested_time": "2024-12-15T09:59:15.956923+07:00"
		},
		"status_code": 200,
		"total": 0
	}
	`
	t := time.Now()
	w, err := wrapify.UnwrapJSON(jsonStr)
	diff := time.Since(t)
	if err != nil {
		log.Fatalf("Error parsing JSON: %v", err)
	}
	fmt.Printf("Executed in: %+v\n", diff.String())
	fmt.Printf("%+v\n", w.JsonPretty())
	fmt.Printf("%+v\n", w.Meta().Locale())
}

func Test06() {
	// Create a new wrapper and add some response data
	responseWrapper := wrapify.New().
		WithStatusCode(200).
		WithMessage("Success").
		WithBody(map[string]any{
			"users": []string{"Alice", "Bob", "Charlie"},
			"count": 3,
		})

	// Call the Stream function to get a channel of byte chunks
	chunks := responseWrapper.Stream()

	// Iterate over each received chunk and process it
	for chunk := range chunks {
		fmt.Printf("Received chunk (%d bytes): %s\n", len(chunk), string(chunk))
	}
}