feat: add job status tracking and analysis queue infrastructure

Makefile

@@ -6,6 +6,8 @@ API_DIR=./api
 DOCS_DIR=./docs
 BIN_DIR=./bin
 MAIN_FILE=$(API_DIR)/main.go
+CRON_DIR=./cmd/cron-analyzer
+CRON_MAIN_FILE=$(CRON_DIR)/main.go
 
 # Default target
 
@@ -21,6 +23,11 @@ run: ## Run the application
 	@echo "Running $(APP_NAME)..."
 	@go run $(MAIN_FILE)
 
+.PHONY: cron
+cron: ## Run the job analysis cron service
+	@echo "Starting Job Analysis Cron Service..."
+	@go run $(CRON_MAIN_FILE)
+
 .PHONY: clean
 clean: ## Clean build artifacts
 	@echo "Cleaning build artifacts..."

cmd/cron-analyzer/main.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"log"
+
+	"github.com/jobs-scraper/cron"
+	"github.com/jobs-scraper/infrastructure"
+	"github.com/joho/godotenv"
+)
+
+func main() {
+	// Try to load .local.env first, then fallback to .env
+	if err := godotenv.Load("../../.local.env"); err != nil {
+		log.Println("No .local.env file found, trying .env")
+		if err := godotenv.Load("../../.env"); err != nil {
+			log.Println("No .env file found, using system environment variables")
+		}
+	}
+
+	dbConfig := infrastructure.LoadConfigFromEnv()
+
+	db, err := infrastructure.NewConnection(dbConfig)
+	if err != nil {
+		log.Fatal("Error connecting to db")
+	}
+
+	err = db.Ping()
+	if err != nil {
+		log.Fatal("Error pinging db")
+	}
+
+	log.Println("Successfully connected to db")
+
+	analyzer := cron.NewJobAnalyzer(db)
+	
+	if err := analyzer.AnalyzeJobs(); err != nil {
+		log.Fatalf("Error analyzing jobs: %v", err)
+	}
+
+	log.Println("Job analysis completed successfully")
+}

cron/analyze-jobs.go

