jobs-monorepo/internal/repo/job.go

package repo

import (
	"bytes"
	"database/sql"
	"fmt"
	"strconv"

	"github.com/jobs-scraper/internal/domain"
)

type JobRepository struct {
	db *sql.DB
}

func NewJobRepository(db *sql.DB) *JobRepository {
	return &JobRepository{db: db}
}

func (r *JobRepository) SaveJobs(jobs []domain.Job) error {
	if len(jobs) == 0 {
		return nil
	}

	// Deduplicate jobs by ID to avoid duplicate errors
	jobMap := make(map[int64]domain.Job)
	for _, job := range jobs {
		jobMap[job.ID] = job
	}

	// Convert to slice of pointers for the bulk function
	uniqueJobs := make([]*domain.Job, 0, len(jobMap))
	for _, job := range jobMap {
		jobCopy := job // Important: create a copy to avoid pointer issues
		uniqueJobs = append(uniqueJobs, &jobCopy)
	}

	sqlTemplate := `
        INSERT INTO jobs (id, title, company, company_link, location, job_link, job_timestamp)
        VALUES %s
        ON CONFLICT (id) DO UPDATE SET
        title = EXCLUDED.title,
        company = EXCLUDED.company,
        company_link = EXCLUDED.company_link,
        location = EXCLUDED.location,
        job_link = EXCLUDED.job_link,
        job_timestamp = EXCLUDED.job_timestamp
    `

	sqlStatement, vals := prepareQueryCreateBulk(sqlTemplate, uniqueJobs)

	_, err := r.db.Exec(sqlStatement, vals...)
	if err != nil {
		return fmt.Errorf("error inserting jobs: %v", err)
	}

	return nil
}

func (r *JobRepository) GetAllJobs() ([]domain.Job, error) {
	rows, err := r.db.Query("SELECT * FROM jobs")
	if err != nil {
		return nil, fmt.Errorf("error querying jobs: %v", err)
	}
	defer rows.Close()

	var jobs []domain.Job
	for rows.Next() {
		var job domain.Job
		if err := rows.Scan(&job.ID, &job.Title, &job.Company, &job.CompanyLink, &job.Location, &job.JobLink); err != nil {
			return nil, fmt.Errorf("error scanning job row: %v", err)
		}
		jobs = append(jobs, job)
	}

	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("error iterating over job rows: %v", err)
	}

	return jobs, nil
}
func (r *JobRepository) GetJobByID(id int) (*domain.Job, error) {
	var job domain.Job

	sqlStatement := `
		SELECT id, title, company, company_link, location, job_link, job_timestamp 
		FROM jobs 
		WHERE id = $1
	`

	err := r.db.QueryRow(sqlStatement, id).Scan(
		&job.ID,
		&job.Title,
		&job.Company,
		&job.CompanyLink,
		&job.Location,
		&job.JobLink,
		&job.JobPostTime,
	)

	if err == sql.ErrNoRows {
		return nil, fmt.Errorf("job with ID %d not found", id)
	}

	if err != nil {
		return nil, fmt.Errorf("error querying job: %v", err)
	}

	return &job, nil
}

func prepareQueryCreateBulk(s string, models []*domain.Job) (string, []interface{}) {
	bf := bytes.Buffer{}
	values := make([]interface{}, 0, len(models)*7)

	for i, v := range models {
		values = append(values, v.ID, v.Title, v.Company,
			v.CompanyLink, v.Location, v.JobLink, v.JobPostTime,
		)

		numFields := 7 // the number of fields you are inserting
		n := i * numFields

		bf.WriteString("(")
		for j := 0; j < numFields; j++ {
			bf.WriteString("$")
			bf.WriteString(strconv.Itoa(n + j + 1))
			if j < numFields-1 {
				bf.WriteString(", ")
			}
		}
		bf.WriteString(")")
		if i < len(models)-1 {
			bf.WriteString(", ")
		}
	}

	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
}