Concurrent Video Processing with Go Goroutines for European Content

package pipeline

import (
    "context"
    "fmt"
    "net/http"
    "sync"
    "time"
)

type RawVideo struct {
    ID       string
    Title    string
    Channel  string
    Views    int64
    Region   string
    ThumbURL string
}

type ProcessedVideo struct {
    RawVideo
    Language      string
    ThumbValid    bool
    EngagementPct float64
    ProcessedAt   time.Time
}

type ProcessResult struct {
    Video ProcessedVideo
    Err   error
}

func ProcessBatch(ctx context.Context, videos []RawVideo, workers int) []ProcessResult {
    jobs := make(chan RawVideo, len(videos))
    results := make(chan ProcessResult, len(videos))

    var wg sync.WaitGroup
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            client := &http.Client{Timeout: 3 * time.Second}
            for raw := range jobs {
                select {
                case <-ctx.Done():
                    return
                default:
                    processed, err := processVideo(client, raw)
                    results <- ProcessResult{Video: processed, Err: err}
                }
            }
        }(w)
    }

    for _, v := range videos {
        jobs <- v
    }
    close(jobs)

    go func() {
        wg.Wait()
        close(results)
    }()

    var out []ProcessResult
    for r := range results {
        out = append(out, r)
    }
    return out
}

var regionLanguages = map[string]string{
    "PL": "pl", "NL": "nl", "SE": "sv",
    "NO": "no", "AT": "de", "GB": "en", "US": "en",
}

func processVideo(client *http.Client, raw RawVideo) (ProcessedVideo, error) {
    p := ProcessedVideo{
        RawVideo:    raw,
        Language:    regionLanguages[raw.Region],
        ProcessedAt: time.Now(),
    }

    // Validate thumbnail with HEAD request
    if raw.ThumbURL != "" {
        resp, err := client.Head(raw.ThumbURL)
        if err == nil {
            p.ThumbValid = resp.StatusCode == 200
            resp.Body.Close()
        }
    }
    if !p.ThumbValid {
        p.ThumbURL = fmt.Sprintf("https://i.ytimg.com/vi/%s/mqdefault.jpg", raw.ID)
        p.ThumbValid = true
    }

    return p, nil
}

package fetcher

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "sync"
    "time"
)

type RegionResult struct {
    Region string
    Videos []RawVideo
    Err    error
}

type EuropeFetcher struct {
    apiKey string
    client *http.Client
    mu     sync.Mutex
    quota  int
}

func NewEuropeFetcher(apiKey string) *EuropeFetcher {
    return &EuropeFetcher{
        apiKey: apiKey,
        client: &http.Client{Timeout: 10 * time.Second},
    }
}

func (f *EuropeFetcher) FetchAllRegions(ctx context.Context) []RegionResult {
    regions := []string{"US", "GB", "PL", "NL", "SE", "NO", "AT"}
    results := make(chan RegionResult, len(regions))

    // Rate limiter: 2 requests per second
    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()

    var wg sync.WaitGroup
    for _, region := range regions {
        <-ticker.C
        wg.Add(1)
        go func(r string) {
            defer wg.Done()
            videos, err := f.fetchOne(ctx, r)
            results <- RegionResult{Region: r, Videos: videos, Err: err}
        }(region)
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    var out []RegionResult
    for r := range results {
        if r.Err != nil {
            fmt.Printf("[%s] error: %v\n", r.Region, r.Err)
        } else {
            fmt.Printf("[%s] fetched %d videos\n", r.Region, len(r.Videos))
        }
        out = append(out, r)
        f.mu.Lock()
        f.quota++
        f.mu.Unlock()
    }
    return out
}

func (f *EuropeFetcher) fetchOne(ctx context.Context, region string) ([]RawVideo, error) {
    url := fmt.Sprintf(
        "https://www.googleapis.com/youtube/v3/videos?part=snippet,statistics&chart=mostPopular&regionCode=%s&maxResults=25&key=%s",
        region, f.apiKey,
    )

    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        return nil, err
    }

    resp, err := f.client.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    if resp.StatusCode != 200 {
        return nil, fmt.Errorf("status %d for %s", resp.StatusCode, region)
    }

    var data struct {
        Items []struct {
            ID      string `json:"id"`
            Snippet struct {
                Title   string `json:"title"`
                Channel string `json:"channelTitle"`
            } `json:"snippet"`
            Stats struct {
                Views string `json:"viewCount"`
            } `json:"statistics"`
        } `json:"items"`
    }

    if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
        return nil, err
    }

    var videos []RawVideo
    for _, item := range data.Items {
        videos = append(videos, RawVideo{
            ID:      item.ID,
            Title:   item.Snippet.Title,
            Channel: item.Snippet.Channel,
            Region:  region,
        })
    }
    return videos, nil
}

func RunPipeline(ctx context.Context, apiKey string) error {
    fetcher := NewEuropeFetcher(apiKey)

    // Stage 1: Fetch all regions
    regionResults := fetcher.FetchAllRegions(ctx)

    // Stage 2: Flatten into processing batch
    var allVideos []RawVideo
    for _, rr := range regionResults {
        if rr.Err == nil {
            allVideos = append(allVideos, rr.Videos...)
        }
    }
    fmt.Printf("Total raw videos: %d\n", len(allVideos))

    // Stage 3: Process with worker pool
    processed := ProcessBatch(ctx, allVideos, 8)

    valid := 0
    for _, p := range processed {
        if p.Err == nil && p.Video.ThumbValid {
            valid++
        }
    }
    fmt.Printf("Processed: %d total, %d valid\n", len(processed), valid)

    return nil
}