@@ -0,0 +1,122 @@
+package cron
+
+import (
+	"database/sql"
+	"encoding/json"
+	"log"
+	"os"
+
+	"github.com/jobs-scraper/infrastructure/rabbitmq"
+	"github.com/jobs-scraper/internal/domain"
+	"github.com/jobs-scraper/internal/repo"
+	"github.com/jobs-scraper/internal/services"
+)
+
+type JobAnalyzer struct {
+	db                 *sql.DB
+	jobRepo            *repo.JobRepository
+	jobDescriptionRepo *repo.JobDescriptionRepository
+	openRouterService  services.OpenRouterService
+	rmq                *rabbitmq.RabbitMQClient
+}
+
+func NewJobAnalyzer(db *sql.DB) *JobAnalyzer {
+	apiKey := os.Getenv("OPENROUTER_API_KEY")
+	model := os.Getenv("CV_AI_MODEL")
+
+	jobRepo := repo.NewJobRepository(db)
+	jobDescriptionRepo := repo.NewJobDescriptionRepository(db)
+	openRouterService := services.NewOpenRouterService(model, apiKey)
+
+	rmq, err := rabbitmq.NewRabbitMQClient()
+	if err != nil {
+		log.Printf("Warning: Failed to create RabbitMQ client: %v", err)
+	}
+
+	return &JobAnalyzer{
+		db:                 db,
+		jobRepo:            jobRepo,
+		jobDescriptionRepo: jobDescriptionRepo,
+		openRouterService:  openRouterService,
+		rmq:                rmq,
+	}
+}
+
+func (ja *JobAnalyzer) AnalyzeJobs() error {
+	// Get jobs that need analysis (status = created)
+	jobs, err := ja.jobRepo.GetJobsByStatus(domain.JobStatusCreated)
+	if err != nil {
+		return err
+	}
+
+	log.Printf("Found %d jobs to analyze", len(jobs))
+
+	// Read CV file
+	cv, err := os.ReadFile("cv.txt")
+	if err != nil {
+		return err
+	}
+
+	for _, job := range jobs {
+		if err := ja.analyzeJob(job, string(cv)); err != nil {
+			log.Printf("Error analyzing job %d: %v", job.ID, err)
+			// Update job status to error
+			if updateErr := ja.jobRepo.UpdateJobStatus(job.ID, domain.JobStatusError); updateErr != nil {
+				log.Printf("Error updating job status to error: %v", updateErr)
+			}
+			continue
+		}
+
+		// Update job status to analyzed
+		if err := ja.jobRepo.UpdateJobStatus(job.ID, domain.JobStatusAnalyzed); err != nil {
+			log.Printf("Error updating job status to analyzed: %v", err)
+		}
+
+		log.Printf("Successfully analyzed job %d", job.ID)
+	}
+
+	return nil
+}
+
+func (ja *JobAnalyzer) analyzeJob(job domain.Job, cv string) error {
+	// Get job description
+	jobDescription, jobCriteria, err := ja.jobDescriptionRepo.GetJobDescriptionByJobID(job.ID)
+	if err != nil {
+		return err
+	}
+
+	// Analyze job with CV
+	result, err := ja.openRouterService.AnalyzeJobDescription(cv, domain.JobDescription{
+		JobID:       job.ID,
+		Description: jobDescription,
+		Criteria:    jobCriteria,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	log.Printf("Analysis result for job %d: Recommendation=%s, Score=%d", job.ID, result.Recommendation, result.ConfidenceScore)
+
+	// Publish CV analysis message to RabbitMQ if available
+	if ja.rmq != nil {
+		message := rabbitmq.CvAnalyzeMessage{
+			JobID: job.ID,
+		}
+		
+		if err := ja.publishCvAnalysisMessage(message); err != nil {
+			log.Printf("Warning: Failed to publish CV analysis message: %v", err)
+		}
+	}
+
+	return nil
+}
+
+func (ja *JobAnalyzer) publishCvAnalysisMessage(message rabbitmq.CvAnalyzeMessage) error {
+	data, err := json.Marshal(message)
+	if err != nil {
+		return err
+	}
+
+	return ja.rmq.Publish(rabbitmq.CvAnalyzeQueue, rabbitmq.CvAnalyzeExchange, data)
+}

infrastructure/rabbitmq/messages.go

@@ -0,0 +1,5 @@
+package rabbitmq
+
+type CvAnalyzeMessage struct {
+	JobID int64 `json:"jobID"`
+}

infrastructure/rabbitmq/rabbitmq.go

@@ -17,6 +17,8 @@ const (
 	TokyoDevQueue      = "scraper.tokyodev"
 	JapanDevQueue      = "scraper.japandev"
 	DeadLetterExchange = "scraper_dlx"
+	CvAnalyzeExchange  = "cv_exchange"
+	CvAnalyzeQueue     = "cv.analyze"
 )
 
 type RabbitMQClient struct {

internal/domain/job.go

@@ -5,6 +5,8 @@ import "time"
 // JobProvider represents different job board providers
 type JobProvider int
 
+type JobStatus int
+
 const (
 	LinkedIn JobProvider = iota
 	Indeed
@@ -14,6 +16,12 @@ const (
 	JapanDev
 )
 
+const (
+	JobStatusCreated JobStatus = iota
+	JobStatusAnalyzed
+	JobStatusError
+)
+
 type Job struct {
 	ID          int64
 	Title       string
@@ -23,6 +31,7 @@ type Job struct {
 	JobLink     string
 	Provider    JobProvider
 	JobPostTime *time.Time
+	Status      JobStatus
 }
 
 type SearchQuery struct {

internal/repo/job.go

@@ -137,3 +137,43 @@ func prepareQueryCreateBulk(s string, models []*domain.Job) (string, []interface
 
 	return fmt.Sprintf(s, bf.String()), values
 }
+
+func (r *JobRepository) GetJobsByStatus(status domain.JobStatus) ([]domain.Job, error) {
+	query := `
+		SELECT id, title, company, company_link, location, job_link, job_timestamp
+		FROM jobs 
+		WHERE status = $1
+	`
+	
+	rows, err := r.db.Query(query, int(status))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var jobs []domain.Job
+	for rows.Next() {
+		var job domain.Job
+		err := rows.Scan(
+			&job.ID,
+			&job.Title,
+			&job.Company,
+			&job.CompanyLink,
+			&job.Location,
+			&job.JobLink,
+			&job.JobPostTime,
+		)
+		if err != nil {
+			return nil, err
+		}
+		jobs = append(jobs, job)
+	}
+
+	return jobs, rows.Err()
+}
+
+func (r *JobRepository) UpdateJobStatus(jobID int64, status domain.JobStatus) error {
+	query := `UPDATE jobs SET status = $1 WHERE id = $2`
+	_, err := r.db.Exec(query, int(status), jobID)
+	return err
+}

migrations/007_add_job_status.down.sql

@@ -0,0 +1 @@
+ALTER TABLE jobs DROP COLUMN status;

migrations/007_add_job_status.up.sql

@@ -0,0 +1 @@
+ALTER TABLE jobs ADD COLUMN status INTEGER DEFAULT 0;