Go (Golang) ile Yüksek Eş Zamanlı SMS Doğrulama Worker Havuzu: Goroutine, Channel ve Mutex Yönetimi

+-----------------------------------------------------------------------------------+
|               GOLANG CONCURRENCY & BOUNDED WORKER POOL MİMARİSİ                   |
+-----------------------------------------------------------------------------------+
|                                                                                   |
|   [ HTTP / gRPC / Webhook İstekleri ] ---> Saniyede 20.000+ Doğrulama Talebi      |
|                               |                                                   |
|                               v                                                   |
|   [ Request Dispatcher ] ---> Token Bucket Rate Limiter (golang.org/x/time/rate)  |
|                               |                                                   |
|                               v                                                   |
|   +---------------------------------------------------------------------------+   |
|   | Buffered Job Channel (make(chan SMSJob, 50000))                           |   |
|   +---------------------------+-----------------------------------------------+   |
|                               |                                                   |
|           +-------------------+-------------------+                               |
|           |                   |                   |                               |
|           v                   v                   v                               |
|   +---------------+   +---------------+   +---------------+                       |
|   | Worker 1 (Go) |   | Worker 2 (Go) |   | Worker N (Go) | (Max: 50 Goroutines)  |
|   +-------+-------+   +-------+-------+   +-------+-------+                       |
|           |                   |                   |                               |
|           +-------------------+-------------------+                               |
|                               |                                                   |
|                               v  (HTTP Keep-Alive Connection Pool, MaxIdle=100)   |
|            [ EnOnay Global REST API Gateway v1.4 (HTTPS) ]                        |
|                               |                                                   |
|                               v                                                   |
|   +---------------------------------------------------------------------------+   |
|   | Result Channel & sync.WaitGroup / Prometheus Metrics                      |   |
|   +---------------------------------------------------------------------------+   |
+-----------------------------------------------------------------------------------+

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"golang.org/x/time/rate"
)

type SMSJob struct {
	ID          string `json:"job_id"`
	PhoneNumber string `json:"phone_number"`
	Service     string `json:"service"`
	Retries     int    `json:"retries"`
}

type SMSResult struct {
	JobID     string
	Success   bool
	SessionID string
	Error     error
	Duration  time.Duration
}

type WorkerPool struct {
	maxWorkers int
	jobQueue   chan SMSJob
	results    chan SMSResult
	wg         sync.WaitGroup
	httpClient *http.Client
	limiter    *rate.Limiter
	apiKey     string
	apiURL     string
}

func NewWorkerPool(workers int, queueSize int, apiKey string, rps int) *WorkerPool {
	// Yüksek performanslı HTTP Keep-Alive istemcisi
	tr := &http.Transport{
		MaxIdleConns:        500,
		MaxIdleConnsPerHost: 100,
		IdleConnTimeout:     90 * time.Second,
	}

	return &WorkerPool{
		maxWorkers: workers,
		jobQueue:   make(chan SMSJob, queueSize),
		results:    make(chan SMSResult, queueSize),
		httpClient: &http.Client{Transport: tr, Timeout: 4 * time.Second},
		limiter:    rate.NewLimiter(rate.Limit(rps), rps*2),
		apiKey:     apiKey,
		apiURL:     "https://enonay.com/api/v1.php",
	}
}

func (p *WorkerPool) Start(ctx context.Context) {
	for i := 1; i <= p.maxWorkers; i++ {
		p.wg.Add(1)
		go p.workerRoutine(ctx, i)
	}
}

func (p *WorkerPool) workerRoutine(ctx context.Context, workerID int) {
	defer p.wg.Done()
	log.Printf("[Worker #%d] Başlatıldı ve kuyruk dinleniyor...", workerID)

	for {
		select {
		case <-ctx.Done():
			log.Printf("[Worker #%d] İptal sinyali alındı, kapanıyor.", workerID)
			return
		case job, ok := <-p.jobQueue:
			if !ok {
				return // Kanal kapandı
			}

			// Token Bucket Hız Sınırlaması Beklemesi
			if err := p.limiter.Wait(ctx); err != nil {
				return
			}

			start := time.Now()
			res, err := p.executeEnOnayRequest(ctx, job)
			duration := time.Since(start)

			p.results <- SMSResult{
				JobID:     job.ID,
				Success:   err == nil,
				SessionID: res,
				Error:     err,
				Duration:  duration,
			}
		}
	}
}

func (p *WorkerPool) executeEnOnayRequest(ctx context.Context, job SMSJob) (string, error) {
	payload := map[string]interface{}{
		"action":  "order",
		"service": job.Service,
		"phone":   job.PhoneNumber,
	}
	bodyBytes, _ := json.Marshal(payload)

	req, err := http.NewRequestWithContext(ctx, "POST", p.apiURL, bytes.NewBuffer(bodyBytes))
	if err != nil {
		return "", err
	}
	req.Header.Set("Authorization", "Bearer "+p.apiKey)
	req.Header.Set("Content-Type", "application/json")

	resp, err := p.httpClient.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("api status code error: %d", resp.StatusCode)
	}

	return fmt.Sprintf("session-%s-verified", job.ID), nil
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 50 Worker, 10.000 Kuyruk Derinliği, Saniyede 500 RPS Token Bucket Limiti
	pool := NewWorkerPool(50, 10000, "ENONAY_GO_SECRET_KEY", 500)
	pool.Start(ctx)

	// Örnek İş Gönderimi
	go func() {
		for i := 1; i <= 500; i++ {
			pool.jobQueue <- SMSJob{
				ID:          fmt.Sprintf("REQ-%d", i),
				PhoneNumber: "+905320000000",
				Service:     "whatsapp",
			}
		}
	}()

	// Graceful Shutdown Yakalama
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	<-sigChan
	log.Println("Kapatma sinyali alındı! Kuyruk boşaltılıyor...")
	cancel()
	close(pool.jobQueue)
	pool.wg.Wait()
	log.Println("Tüm Go worker rutinleri güvenle sonlandırıldı.")
}