bugblitz77/jobs-monorepo
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

retryable http

Browse source
This commit is contained in:
Elshimy Ziad Magdy Taha 2025-09-28 21:31:11 +05:00
parent a0e1728276
commit c470edd187
5 changed files with 42 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.env
View file

@ -8,9 +8,5 @@ DB_SSLMODE=disable
SERVER_PORT=8080
SERVER_HOST=localhost
# LinkedIn parsing settings
LINKEDIN_BASE_URL=https://www.linkedin.com
REQUEST_TIMEOUT=30s
RATE_LIMIT_DELAY=2s
CV_AI_MODEL=your_ai_model
OPENROUTER_API_KEY=your_openrouter_api_key

5
internal/pipeline/test.go → internal/pipeline/job_pipeline-workers.go
View file

@ -25,13 +25,12 @@ func GetJobs(context context.Context, scraperService *Scraper) <-chan models.Job
return jobChan
}
func GetJobDescription(context context.Context, scraperService *Scraper, jobChan <-chan models.Job) <-chan models.JobWithDescription {
func GetJobDescription(context context.Context, scraperService *Scraper, jobChan <-chan models.Job, numWorkers int) <-chan models.JobWithDescription {
jobDescriptionChan := make(chan models.JobWithDescription, 100)
const numWorkers = 3
var wg sync.WaitGroup
// Start 3 worker goroutines
// Start worker goroutines
for range numWorkers {
wg.Add(1)
go func() {

43
internal/pipeline/job_pipeline.go
View file

@ -11,7 +11,6 @@ import (
"github.com/jobs-scraper/internal/models"
"github.com/jobs-scraper/internal/repo"
"golang.org/x/time/rate"
)
// JobDescriptionResult represents the result of job description scraping
@ -43,10 +42,9 @@ func (p *JobPipeline) ProcessJobsStreaming(ctx context.Context, numPages int, jo
// Create channels for the pipeline
allJobs := make([]models.Job, 0, 100)
allJobDescriptions := make([]models.JobDescription, 0, 100)
// var mu sync.Mutex
var jbMu sync.Mutex
jobsChan := GetJobs(ctx, p.scraperService)
jobWithDescriptionChan := GetJobDescription(ctx, p.scraperService, jobsChan)
jobWithDescriptionChan := GetJobDescription(ctx, p.scraperService, jobsChan, 3)
for jobWithDescription := range jobWithDescriptionChan {
fmt.Printf("Received job description for job : %d\n", jobWithDescription.Job.ID)
@ -66,42 +64,3 @@ func (p *JobPipeline) ProcessJobsStreaming(ctx context.Context, numPages int, jo
return nil
}
// jobDescriptionWorker processes jobs from jobChan and sends results to jobDescriptionChan
func (p *JobPipeline) jobDescriptionWorker(ctx context.Context, jobChan <-chan models.Job, resultChan chan<- JobDescriptionResult) {
limiter := rate.NewLimiter(rate.Every(2*time.Second), 1) // 1 request per second
for {
select {
case <-ctx.Done():
return
case job, ok := <-jobChan:
if !ok {
return
}
if err := limiter.Wait(ctx); err != nil {
return
}
fmt.Printf("Processing job: %s\n", job.Title)
description, criteria, err := p.scraperService.ScrapeJobDescriptionWithContext(ctx, job)
result := JobDescriptionResult{
Job: job,
Description: description,
Criteria: criteria,
Error: err,
}
select {
case resultChan <- result:
case <-ctx.Done():
return
}
}
}
}

16
internal/pipeline/scraper.go
View file

@ -88,7 +88,13 @@ func (s *Scraper) ScrapeJobsWithContext(ctx context.Context, page int, params mo
jobs := make([]models.Job, 0, 10)
url := s.buildSearchURL(params, page)
res, err := utils.RetryableHTTPRequest(ctx, url)
retryableRequest := utils.NewRetryableHTTPRequest(utils.RetryConfig{
BaseDelay: s.config.BaseDelay,
MaxDelay: s.config.MaxDelay,
MaxRetries: s.config.MaxRetries,
})
res, err := retryableRequest.RetryableHTTPRequest(ctx, url, "GET", nil, nil)
if err != nil {
fmt.Printf("Error fetching URL after retries: %v\n", err)
return jobs, err
@ -132,7 +138,13 @@ func (s *Scraper) ScrapeJobDescriptionWithContext(ctx context.Context, job model
var jobDescription string
url := s.buildJobDescriptionSearchURL(job.JobLink)
res, err := utils.RetryableHTTPRequest(ctx, url)
retryableRequest := utils.NewRetryableHTTPRequest(utils.RetryConfig{
MaxRetries: s.config.MaxRetries,
BaseDelay: s.config.BaseDelay,
MaxDelay: s.config.MaxDelay,
})
res, err := retryableRequest.RetryableHTTPRequest(ctx, url, "GET", nil, nil)
if err != nil {
fmt.Printf("Error fetching job description URL after retries: %v\n", err)
return "", map[string]string{}, err

34
internal/utils/retryable-http-request.go
View file

@ -3,7 +3,7 @@ package utils
import (
"context"
"fmt"
"math"
"io"
"net/http"
"time"
)
@ -14,23 +14,39 @@ type RetryConfig struct {
MaxDelay time.Duration
}
func RetryableHTTPRequest(ctx context.Context, url string) (*http.Response, error) {
type RetryableHTTPRequestImpl struct {
client *http.Client
config RetryConfig
}
func NewRetryableHTTPRequest(config RetryConfig) *RetryableHTTPRequestImpl {
client := &http.Client{
Timeout: 30 * time.Second,
}
return &RetryableHTTPRequestImpl{
client: client,
config: config,
}
}
func (s *RetryableHTTPRequestImpl) RetryableHTTPRequest(ctx context.Context, url, method string, body io.Reader, headers []http.Header) (*http.Response, error) {
var lastErr error
for attempt := 0; attempt <= 3; attempt++ {
// Create request with context
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
for attempt := 0; attempt <= s.config.MaxRetries; attempt++ {
req, err := http.NewRequestWithContext(ctx, method, url, body)
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36")
for _, header := range headers {
req.Header.Set(header.Get("Key"), header.Get("Value"))
}
resp, err := client.Do(req)
resp, err := s.client.Do(req)
if err != nil {
lastErr = err
fmt.Printf("Request attempt %d failed: %v\n", attempt+1, err)
@ -49,10 +65,10 @@ func RetryableHTTPRequest(ctx context.Context, url string) (*http.Response, erro
}
// max attempts for now = 3
if attempt < 3 {
delay := min(time.Duration(math.Pow(2, float64(attempt)))*1*time.Second, 30*time.Second)
if attempt < s.config.MaxRetries {
delay := time.Second * 2
fmt.Printf("Retrying in %v... (attempt %d/%d)\n", delay, attempt+1, 3)
fmt.Printf("Retrying in %v... (attempt %d/%d)\n", delay, attempt+1, s.config.MaxRetries)
select {
case <-ctx.Done():