From 890469fed8dc1c51e7226ae71832cfe2677886d9 Mon Sep 17 00:00:00 2001 From: haopengzhan Date: Wed, 12 Aug 2026 23:28:31 -0700 Subject: [PATCH] refactor: implement worker pool pattern, add persistent task queue, and decouple target loading logic --- cmd/hireme/main.go | 21 +-- cmd/hireme/main_test.go | 109 ------------ internal/analyzer/deepseek.go | 2 +- internal/analyzer/gemini.go | 6 +- internal/crawler/generic.go | 2 +- internal/db/db.go | 13 +- internal/db/models.go | 13 +- internal/db/queries.go | 82 ++++----- internal/db/queries_test.go | 6 +- internal/db/targets_loader.go | 48 ++++++ internal/db/tasks.go | 159 +++++++++++++++++ internal/filter/filter.go | 11 +- internal/filter/filter_test.go | 4 +- internal/service/pipeline.go | 226 ------------------------ internal/ui/handlers.go | 77 +++++++-- internal/ui/templates.go | 44 +++-- internal/worker/pool.go | 306 +++++++++++++++++++++++++++++++++ pkg/crawler/google/google.go | 7 +- 18 files changed, 694 insertions(+), 442 deletions(-) delete mode 100644 cmd/hireme/main_test.go create mode 100644 internal/db/targets_loader.go create mode 100644 internal/db/tasks.go delete mode 100644 internal/service/pipeline.go create mode 100644 internal/worker/pool.go diff --git a/cmd/hireme/main.go b/cmd/hireme/main.go index e548596..686b99f 100644 --- a/cmd/hireme/main.go +++ b/cmd/hireme/main.go @@ -10,8 +10,8 @@ import ( "github.com/hireme/hireme/internal/analyzer" "github.com/hireme/hireme/internal/config" "github.com/hireme/hireme/internal/db" - "github.com/hireme/hireme/internal/service" "github.com/hireme/hireme/internal/ui" + "github.com/hireme/hireme/internal/worker" "github.com/spf13/cobra" ) @@ -42,7 +42,12 @@ func newRootCmd() *cobra.Command { if err != nil { return fmt.Errorf("fatal DB error: %w", err) } - defer database.Close() + defer func() { _ = database.Close() }() + + // Load targets from YAML config if database is empty + if err := database.LoadTargetsFromYAML(cfg.TargetConfig); err != nil { + log.Printf("[Main] Target YAML loading status: %v", err) + } // Initialize LLM Analyzer llmAnalyzer, err := analyzer.NewAnalyzer(cfg) @@ -50,16 +55,12 @@ func newRootCmd() *cobra.Command { return fmt.Errorf("fatal Analyzer setup error: %w", err) } - // Initialize Pipeline Service - pipeline := service.NewPipelineService(database, llmAnalyzer) - - // Load targets from YAML config if database is empty - if err := pipeline.LoadTargetsFromYAML(cfg.TargetConfig); err != nil { - log.Printf("[Main] Target YAML loading status: %v", err) - } + // Initialize Worker Pool + pool := worker.NewWorkerPool(database, llmAnalyzer, 3) + pool.Start() // Auto start the worker pool // Setup UI Server & Routes - server := ui.NewServer(cfg, database, pipeline) + server := ui.NewServer(cfg, database, pool) mux := http.NewServeMux() server.RegisterRoutes(mux) diff --git a/cmd/hireme/main_test.go b/cmd/hireme/main_test.go deleted file mode 100644 index 82eaa6e..0000000 --- a/cmd/hireme/main_test.go +++ /dev/null @@ -1,109 +0,0 @@ -package main - -import ( - "net/http" - "net/http/httptest" - "os" - "path/filepath" - "strings" - "testing" - - "github.com/hireme/hireme/internal/analyzer" - "github.com/hireme/hireme/internal/config" - "github.com/hireme/hireme/internal/db" - "github.com/hireme/hireme/internal/service" - "github.com/hireme/hireme/internal/ui" -) - -func TestFullApplicationIntegration(t *testing.T) { - tempDir := t.TempDir() - dbPath := filepath.Join(tempDir, "test_hireme.db") - yamlPath := filepath.Join(tempDir, "test_targets.yaml") - - yamlContent := `targets: - - name: "Google Software Engineer (Seattle)" - target_url: "https://www.google.com/about/careers/applications/jobs/results/?employment_type=FULL_TIME&q=%22Software%20Engineer%22&location=Seattle%2C%20WA%2C%20USA&target_level=MID" - platform: "google_jobs" - is_active: true -` - if err := os.WriteFile(yamlPath, []byte(yamlContent), 0644); err != nil { - t.Fatalf("failed to create test targets.yaml: %v", err) - } - - cfg := &config.Config{ - Port: "8080", - DBPath: dbPath, - TargetConfig: yamlPath, - LLMProvider: "mock", - } - - database, err := db.InitDB(cfg.DBPath) - if err != nil { - t.Fatalf("InitDB failed: %v", err) - } - defer database.Close() - - llmAnalyzer, err := analyzer.NewAnalyzer(cfg) - if err != nil { - t.Fatalf("NewAnalyzer failed: %v", err) - } - - pipeline := service.NewPipelineService(database, llmAnalyzer) - - // Verify target loading from YAML into SQLite - if err := pipeline.LoadTargetsFromYAML(cfg.TargetConfig); err != nil { - t.Fatalf("LoadTargetsFromYAML failed: %v", err) - } - - targets, err := database.GetAllTargets() - if err != nil { - t.Fatalf("GetAllTargets failed: %v", err) - } - if len(targets) != 1 { - t.Errorf("expected 1 target from YAML, got %d", len(targets)) - } - - // Verify jobs start empty on clean SQLite DB startup (0 jobs) - jobs, err := database.GetJobsWithAnalysis() - if err != nil { - t.Fatalf("GetJobsWithAnalysis failed: %v", err) - } - if len(jobs) != 0 { - t.Errorf("expected 0 jobs on clean startup, got %d", len(jobs)) - } - - // Setup UI Server & routes - server := ui.NewServer(cfg, database, pipeline) - mux := http.NewServeMux() - server.RegisterRoutes(mux) - - // Test GET / - reqRoot := httptest.NewRequest("GET", "/", nil) - wRoot := httptest.NewRecorder() - mux.ServeHTTP(wRoot, reqRoot) - - if wRoot.Code != http.StatusOK { - t.Errorf("expected 200 OK for GET /, got %d", wRoot.Code) - } - if !strings.Contains(wRoot.Body.String(), "Hireme") { - t.Errorf("expected homepage HTML to contain 'Hireme'") - } - - // Test GET /targets - reqTargets := httptest.NewRequest("GET", "/targets", nil) - wTargets := httptest.NewRecorder() - mux.ServeHTTP(wTargets, reqTargets) - - if wTargets.Code != http.StatusOK { - t.Errorf("expected 200 OK for GET /targets, got %d", wTargets.Code) - } - - // Test GET /jobs - reqJobs := httptest.NewRequest("GET", "/jobs", nil) - wJobs := httptest.NewRecorder() - mux.ServeHTTP(wJobs, reqJobs) - - if wJobs.Code != http.StatusOK { - t.Errorf("expected 200 OK for GET /jobs, got %d", wJobs.Code) - } -} diff --git a/internal/analyzer/deepseek.go b/internal/analyzer/deepseek.go index f197e69..bfd64eb 100644 --- a/internal/analyzer/deepseek.go +++ b/internal/analyzer/deepseek.go @@ -68,7 +68,7 @@ Job Content: if err != nil { return nil, fmt.Errorf("DeepSeek API call failed: %w", err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() respBytes, err := io.ReadAll(resp.Body) if err != nil { diff --git a/internal/analyzer/gemini.go b/internal/analyzer/gemini.go index 2efb265..7a145d9 100644 --- a/internal/analyzer/gemini.go +++ b/internal/analyzer/gemini.go @@ -63,9 +63,9 @@ Respond STRICTLY with valid JSON matching the format: resp, err := g.client.Do(req) if err != nil { - return nil, fmt.Errorf("Gemini API call failed: %w", err) + return nil, fmt.Errorf("gemini api call failed: %w", err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() respBytes, err := io.ReadAll(resp.Body) if err != nil { @@ -73,7 +73,7 @@ Respond STRICTLY with valid JSON matching the format: } if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("Gemini API error (status %d): %s", resp.StatusCode, string(respBytes)) + return nil, fmt.Errorf("gemini api error (status %d): %s", resp.StatusCode, string(respBytes)) } var geminiResp struct { diff --git a/internal/crawler/generic.go b/internal/crawler/generic.go index 5a3f78e..6e9f6aa 100644 --- a/internal/crawler/generic.go +++ b/internal/crawler/generic.go @@ -50,7 +50,7 @@ func (c *GenericCrawler) FetchJobDetail(ctx context.Context, applyURL string) (* if err != nil { return nil, fmt.Errorf("HTTP request failed: %w", err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("HTTP status error %d for URL %s", resp.StatusCode, applyURL) diff --git a/internal/db/db.go b/internal/db/db.go index 2be1128..869fef4 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -55,7 +55,6 @@ func (db *DB) migrate() error { location TEXT, apply_url TEXT NOT NULL, raw_content TEXT NOT NULL, - status TEXT NOT NULL DEFAULT 'PENDING', crawled_at DATETIME DEFAULT CURRENT_TIMESTAMP ); @@ -79,6 +78,18 @@ func (db *DB) migrate() error { applied_at DATETIME, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ); + + CREATE TABLE IF NOT EXISTS tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_type TEXT NOT NULL, + payload TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'PENDING', + error_message TEXT, + retry_count INTEGER DEFAULT 0, + next_retry_at DATETIME DEFAULT CURRENT_TIMESTAMP, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP + ); ` _, err := db.Exec(schema) diff --git a/internal/db/models.go b/internal/db/models.go index ec67286..26c3dbc 100644 --- a/internal/db/models.go +++ b/internal/db/models.go @@ -22,10 +22,21 @@ type RawJob struct { Location string `json:"location"` ApplyURL string `json:"apply_url"` RawContent string `json:"raw_content"` - Status string `json:"status"` // 'PENDING', 'PROCESSED', 'FAILED', 'DUPLICATE' CrawledAt time.Time `json:"crawled_at"` } +type Task struct { + ID int64 `json:"id"` + TaskType string `json:"task_type"` // 'DISCOVER', 'FETCH', 'PROCESS', 'ANALYZE' + Payload string `json:"payload"` // JSON string + Status string `json:"status"` // 'PENDING', 'RUNNING', 'COMPLETED', 'FAILED' + ErrorMessage string `json:"error_message"` + RetryCount int `json:"retry_count"` + NextRetryAt time.Time `json:"next_retry_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + type JobAnalysis struct { ID int64 `json:"id"` RawJobID int64 `json:"raw_job_id"` diff --git a/internal/db/queries.go b/internal/db/queries.go index 4220c60..55fe354 100644 --- a/internal/db/queries.go +++ b/internal/db/queries.go @@ -31,7 +31,7 @@ func (db *DB) GetActiveTargets() ([]SearchTarget, error) { if err != nil { return nil, err } - defer rows.Close() + defer func() { _ = rows.Close() }() var targets []SearchTarget for rows.Next() { @@ -54,7 +54,7 @@ func (db *DB) GetAllTargets() ([]SearchTarget, error) { if err != nil { return nil, err } - defer rows.Close() + defer func() { _ = rows.Close() }() var targets []SearchTarget for rows.Next() { @@ -104,15 +104,15 @@ func (db *DB) InsertRawJob(canonicalID string, targetID *int64, title, company, } query := ` - INSERT INTO raw_jobs (canonical_id, target_id, title, company, location, apply_url, raw_content, status) - VALUES (?, ?, ?, ?, ?, ?, ?, 'PENDING') + INSERT INTO raw_jobs (canonical_id, target_id, title, company, location, apply_url, raw_content) + VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(canonical_id) DO NOTHING - RETURNING id, canonical_id, target_id, title, company, location, apply_url, raw_content, status, crawled_at + RETURNING id, canonical_id, target_id, title, company, location, apply_url, raw_content, crawled_at ` var job RawJob var tID sql.NullInt64 err = db.QueryRow(query, canonicalID, targetID, title, company, location, applyURL, rawContent).Scan( - &job.ID, &job.CanonicalID, &tID, &job.Title, &job.Company, &job.Location, &job.ApplyURL, &job.RawContent, &job.Status, &job.CrawledAt, + &job.ID, &job.CanonicalID, &tID, &job.Title, &job.Company, &job.Location, &job.ApplyURL, &job.RawContent, &job.CrawledAt, ) if err == sql.ErrNoRows { return nil, false, nil // Duplicate job @@ -139,7 +139,7 @@ func (db *DB) getAllRawJobsForDeduplication() ([]RawJob, error) { if err != nil { return nil, err } - defer rows.Close() + defer func() { _ = rows.Close() }() var jobs []RawJob for rows.Next() { @@ -207,34 +207,7 @@ func tokenizeToVector(text string) map[string]int { return freqMap } -func (db *DB) GetPendingRawJobs() ([]RawJob, error) { - query := `SELECT id, canonical_id, target_id, title, company, location, apply_url, raw_content, status, crawled_at FROM raw_jobs WHERE status = 'PENDING'` - rows, err := db.Query(query) - if err != nil { - return nil, err - } - defer rows.Close() - var jobs []RawJob - for rows.Next() { - var j RawJob - var tID sql.NullInt64 - if err := rows.Scan(&j.ID, &j.CanonicalID, &tID, &j.Title, &j.Company, &j.Location, &j.ApplyURL, &j.RawContent, &j.Status, &j.CrawledAt); err != nil { - return nil, err - } - if tID.Valid { - j.TargetID = &tID.Int64 - } - jobs = append(jobs, j) - } - return jobs, nil -} - -func (db *DB) UpdateRawJobStatus(id int64, status string) error { - query := `UPDATE raw_jobs SET status = ? WHERE id = ?` - _, err := db.Exec(query, status, id) - return err -} // Job Analyses Queries func (db *DB) CreateJobAnalysis(rawJobID int64, score float64, isQualified bool, status string) error { @@ -247,6 +220,25 @@ func (db *DB) CreateJobAnalysis(rawJobID int64, score float64, isQualified bool, return err } +func (db *DB) CreateJobAnalysisFull(rawJobID int64, score float64, isQualified bool, matched, missing, tailoring, status string) error { + query := ` + INSERT INTO job_analyses (raw_job_id, score, is_qualified, matched_highlights, missing_highlights, tailoring_suggestions, status, analyzed_at) + VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) + ` + _, err := db.Exec(query, rawJobID, score, isQualified, matched, missing, tailoring, status) + return err +} + +func (db *DB) UpdateJobAnalysisFull(rawJobID int64, score float64, isQualified bool, matchedJSON, missingJSON, suggestionsJSON, status string) error { + query := ` + UPDATE job_analyses + SET score = ?, is_qualified = ?, matched_highlights = ?, missing_highlights = ?, tailoring_suggestions = ?, status = ?, analyzed_at = CURRENT_TIMESTAMP + WHERE raw_job_id = ? + ` + _, err := db.Exec(query, score, isQualified, matchedJSON, missingJSON, suggestionsJSON, status, rawJobID) + return err +} + func (db *DB) UpdateJobAnalysisLLM(rawJobID int64, matchedJSON, missingJSON, suggestionsJSON string) error { query := ` UPDATE job_analyses @@ -259,7 +251,7 @@ func (db *DB) UpdateJobAnalysisLLM(rawJobID int64, matchedJSON, missingJSON, sug func (db *DB) GetPendingQualifiedAnalyses() ([]JobWithAnalysis, error) { query := ` - SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.status, r.crawled_at, + SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.crawled_at, a.id, a.score, a.is_qualified, COALESCE(a.matched_highlights, '[]'), COALESCE(a.missing_highlights, '[]'), COALESCE(a.tailoring_suggestions, '[]'), a.status, a.analyzed_at FROM raw_jobs r JOIN job_analyses a ON r.id = a.raw_job_id @@ -269,7 +261,7 @@ func (db *DB) GetPendingQualifiedAnalyses() ([]JobWithAnalysis, error) { if err != nil { return nil, err } - defer rows.Close() + defer func() { _ = rows.Close() }() var results []JobWithAnalysis for rows.Next() { @@ -282,7 +274,7 @@ func (db *DB) GetPendingQualifiedAnalyses() ([]JobWithAnalysis, error) { var analyzedAt sql.NullTime err := rows.Scan( - &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.Status, &item.CrawledAt, + &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.CrawledAt, &aID, &score, &isQualified, &matched, &missing, &suggestions, &aStatus, &analyzedAt, ) if err != nil { @@ -312,7 +304,7 @@ func (db *DB) GetPendingQualifiedAnalyses() ([]JobWithAnalysis, error) { func (db *DB) GetJobsWithAnalysis() ([]JobWithAnalysis, error) { query := ` - SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.status, r.crawled_at, + SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.crawled_at, a.id, a.score, a.is_qualified, COALESCE(a.matched_highlights, '[]'), COALESCE(a.missing_highlights, '[]'), COALESCE(a.tailoring_suggestions, '[]'), COALESCE(a.status, 'PENDING'), a.analyzed_at, app.id, app.status, COALESCE(app.notes, ''), app.applied_at, app.updated_at FROM raw_jobs r @@ -324,7 +316,7 @@ func (db *DB) GetJobsWithAnalysis() ([]JobWithAnalysis, error) { if err != nil { return nil, err } - defer rows.Close() + defer func() { _ = rows.Close() }() var results []JobWithAnalysis for rows.Next() { @@ -340,7 +332,7 @@ func (db *DB) GetJobsWithAnalysis() ([]JobWithAnalysis, error) { var updatedAt sql.NullTime err := rows.Scan( - &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.Status, &item.CrawledAt, + &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.CrawledAt, &aID, &score, &isQualified, &matched, &missing, &suggestions, &aStatus, &analyzedAt, &appID, &appStatus, ¬es, &appliedAt, &updatedAt, ) @@ -386,7 +378,7 @@ func (db *DB) GetJobsWithAnalysis() ([]JobWithAnalysis, error) { func (db *DB) GetJobByID(rawJobID int64) (*JobWithAnalysis, error) { query := ` - SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.status, r.crawled_at, + SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.crawled_at, a.id, a.score, a.is_qualified, COALESCE(a.matched_highlights, '[]'), COALESCE(a.missing_highlights, '[]'), COALESCE(a.tailoring_suggestions, '[]'), COALESCE(a.status, 'PENDING'), a.analyzed_at, app.id, app.status, COALESCE(app.notes, ''), app.applied_at, app.updated_at FROM raw_jobs r @@ -407,7 +399,7 @@ func (db *DB) GetJobByID(rawJobID int64) (*JobWithAnalysis, error) { var updatedAt sql.NullTime err := row.Scan( - &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.Status, &item.CrawledAt, + &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.CrawledAt, &aID, &score, &isQualified, &matched, &missing, &suggestions, &aStatus, &analyzedAt, &appID, &appStatus, ¬es, &appliedAt, &updatedAt, ) @@ -483,7 +475,7 @@ func (db *DB) CreateOrUpdateApplication(rawJobID int64, status, notes string) (* func (db *DB) GetApplications() ([]JobWithAnalysis, error) { query := ` - SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.status, r.crawled_at, + SELECT r.id, r.canonical_id, r.target_id, r.title, r.company, r.location, r.apply_url, r.raw_content, r.crawled_at, a.id, a.score, a.is_qualified, COALESCE(a.matched_highlights, '[]'), COALESCE(a.missing_highlights, '[]'), COALESCE(a.tailoring_suggestions, '[]'), COALESCE(a.status, 'PENDING'), a.analyzed_at, app.id, app.status, COALESCE(app.notes, ''), app.applied_at, app.updated_at FROM job_applications app @@ -495,7 +487,7 @@ func (db *DB) GetApplications() ([]JobWithAnalysis, error) { if err != nil { return nil, err } - defer rows.Close() + defer func() { _ = rows.Close() }() var results []JobWithAnalysis for rows.Next() { @@ -511,7 +503,7 @@ func (db *DB) GetApplications() ([]JobWithAnalysis, error) { var updatedAt time.Time err := rows.Scan( - &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.Status, &item.CrawledAt, + &item.ID, &item.CanonicalID, &tID, &item.Title, &item.Company, &item.Location, &item.ApplyURL, &item.RawContent, &item.CrawledAt, &aID, &score, &isQualified, &matched, &missing, &suggestions, &aStatus, &analyzedAt, &appID, &appStatus, ¬es, &appliedAt, &updatedAt, ) diff --git a/internal/db/queries_test.go b/internal/db/queries_test.go index 015475a..885de49 100644 --- a/internal/db/queries_test.go +++ b/internal/db/queries_test.go @@ -13,7 +13,7 @@ func TestVectorSimilarityDeduplication(t *testing.T) { if err != nil { t.Fatalf("failed to init db: %v", err) } - defer database.Close() + defer func() { _ = database.Close() }() contentA := "Senior Software Engineer - Go Distributed Systems. Experience with microservices, gRPC, and SQLite database pipelines." contentSimilar := "Senior Software Engineer - Go Distributed Systems. Experience with microservices, gRPC, and SQLite database pipelines." @@ -41,9 +41,9 @@ func TestVectorSimilarityDeduplication(t *testing.T) { } // Verify only 2 raw jobs exist in DB - allJobs, err := database.GetPendingRawJobs() + allJobs, err := database.getAllRawJobsForDeduplication() if err != nil { - t.Fatalf("GetPendingRawJobs failed: %v", err) + t.Fatalf("getAllRawJobsForDeduplication failed: %v", err) } if len(allJobs) != 2 { t.Errorf("expected 2 raw jobs in DB after deduplication, got %d", len(allJobs)) diff --git a/internal/db/targets_loader.go b/internal/db/targets_loader.go new file mode 100644 index 0000000..38d7f44 --- /dev/null +++ b/internal/db/targets_loader.go @@ -0,0 +1,48 @@ +package db + +import ( + "fmt" + "os" + + "gopkg.in/yaml.v3" +) + +type targetConfig struct { + Targets []struct { + Name string `yaml:"name"` + TargetURL string `yaml:"target_url"` + Platform string `yaml:"platform"` + IsActive bool `yaml:"is_active"` + } `yaml:"targets"` +} + +// LoadTargetsFromYAML reads a YAML file and inserts targets into the database if it's empty +func (db *DB) LoadTargetsFromYAML(filepath string) error { + data, err := os.ReadFile(filepath) + if err != nil { + return fmt.Errorf("could not read target config: %w", err) + } + + var config targetConfig + if err := yaml.Unmarshal(data, &config); err != nil { + return fmt.Errorf("could not parse target config: %w", err) + } + + // Check if we already have targets + targets, err := db.GetAllTargets() + if err != nil { + return err + } + if len(targets) > 0 { + // Do not overwrite existing targets + return nil + } + + for _, t := range config.Targets { + _, err := db.CreateTarget(t.Name, t.TargetURL, t.Platform) + if err != nil { + return fmt.Errorf("failed to insert target %s: %w", t.Name, err) + } + } + return nil +} diff --git a/internal/db/tasks.go b/internal/db/tasks.go new file mode 100644 index 0000000..976d375 --- /dev/null +++ b/internal/db/tasks.go @@ -0,0 +1,159 @@ +package db + +import ( + "database/sql" + "encoding/json" + "fmt" + "log" +) + +// EnqueueTask adds a new task to the queue +func (db *DB) EnqueueTask(taskType string, payload interface{}) error { + payloadBytes, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("failed to marshal task payload: %w", err) + } + + query := ` + INSERT INTO tasks (task_type, payload, status) + VALUES (?, ?, 'PENDING') + ` + _, err = db.Exec(query, taskType, string(payloadBytes)) + return err +} + +// ClaimNextTask atomically finds and claims the next pending task +func (db *DB) ClaimNextTask() (*Task, error) { + tx, err := db.Begin() + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback() }() + + // Find the oldest PENDING task that is ready to be retried + query := ` + SELECT id, task_type, payload, status, error_message, retry_count, next_retry_at, created_at, updated_at + FROM tasks + WHERE status = 'PENDING' AND next_retry_at <= CURRENT_TIMESTAMP + ORDER BY created_at ASC + LIMIT 1 + ` + + var task Task + var errMsg sql.NullString + err = tx.QueryRow(query).Scan( + &task.ID, &task.TaskType, &task.Payload, &task.Status, &errMsg, + &task.RetryCount, &task.NextRetryAt, &task.CreatedAt, &task.UpdatedAt, + ) + + if err == sql.ErrNoRows { + return nil, nil // No tasks available + } + if err != nil { + return nil, err + } + + if errMsg.Valid { + task.ErrorMessage = errMsg.String + } + + // Claim the task + updateQuery := `UPDATE tasks SET status = 'RUNNING', updated_at = CURRENT_TIMESTAMP WHERE id = ?` + _, err = tx.Exec(updateQuery, task.ID) + if err != nil { + return nil, err + } + + if err := tx.Commit(); err != nil { + return nil, err + } + + return &task, nil +} + +// CompleteTask marks a task as COMPLETED +func (db *DB) CompleteTask(id int64) error { + query := `UPDATE tasks SET status = 'COMPLETED', updated_at = CURRENT_TIMESTAMP WHERE id = ?` + _, err := db.Exec(query, id) + return err +} + +// FailTask marks a task as FAILED or PENDING (for retry) using exponential backoff +func (db *DB) FailTask(id int64, errStr string) error { + // First get current retry count + var retryCount int + err := db.QueryRow(`SELECT retry_count FROM tasks WHERE id = ?`, id).Scan(&retryCount) + if err != nil { + log.Printf("[DB] FailTask: failed to get retry_count for task %d: %v", id, err) + return err + } + + retryCount++ + + // Max retries = 5 (1s, 2s, 4s, 8s, 16s) + status := "PENDING" + if retryCount > 5 { + status = "FAILED" + } + + // Exponential backoff + backoffSeconds := 1 << retryCount // 2^retryCount + + query := ` + UPDATE tasks + SET status = ?, + error_message = ?, + retry_count = ?, + next_retry_at = datetime('now', '+' || ? || ' seconds'), + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + ` + _, err = db.Exec(query, status, errStr, retryCount, fmt.Sprintf("%d", backoffSeconds), id) + return err +} + +// HasPendingFetchTask checks if there is already a PENDING or RUNNING fetch task for a given canonical ID +func (db *DB) HasPendingFetchTask(canonicalID string) (bool, error) { + query := ` + SELECT COUNT(*) FROM tasks + WHERE task_type = 'FETCH' + AND status IN ('PENDING', 'RUNNING') + AND payload LIKE ? + ` + likePattern := fmt.Sprintf(`%%"canonical_id":"%s"%%`, canonicalID) + + var count int + err := db.QueryRow(query, likePattern).Scan(&count) + if err != nil { + return false, err + } + return count > 0, nil +} + +// ClearTaskQueue deletes all tasks or only pending/failed ones +func (db *DB) ClearTaskQueue() error { + query := `DELETE FROM tasks WHERE status IN ('PENDING', 'FAILED', 'COMPLETED')` + _, err := db.Exec(query) + return err +} + +// GetQueueStatus returns counts of tasks by status +func (db *DB) GetQueueStatus() (map[string]int, error) { + query := `SELECT status, COUNT(*) FROM tasks GROUP BY status` + rows, err := db.Query(query) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + counts := make(map[string]int) + for rows.Next() { + var status string + var count int + if err := rows.Scan(&status, &count); err != nil { + return nil, err + } + counts[status] = count + } + return counts, nil +} diff --git a/internal/filter/filter.go b/internal/filter/filter.go index f2da902..7567750 100644 --- a/internal/filter/filter.go +++ b/internal/filter/filter.go @@ -21,14 +21,17 @@ func NewPipeline(filters ...JobFilter) *FilterPipeline { return &FilterPipeline{filters: filters} } -func (p *FilterPipeline) Execute(ctx context.Context, title, rawContent string) (bool, float64) { +func (p *FilterPipeline) Execute(ctx context.Context, title, rawContent string) (bool, float64, string) { var totalScore float64 for _, f := range p.filters { res, err := f.Process(ctx, title, rawContent) - if err != nil || !res.Keep { - return false, 0 + if err != nil { + return false, 0, err.Error() + } + if !res.Keep { + return false, 0, res.Reason } totalScore += res.Score } - return true, totalScore + return true, totalScore, "" } diff --git a/internal/filter/filter_test.go b/internal/filter/filter_test.go index 3a90fa8..e81de0c 100644 --- a/internal/filter/filter_test.go +++ b/internal/filter/filter_test.go @@ -39,7 +39,7 @@ func TestFilterPipeline(t *testing.T) { pipeline := NewPipeline(hardFilter, scoringFilter) // Test job 1: Unpaid job -> rejected by hard filter - keep1, score1 := pipeline.Execute(ctx, "Backend Intern", "Unpaid position working with Go.") + keep1, score1, _ := pipeline.Execute(ctx, "Backend Intern", "Unpaid position working with Go.") if keep1 { t.Errorf("expected job 1 to be filtered out, got keep=true") } @@ -48,7 +48,7 @@ func TestFilterPipeline(t *testing.T) { } // Test job 2: Qualified job -> passes both filters and accumulates score - keep2, score2 := pipeline.Execute(ctx, "Senior Go Backend Engineer", "Building high throughput services.") + keep2, score2, _ := pipeline.Execute(ctx, "Senior Go Backend Engineer", "Building high throughput services.") if !keep2 { t.Errorf("expected job 2 to pass filter pipeline, got keep=false") } diff --git a/internal/service/pipeline.go b/internal/service/pipeline.go deleted file mode 100644 index 5821f54..0000000 --- a/internal/service/pipeline.go +++ /dev/null @@ -1,226 +0,0 @@ -package service - -import ( - "context" - "encoding/json" - "fmt" - "log" - "os" - "time" - - "github.com/hireme/hireme/internal/analyzer" - "github.com/hireme/hireme/internal/crawler" - "github.com/hireme/hireme/internal/db" - "github.com/hireme/hireme/internal/filter" - "github.com/hireme/hireme/pkg/crawler/google" - "gopkg.in/yaml.v3" -) - -type TargetConfigYAML struct { - Targets []struct { - Name string `yaml:"name"` - TargetURL string `yaml:"target_url"` - Platform string `yaml:"platform"` - IsActive bool `yaml:"is_active"` - } `yaml:"targets"` -} - -type PipelineService struct { - db *db.DB - crawlers []crawler.SiteCrawler - filterPipe *filter.FilterPipeline - llmAnalyzer analyzer.LLMAnalyzer -} - -func NewPipelineService(database *db.DB, llmAnalyzer analyzer.LLMAnalyzer) *PipelineService { - hardFilter := filter.NewHardKeywordFilter(nil) - scoringFilter := filter.NewScoringFilter(nil) - pipeline := filter.NewPipeline(hardFilter, scoringFilter) - - crawlers := []crawler.SiteCrawler{ - google.NewGoogleCrawler(), - crawler.NewGenericCrawler(), - } - - return &PipelineService{ - db: database, - crawlers: crawlers, - filterPipe: pipeline, - llmAnalyzer: llmAnalyzer, - } -} - -// LoadTargetsFromYAML loads search targets from YAML configuration file if DB targets table is empty -func (s *PipelineService) LoadTargetsFromYAML(yamlPath string) error { - targets, err := s.db.GetAllTargets() - if err != nil { - return err - } - if len(targets) > 0 { - return nil - } - - data, err := os.ReadFile(yamlPath) - if err != nil { - return fmt.Errorf("could not read targets YAML config at %s: %w", yamlPath, err) - } - - var configData TargetConfigYAML - if err := yaml.Unmarshal(data, &configData); err != nil { - return fmt.Errorf("failed to parse YAML targets config: %w", err) - } - - for _, t := range configData.Targets { - _, err := s.db.CreateTarget(t.Name, t.TargetURL, t.Platform) - if err != nil { - log.Printf("[Pipeline] Error loading target %s: %v", t.Name, err) - } - } - - log.Printf("[Pipeline] Loaded %d search targets from YAML config (%s)", len(configData.Targets), yamlPath) - return nil -} - -func (s *PipelineService) getCrawlerForPlatform(platform string) crawler.SiteCrawler { - for _, c := range s.crawlers { - if c.CanHandle(platform) { - return c - } - } - return s.crawlers[len(s.crawlers)-1] // fallback to generic -} - -// RunPipeline executes Stage 1 (Ingestion) and Stage 2 (Filtering & LLM Enrichment) -func (s *PipelineService) RunPipeline(ctx context.Context) (int, int, error) { - log.Println("[Pipeline] Starting ingestion & processing run...") - - // 1. Fetch active targets - targets, err := s.db.GetActiveTargets() - if err != nil { - return 0, 0, fmt.Errorf("failed to get active targets: %w", err) - } - - newJobsCount := 0 - for _, target := range targets { - log.Printf("[Pipeline] Discovering jobs for target #%d: %s (%s)", target.ID, target.Name, target.TargetURL) - c := s.getCrawlerForPlatform(target.Platform) - discovered, err := c.DiscoverJobs(ctx, target.TargetURL) - if err != nil { - log.Printf("[Pipeline] DiscoverJobs error for target #%d: %v", target.ID, err) - continue - } - - log.Printf("[Pipeline] Discovered %d potential job links for target #%d", len(discovered), target.ID) - - for _, disc := range discovered { - // Check DB deduplication first - exists, err := s.db.CanonicalIDExists(disc.CanonicalID) - if err == nil && exists { - continue // Skip existing job link! - } - - // Fetch & parse detail page for new job only - extracted, err := c.FetchJobDetail(ctx, disc.ApplyURL) - if err != nil { - log.Printf("[Pipeline] FetchJobDetail error for %s: %v", disc.ApplyURL, err) - continue - } - - tID := target.ID - _, inserted, err := s.db.InsertRawJob(extracted.CanonicalID, &tID, extracted.Title, extracted.Company, extracted.Location, extracted.ApplyURL, extracted.RawContent) - if err != nil { - log.Printf("[Pipeline] Error inserting job %s: %v", extracted.Title, err) - continue - } - if inserted { - newJobsCount++ - } - - // Polite delay to prevent HTTP 429 rate-limiting from Google - time.Sleep(500 * time.Millisecond) - } - - _ = s.db.UpdateTargetCrawledAt(target.ID) - } - - // 2. Execute Stage 2 Filter Pipeline on pending raw_jobs - pendingJobs, err := s.db.GetPendingRawJobs() - if err != nil { - return newJobsCount, 0, fmt.Errorf("failed to fetch pending jobs: %w", err) - } - - qualifiedCount := 0 - for _, job := range pendingJobs { - keep, score := s.filterPipe.Execute(ctx, job.Title, job.RawContent) - - if !keep { - log.Printf("[Pipeline] Job #%d (%s) FILTERED OUT by keyword pipeline", job.ID, job.Title) - _ = s.db.CreateJobAnalysis(job.ID, 0, false, "FILTERED_OUT") - _ = s.db.UpdateRawJobStatus(job.ID, "PROCESSED") - continue - } - - log.Printf("[Pipeline] Job #%d (%s) PASSED filter with score %.2f", job.ID, job.Title, score) - _ = s.db.CreateJobAnalysis(job.ID, score, true, "PENDING") - _ = s.db.UpdateRawJobStatus(job.ID, "PROCESSED") - qualifiedCount++ - } - - // 3. Trigger LLM Analysis on qualified PENDING analyses - pendingAnalyses, err := s.db.GetPendingQualifiedAnalyses() - if err != nil { - log.Printf("[Pipeline] Error fetching pending qualified analyses: %v", err) - return newJobsCount, qualifiedCount, nil - } - - for _, item := range pendingAnalyses { - log.Printf("[Pipeline] Triggering LLM analysis for Job #%d (%s @ %s)...", item.ID, item.Title, item.Company) - output, err := s.llmAnalyzer.AnalyzeJD(ctx, item.Title, item.RawContent) - if err != nil { - log.Printf("[Pipeline] LLM analysis failed for job #%d: %v", item.ID, err) - continue - } - - matchedJSON, _ := json.Marshal(output.MatchedHighlights) - missingJSON, _ := json.Marshal(output.MissingHighlights) - suggestionsJSON, _ := json.Marshal(output.TailoringSuggestions) - - err = s.db.UpdateJobAnalysisLLM(item.ID, string(matchedJSON), string(missingJSON), string(suggestionsJSON)) - if err != nil { - log.Printf("[Pipeline] Failed to save LLM analysis for job #%d: %v", item.ID, err) - } else { - log.Printf("[Pipeline] Successfully cached LLM analysis for Job #%d", item.ID) - } - } - - return newJobsCount, qualifiedCount, nil -} - -// AnalyzeJobByID manually triggers LLM analysis on a single job -func (s *PipelineService) AnalyzeJobByID(ctx context.Context, jobID int64) (*db.JobWithAnalysis, error) { - jobWithAnalysis, err := s.db.GetJobByID(jobID) - if err != nil { - return nil, err - } - - output, err := s.llmAnalyzer.AnalyzeJD(ctx, jobWithAnalysis.Title, jobWithAnalysis.RawContent) - if err != nil { - return nil, fmt.Errorf("LLM analysis failed: %w", err) - } - - matchedJSON, _ := json.Marshal(output.MatchedHighlights) - missingJSON, _ := json.Marshal(output.MissingHighlights) - suggestionsJSON, _ := json.Marshal(output.TailoringSuggestions) - - if jobWithAnalysis.Analysis == nil { - _, score := s.filterPipe.Execute(ctx, jobWithAnalysis.Title, jobWithAnalysis.RawContent) - _ = s.db.CreateJobAnalysis(jobID, score, true, "PENDING") - } - - err = s.db.UpdateJobAnalysisLLM(jobID, string(matchedJSON), string(missingJSON), string(suggestionsJSON)) - if err != nil { - return nil, err - } - - return s.db.GetJobByID(jobID) -} diff --git a/internal/ui/handlers.go b/internal/ui/handlers.go index 90b87bc..a0cbbbb 100644 --- a/internal/ui/handlers.go +++ b/internal/ui/handlers.go @@ -1,7 +1,6 @@ package ui import ( - "context" "log" "net/http" "strconv" @@ -9,20 +8,20 @@ import ( "github.com/hireme/hireme/internal/config" "github.com/hireme/hireme/internal/db" - "github.com/hireme/hireme/internal/service" + "github.com/hireme/hireme/internal/worker" ) type Server struct { - cfg *config.Config - db *db.DB - pipeline *service.PipelineService + cfg *config.Config + db *db.DB + pool *worker.WorkerPool } -func NewServer(cfg *config.Config, database *db.DB, pipeline *service.PipelineService) *Server { +func NewServer(cfg *config.Config, database *db.DB, pool *worker.WorkerPool) *Server { return &Server{ - cfg: cfg, - db: database, - pipeline: pipeline, + cfg: cfg, + db: database, + pool: pool, } } @@ -36,7 +35,9 @@ func (s *Server) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("POST /jobs/{id}/analyze", s.handleAnalyzeJob) mux.HandleFunc("POST /applications", s.handleCreateApplication) mux.HandleFunc("PATCH /applications/{id}/status", s.handleUpdateAppStatus) - mux.HandleFunc("POST /pipeline/run", s.handleRunPipeline) + mux.HandleFunc("POST /api/tasks/discover", s.handleDiscoverTasks) + mux.HandleFunc("POST /api/tasks/clean", s.handleCleanTasks) + mux.HandleFunc("POST /api/tasks/toggle", s.handleToggleTasks) } func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) { @@ -175,14 +176,16 @@ func (s *Server) handleAnalyzeJob(w http.ResponseWriter, r *http.Request) { return } - job, err := s.pipeline.AnalyzeJobByID(r.Context(), id) + // Enqueue analyze task instead of synchronous processing + payload := worker.AnalyzePayload{RawJobID: int64(id)} + err = s.db.EnqueueTask("ANALYZE", payload) if err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + http.Error(w, "Failed to enqueue analysis task: "+err.Error(), http.StatusInternalServerError) return } - w.Header().Set("Content-Type", "text/html; charset=utf-8") - _ = RenderJobDetail(w, job) + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`Analysis Queued`)) } func (s *Server) handleCreateApplication(w http.ResponseWriter, r *http.Request) { @@ -269,12 +272,48 @@ func (s *Server) handleUpdateAppStatus(w http.ResponseWriter, r *http.Request) { _ = RenderKanbanBoard(w, updatedApps) } -func (s *Server) handleRunPipeline(w http.ResponseWriter, r *http.Request) { - ctx := context.Background() - _, _, err := s.pipeline.RunPipeline(ctx) +func (s *Server) handleDiscoverTasks(w http.ResponseWriter, r *http.Request) { + targets, err := s.db.GetAllTargets() if err != nil { - log.Printf("[Server] Pipeline run error: %v", err) + http.Error(w, err.Error(), http.StatusInternalServerError) + return } - s.handleGetJobs(w, r) + count := 0 + for _, target := range targets { + if !target.IsActive { + continue + } + payload := worker.DiscoverPayload{ + TargetID: target.ID, + TargetURL: target.TargetURL, + Platform: target.Platform, + } + if err := s.db.EnqueueTask("DISCOVER", payload); err == nil { + count++ + } + } + + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"status": "success", "message": "Queued ` + strconv.Itoa(count) + ` discovery tasks"}`)) +} + +func (s *Server) handleCleanTasks(w http.ResponseWriter, r *http.Request) { + err := s.db.ClearTaskQueue() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"status": "success"}`)) +} + +func (s *Server) handleToggleTasks(w http.ResponseWriter, r *http.Request) { + isRunning := s.pool.Toggle() + status := "paused" + if isRunning { + status = "running" + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"status": "` + status + `"}`)) } diff --git a/internal/ui/templates.go b/internal/ui/templates.go index 3e836bb..8a2a42f 100644 --- a/internal/ui/templates.go +++ b/internal/ui/templates.go @@ -180,16 +180,26 @@ const mainLayoutHTML = `{{ define "page" }} {{ .LLMProvider }} - +
+ + + +
@@ -454,9 +464,15 @@ const jobListHTML = `{{ define "jobs" }} ★ {{ printf "%.0f" .Analysis.Score }} Qualified {{ else }} - - Not Qualified - +
+ + Not Qualified + + {{ $missing := parseJSON .Analysis.MissingHighlights }} + {{ if $missing }} + {{ index $missing 0 }} + {{ end }} +
{{ end }} {{ else }} - @@ -504,7 +520,7 @@ const jobListHTML = `{{ define "jobs" }} {{ else }} - No job postings found yet. Click ⚡ Run Pipeline to ingest jobs. + No job postings found yet. Click 🔍 Discover Jobs to enqueue crawler tasks. {{ end }} diff --git a/internal/worker/pool.go b/internal/worker/pool.go new file mode 100644 index 0000000..acca636 --- /dev/null +++ b/internal/worker/pool.go @@ -0,0 +1,306 @@ +package worker + +import ( + "context" + "encoding/json" + "fmt" + "log" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/hireme/hireme/internal/analyzer" + "github.com/hireme/hireme/internal/crawler" + "github.com/hireme/hireme/internal/db" + "github.com/hireme/hireme/internal/filter" + "github.com/hireme/hireme/pkg/crawler/google" +) + +type WorkerPool struct { + db *db.DB + analyzer analyzer.LLMAnalyzer + filterPipe *filter.FilterPipeline + isRunning atomic.Bool + workerCount int + wg sync.WaitGroup + ctx context.Context + cancel context.CancelFunc +} + +func NewWorkerPool(database *db.DB, analyzer analyzer.LLMAnalyzer, numWorkers int) *WorkerPool { + // Initialize default filters + hardKeywords := []string{"cleared", "clearance", "polygraph", "ts/sci", "secret", "citizen", "citizenship", "phd", "ph.d"} + scoringWeights := []filter.KeywordWeight{ + {Keyword: "golang", Weight: 3.0}, {Keyword: "go", Weight: 2.0}, + {Keyword: "python", Weight: 2.0}, {Keyword: "kubernetes", Weight: 3.0}, + {Keyword: "k8s", Weight: 3.0}, {Keyword: "docker", Weight: 2.0}, + {Keyword: "aws", Weight: 1.0}, {Keyword: "gcp", Weight: 1.0}, + {Keyword: "distributed systems", Weight: 4.0}, + {Keyword: "backend", Weight: 3.0}, {Keyword: "ai", Weight: -5.0}, + {Keyword: "machine learning", Weight: -5.0}, + } + + filterPipe := filter.NewPipeline( + filter.NewHardKeywordFilter(hardKeywords), + filter.NewScoringFilter(scoringWeights), + ) + + return &WorkerPool{ + db: database, + analyzer: analyzer, + filterPipe: filterPipe, + workerCount: numWorkers, + } +} + +func (p *WorkerPool) IsRunning() bool { + return p.isRunning.Load() +} + +func (p *WorkerPool) Toggle() bool { + if p.IsRunning() { + p.Stop() + return false + } + p.Start() + return true +} + +func (p *WorkerPool) Start() { + if p.isRunning.Load() { + return + } + p.ctx, p.cancel = context.WithCancel(context.Background()) + p.isRunning.Store(true) + + for i := 0; i < p.workerCount; i++ { + p.wg.Add(1) + go p.workerLoop(i) + } + log.Printf("[WorkerPool] Started %d workers", p.workerCount) +} + +func (p *WorkerPool) Stop() { + if !p.isRunning.Load() { + return + } + p.cancel() + p.isRunning.Store(false) + p.wg.Wait() + log.Println("[WorkerPool] All workers stopped") +} + +func (p *WorkerPool) workerLoop(workerID int) { + defer p.wg.Done() + log.Printf("[Worker %d] Starting loop", workerID) + + for { + select { + case <-p.ctx.Done(): + log.Printf("[Worker %d] Context cancelled, exiting", workerID) + return + default: + task, err := p.db.ClaimNextTask() + if err != nil { + log.Printf("[Worker %d] Error claiming task: %v", workerID, err) + time.Sleep(2 * time.Second) + continue + } + + if task == nil { + // No tasks available, sleep and poll again + time.Sleep(1 * time.Second) + continue + } + + log.Printf("[Worker %d] Picked up task %d (%s)", workerID, task.ID, task.TaskType) + + // Process task based on type + var processErr error + switch task.TaskType { + case "DISCOVER": + processErr = p.handleDiscover(p.ctx, task) + case "FETCH": + processErr = p.handleFetch(p.ctx, task) + case "PROCESS": + processErr = p.handleProcess(p.ctx, task) + case "ANALYZE": + processErr = p.handleAnalyze(p.ctx, task) + default: + processErr = fmt.Errorf("unknown task type: %s", task.TaskType) + } + + if processErr != nil { + log.Printf("[Worker %d] Task %d FAILED: %v", workerID, task.ID, processErr) + _ = p.db.FailTask(task.ID, processErr.Error()) + } else { + log.Printf("[Worker %d] Task %d COMPLETED", workerID, task.ID) + _ = p.db.CompleteTask(task.ID) + } + } + } +} + +// --------------------------------------------------------- +// Task Payloads +// --------------------------------------------------------- + +type DiscoverPayload struct { + TargetID int64 `json:"target_id"` + TargetURL string `json:"target_url"` + Platform string `json:"platform"` +} + +type FetchPayload struct { + TargetID int64 `json:"target_id"` + Platform string `json:"platform"` + ApplyURL string `json:"apply_url"` + CanonicalID string `json:"canonical_id"` +} + +type ProcessPayload struct { + RawJobID int64 `json:"raw_job_id"` +} + +type AnalyzePayload struct { + RawJobID int64 `json:"raw_job_id"` +} + +// --------------------------------------------------------- +// Task Handlers +// --------------------------------------------------------- + +func (p *WorkerPool) handleDiscover(ctx context.Context, task *db.Task) error { + var payload DiscoverPayload + if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil { + return err + } + + c := p.getCrawlerForPlatform(payload.Platform) + discovered, err := c.DiscoverJobs(ctx, payload.TargetURL) + if err != nil { + return err + } + + for _, disc := range discovered { + // Check deduplication (already fetched/inserted jobs) + exists, err := p.db.CanonicalIDExists(disc.CanonicalID) + if err == nil && exists { + continue // Skip existing job + } + + // Check if it's already in the task queue to be fetched + pending, err := p.db.HasPendingFetchTask(disc.CanonicalID) + if err == nil && pending { + continue // Skip duplicate pending fetch + } + + // Enqueue FETCH task + fetchPayload := FetchPayload{ + TargetID: payload.TargetID, + Platform: payload.Platform, + ApplyURL: disc.ApplyURL, + CanonicalID: disc.CanonicalID, + } + _ = p.db.EnqueueTask("FETCH", fetchPayload) + } + + // Update the target's last crawled time + _ = p.db.UpdateTargetCrawledAt(payload.TargetID) + + return nil +} + +func (p *WorkerPool) handleFetch(ctx context.Context, task *db.Task) error { + var payload FetchPayload + if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil { + return err + } + + c := p.getCrawlerForPlatform(payload.Platform) + extracted, err := c.FetchJobDetail(ctx, payload.ApplyURL) + if err != nil { + return err + } + + tID := payload.TargetID + rawJob, inserted, err := p.db.InsertRawJob(extracted.CanonicalID, &tID, extracted.Title, extracted.Company, extracted.Location, extracted.ApplyURL, extracted.RawContent) + if err != nil { + return err + } + + if inserted && rawJob != nil { + processPayload := ProcessPayload{RawJobID: rawJob.ID} + _ = p.db.EnqueueTask("PROCESS", processPayload) + } + + return nil +} + +func (p *WorkerPool) handleProcess(ctx context.Context, task *db.Task) error { + var payload ProcessPayload + if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil { + return err + } + + job, err := p.db.GetJobByID(payload.RawJobID) + if err != nil { + return err + } + + keep, score, reason := p.filterPipe.Execute(ctx, job.Title, job.RawContent) + if !keep { + // Create a rejected JobAnalysis record. Store reason in missing_highlights JSON array + reasonJSON, _ := json.Marshal([]string{reason}) + _ = p.db.CreateJobAnalysisFull(payload.RawJobID, score, false, "[]", string(reasonJSON), "[]", "PARSED") + return nil + } + + // It's qualified for analysis. Create the base record and enqueue it. + _ = p.db.CreateJobAnalysis(payload.RawJobID, score, true, "PENDING") + + analyzePayload := AnalyzePayload(payload) + _ = p.db.EnqueueTask("ANALYZE", analyzePayload) + + return nil +} + +func (p *WorkerPool) handleAnalyze(ctx context.Context, task *db.Task) error { + var payload AnalyzePayload + if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil { + return err + } + + job, err := p.db.GetJobByID(payload.RawJobID) + if err != nil { + return err + } + + // Use empty matched/missing keywords for now, or pass from filter if stored + result, err := p.analyzer.AnalyzeJD(ctx, job.Title, job.RawContent) + if err != nil { + return err + } + + matchedStr, _ := json.Marshal(result.MatchedHighlights) + missingStr, _ := json.Marshal(result.MissingHighlights) + tailoringStr, _ := json.Marshal(result.TailoringSuggestions) + + err = p.db.UpdateJobAnalysisLLM( + payload.RawJobID, + string(matchedStr), + string(missingStr), + string(tailoringStr), + ) + + return err +} + +func (p *WorkerPool) getCrawlerForPlatform(platform string) crawler.SiteCrawler { + plat := strings.ToLower(platform) + if strings.Contains(plat, "google") { + return google.NewGoogleCrawler() + } + return crawler.NewGenericCrawler() +} diff --git a/pkg/crawler/google/google.go b/pkg/crawler/google/google.go index cc77613..84d39a1 100644 --- a/pkg/crawler/google/google.go +++ b/pkg/crawler/google/google.go @@ -278,9 +278,10 @@ func extractBalancedBracket(s string, startIdx int) string { continue } - if ch == '[' || ch == '{' { + switch ch { + case '[', '{': depth++ - } else if ch == ']' || ch == '}' { + case ']', '}': depth-- if depth == 0 { return s[i : j+1] @@ -328,7 +329,7 @@ func (g *GoogleCrawler) fetchURL(ctx context.Context, urlStr string) (string, er if err != nil { return "", fmt.Errorf("HTTP fetch failed for %s: %w", urlStr, err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusOK { return "", fmt.Errorf("HTTP GET returned status %d for %s", resp.StatusCode, urlStr)