package config

import (
        "fmt"
        "os"
        "strconv"

        _ "github.com/joho/godotenv/autoload"
)

type Config struct {
        GRPCPort    string
        HTTPPort    string
        ScraperHost string
        ScraperPort string

        DBHost     string
        DBPort     string
        DBUser     string
        DBPassword string
        DBName     string

        ProxyFilePath string
        WorkerCount   int
}

func Load() (Config, error) {
        wcStr := getEnv("WORKER_COUNT", "valid int val must be present in env")
        wc, err := strconv.Atoi(wcStr)
        if err != nil {
                return Config{}, fmt.Errorf("worker count is not configure properly: %v", err.Error())
        }

        return Config{
                GRPCPort:      getEnv("GRPC_SERVER_PORT", "9090"),
                HTTPPort:      getEnv("HTTP_SERVER_PORT", "8090"),
                ScraperHost:   getEnv("SCRAPER_GRPC_ADDRESS", "localhost"),
                ScraperPort:   getEnv("SCRAPER_GRPC_PORT", "50051"),
                DBHost:        getEnv("POSTGRES_HOST", "localhost"),
                DBPort:        getEnv("POSTGRES_PORT", "5432"),
                DBUser:        getEnv("POSTGRES_USER", "postgres"),
                DBPassword:    getEnv("POSTGRES_PASSWORD", ""),
                DBName:        getEnv("POSTGRES_DB", "postgres"),
                ProxyFilePath: getEnv("PROXY_FILE_PATH", "proxy.txt"),
                WorkerCount:   wc,
        }, nil
}

func getEnv(key, defaultVal string) string {
        if val := os.Getenv(key); val != "" {
                return val
        }

        return defaultVal
}

package db

import (
        "context"
        "fmt"

        "codeberg.org/at375/pfjobrunner/internal/config"
        "github.com/jackc/pgx/v5/pgxpool"
)

func ConnectPostgres(ctx context.Context, cfg config.Config) (*pgxpool.Pool, error) {
        connStr := buildPostgresConnString(cfg.DBUser, cfg.DBPassword, cfg.DBHost, cfg.DBPort, cfg.DBName)

        pool, err := pgxpool.New(ctx, connStr)
        if err != nil {
                return nil, err
        }

        if err := pool.Ping(ctx); err != nil {
                return nil, err
        }

        return pool, nil
}

func buildPostgresConnString(user string, password string, host string, port string, dbname string) string {
        return fmt.Sprintf(
                "postgres://%s:%s@%s:%s/%s?sslmode=disable",
                user, password, host, port, dbname,
        )
}

package grpcserver

import (
        "context"

        "codeberg.org/at375/pfjobrunner/gen/pbjobrunner"
        "codeberg.org/at375/pfjobrunner/internal/runner"
        "google.golang.org/grpc/codes"
        "google.golang.org/grpc/status"
)

type RunnerGRPCServer struct {
        pbjobrunner.UnimplementedJobRunnerServiceServer
        Jr JobRunner
}

type JobRunner interface {
        Add(ctx context.Context, gazetteID int32, name string, total, completed, failed int64) (int32, error)
        Delete(gazetteID int32) error
        Pause() error
        Resume() error
        Stop() error
        Start() error
        SetOrder(order []int32) error
        SetPatentDeadline(days int, months int, years int) error
        GetFullStreamMessage() runner.StreamMessage
}

func (s *RunnerGRPCServer) AddJob(_ context.Context, req *pbjobrunner.AddJobRequest) (*pbjobrunner.AddJobResponse, error) {
        pos, err := s.Jr.Add(context.Background(), req.GazetteId, req.Name, req.Total, req.Completed, req.Failed)
        if err != nil {
                return nil, status.Errorf(codes.InvalidArgument, "%s", err.Error())
        }
        return &pbjobrunner.AddJobResponse{QueuePosition: pos}, nil
}

func (s *RunnerGRPCServer) DeleteJob(_ context.Context, req *pbjobrunner.DeleteJobRequest) (*pbjobrunner.DeleteJobResponse, error) {
        if err := s.Jr.Delete(req.GazetteId); err != nil {
                return nil, status.Errorf(codes.InvalidArgument, "%s", err.Error())
        }
        return &pbjobrunner.DeleteJobResponse{}, nil
}

func (s *RunnerGRPCServer) Pause(_ context.Context, _ *pbjobrunner.Empty) (*pbjobrunner.Empty, error) {
        if err := s.Jr.Pause(); err != nil {
                return nil, status.Errorf(codes.FailedPrecondition, "%s", err.Error())
        }
        return &pbjobrunner.Empty{}, nil
}

func (s *RunnerGRPCServer) Resume(_ context.Context, _ *pbjobrunner.Empty) (*pbjobrunner.Empty, error) {
        if err := s.Jr.Resume(); err != nil {
                return nil, status.Errorf(codes.FailedPrecondition, "%s", err.Error())
        }
        return &pbjobrunner.Empty{}, nil
}

func (s *RunnerGRPCServer) Stop(_ context.Context, _ *pbjobrunner.Empty) (*pbjobrunner.Empty, error) {
        if err := s.Jr.Stop(); err != nil {
                return nil, status.Errorf(codes.FailedPrecondition, "%s", err.Error())
        }
        return &pbjobrunner.Empty{}, nil
}

func (s *RunnerGRPCServer) Start(_ context.Context, _ *pbjobrunner.Empty) (*pbjobrunner.Empty, error) {
        if err := s.Jr.Start(); err != nil {
                return nil, status.Errorf(codes.FailedPrecondition, "%s", err.Error())
        }
        return &pbjobrunner.Empty{}, nil
}

func (s *RunnerGRPCServer) SetOrder(_ context.Context, req *pbjobrunner.SetOrderRequest) (*pbjobrunner.Empty, error) {
        if err := s.Jr.SetOrder(req.GazetteIds); err != nil {
                return nil, status.Errorf(codes.InvalidArgument, "%s", err.Error())
        }
        return &pbjobrunner.Empty{}, nil
}

func (s *RunnerGRPCServer) SetPatentDeadline(_ context.Context, req *pbjobrunner.SetPatentDeadlineRequest) (*pbjobrunner.Empty, error) {
        if err := s.Jr.SetPatentDeadline(int(req.Days), int(req.Months), int(req.Years)); err != nil {
                return nil, status.Errorf(codes.InvalidArgument, "%s", err.Error())
        }
        return &pbjobrunner.Empty{}, nil
}

func mapJob(j runner.Job) *pbjobrunner.Job {
        return &pbjobrunner.Job{
                GazetteId: j.GazetteID,
                Name:      j.Name,
                Completed: j.Completed,
                Failed:    j.Failed,
                Total:     j.Total,
        }
}

package metrics

import (
        grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
        "github.com/prometheus/client_golang/prometheus"
        "github.com/prometheus/client_golang/prometheus/collectors"
)

// TODO: add metrics for n of streams
var (
        JobsTotal = prometheus.NewCounterVec(
                prometheus.CounterOpts{
                        Name: "pf_job_runner_jobs_total",
                        Help: "Total jobs processed",
                },
                []string{"gazette_id", "status"}, // status: success | missed | error
        )

        JobQueueDepth = prometheus.NewGauge(prometheus.GaugeOpts{
                Name: "pf_job_runner_job_queue_depth",
                Help: "Current number of jobs in the queue",
        })

        GrpcClientConnFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
                Name: "pf_job_runner_grpc_client_conn_failed_total",
                Help: "Number of gRPC calls by the job runner that failed",
        }, []string{"method"})
)

func Register(reg prometheus.Registerer) {
        grpc_prometheus.EnableClientHandlingTimeHistogram()

        reg.MustRegister(
                grpc_prometheus.DefaultClientMetrics,
                GrpcClientConnFailed,
                JobsTotal,
                JobQueueDepth,
                collectors.NewGoCollector(),
                collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
        )
}

package runner

import (
        "context"
        "errors"
        "fmt"
        "io"
        "strconv"
        "sync"
        "time"

        "codeberg.org/at375/patent-flow/common/types"
        "codeberg.org/at375/patent-flow/gen/sqlcgen"
        "codeberg.org/at375/patent-flow/logger"
        "codeberg.org/at375/patent-flow/util/mapper"
        "codeberg.org/at375/patent-flow/util/proxyhelper"
        "codeberg.org/at375/pfjobrunner/gen/pbmodel"
        "codeberg.org/at375/pfjobrunner/internal/metrics"
        "github.com/jackc/pgx/v5"
        "github.com/jackc/pgx/v5/pgxpool"
)

type RunnerQueries interface {
        GetUnprocessedPatentUrlByGazette(ctx context.Context, gazetteID int32) (sqlcgen.PatentUrl, error)
        SetPatentUrlStatus(ctx context.Context, arg sqlcgen.SetPatentUrlStatusParams) error
        SetPatentUrlFailed(ctx context.Context, id int64) error
        IncrementPatentUrlsAttempts(ctx context.Context, id int64) error

        CreateOrGetApplicant(ctx context.Context, arg sqlcgen.CreateOrGetApplicantParams) (sqlcgen.Applicant, error)
        CreateOrGetPatent(ctx context.Context, arg sqlcgen.CreateOrGetPatentParams) (sqlcgen.CreateOrGetPatentRow, error)
        CreateOrGetAgent(ctx context.Context, arg sqlcgen.CreateOrGetAgentParams) (sqlcgen.Agent, error)
        CreatePatentAgent(ctx context.Context, arg sqlcgen.CreatePatentAgentParams) error

        WithTx(tx pgx.Tx) *sqlcgen.Queries
}

// TODO: make queue persistent? just save order somewhere?

type Job struct {
        GazetteID int32  `json:"gazette_id"`
        Name      string `json:"name"`
        Completed int64  `json:"processed"`
        Failed    int64  `json:"failed"`
        Total     int64  `json:"total"`
}

type StreamEventType string

const (
        EventFull     StreamEventType = "full"
        EventProgress StreamEventType = "progress"
        EventOrder    StreamEventType = "order"
        EventStatus   StreamEventType = "status"
)

type StreamMessage struct {
        Type StreamEventType `json:"type"`
        Data any             `json:"data"`
}

type OrderMessage struct {
        Order      []int32 `json:"order"`
        CurrentJob Job     `json:"current_job"`
}

type StatusMessage struct {
        Status JobStatus `json:"status"`
        Msg    string    `json:"message"`
}

type ProgressMessage struct {
        CurrentJob     Job                `json:"current_job"`
        CurrentPatents map[int32][]string `json:"current_patents"`
        Msg            string             `json:"message"`
}

type FullMessage struct {
        ProgressMessage
        Status   JobStatus     `json:"status"`
        Msg      string        `json:"message"`
        Order    []int32       `json:"order"`
        AllJobs  map[int32]Job `json:"all_jobs"`
        Deadline deadline      `json:"pat_deadline"`
}

type subscription struct {
        ch   chan StreamMessage
        done chan struct{}
}

type deadline struct {
        Days   int `json:"d"`
        Months int `json:"m"`
        Years  int `json:"y"`
}

type JobRunner struct {
        mu   sync.RWMutex
        cond *sync.Cond
        wg   sync.WaitGroup

        rootCtx context.Context

        jobs        map[int32]*Job     // protected by mu
        order       []int32            // gazette IDs in priority order, protected by mu
        currentPats map[int32][]string // string paturl names to display
        deleteJobs  map[int32]struct{}
        status      JobStatus
        runningID   int32 // gazette ID of the active goroutine, 0 if none
        msg         string

        pauseCh  chan struct{}
        resumeCh chan struct{}
        cancel   context.CancelFunc

        sem         chan struct{}
        subscribers []*subscription

        pool    *pgxpool.Pool
        queries RunnerQueries

        proxyPool proxyhelper.ProxyProvider
        logger    logger.Logger
        grpc      pbmodel.PatentServiceClient

        patDeadline deadline

        // Injectable for testing. Defaults are set in NewJobRunner to the real implementations.
        storeProcessedPatentFn func(ctx context.Context, patUrl types.PatentUrl, patent types.Patent, applicant types.Applicant, agents []types.Agent) error
        markPatentUrlFailedFn  func(ctx context.Context, patUrl *types.PatentUrl) error
}

func NewJobRunner(
        ctx context.Context,
        pool *pgxpool.Pool,
        queries RunnerQueries,
        logger logger.Logger,
        grpc pbmodel.PatentServiceClient,
        workerCount int,
        proxyPool proxyhelper.ProxyProvider,
) *JobRunner {
        jr := &JobRunner{
                jobs:        make(map[int32]*Job),
                currentPats: make(map[int32][]string),
                deleteJobs:  make(map[int32]struct{}),
                pauseCh:     make(chan struct{}, 1),
                resumeCh:    make(chan struct{}, 1),
                status:      StatusStopped,
                rootCtx:     ctx,
                sem:         make(chan struct{}, workerCount),
                subscribers: make([]*subscription, 0, 100),
                pool:        pool,
                queries:     queries,
                proxyPool:   proxyPool,
                logger:      logger,
                grpc:        grpc,
                patDeadline: deadline{Days: 0, Months: 9, Years: 2},
        }
        jr.cond = sync.NewCond(&jr.mu)

        jr.storeProcessedPatentFn = jr.storeProcessedPatent
        jr.markPatentUrlFailedFn = jr.markPatentUrlFailed

        return jr
}

func (r *JobRunner) error(ctx context.Context, err error, msg string, fields map[string]any) error {
        if fields == nil {
                fields = map[string]any{}
        }
        fields["error"] = err.Error()
        r.logger.Error(ctx, msg, fields)
        return fmt.Errorf("%s: %w", msg, err)
}

func (r *JobRunner) execTx(ctx context.Context, fn func(RunnerQueries) error) error {
        tx, err := r.pool.Begin(ctx)
        if err != nil {
                return fmt.Errorf("failed to begin transaction: %w", err)
        }

        q := r.queries.WithTx(tx)
        if err = fn(q); err != nil {
                if rbErr := tx.Rollback(ctx); rbErr != nil {
                        return fmt.Errorf("tx err: %v, rollback err: %v", err, rbErr)
                }
                return err
        }

        return tx.Commit(ctx)
}

func (r *JobRunner) getUnprocessedPatentUrlByGazette(ctx context.Context, gazetteID int32) (types.PatentUrl, error) {
        patentUrl, err := r.queries.GetUnprocessedPatentUrlByGazette(ctx, gazetteID)
        if err != nil {
                if errors.Is(err, pgx.ErrNoRows) {
                        return types.PatentUrl{}, nil
                }
                return types.PatentUrl{}, r.error(ctx, err, "failed to get unprocessed patent url", map[string]any{
                        "gazette_id": gazetteID,
                })
        }

        return mapper.PatentUrlFromDB(patentUrl), nil
}

func (r *JobRunner) storeProcessedPatent(ctx context.Context, patUrl types.PatentUrl, patent types.Patent, applicant types.Applicant, agents []types.Agent) error {
        return r.execTx(ctx, func(q RunnerQueries) error {
                if applicant.Name == "" || applicant.Address == "" {
                        return r.error(ctx, fmt.Errorf("supplied applicant's critical field is empty"), "validation error", nil)
                }
                appl, err := q.CreateOrGetApplicant(ctx, sqlcgen.CreateOrGetApplicantParams{
                        Name:     applicant.Name,
                        Address:  applicant.Address,
                        Postcode: applicant.Postcode,
                        Country:  applicant.Country,
                })
                if err != nil {
                        return r.error(ctx, err, "failed to create or get applicant", map[string]any{
                                "name":     applicant.Name,
                                "address":  applicant.Address,
                                "postcode": applicant.Postcode,
                                "country":  applicant.Country,
                        })
                }

                if patent.Number == "" ||
                        patent.Name == "" ||
                        patent.PubDate == (time.Time{}) ||
                        patent.PriorityDate == (time.Time{}) {
                        return r.error(ctx, fmt.Errorf("supplied patent's critical field is empty"), "validation error", nil)
                }

                pat, err := q.CreateOrGetPatent(ctx, sqlcgen.CreateOrGetPatentParams{
                        Number:       patent.Number,
                        Name:         patent.Name,
                        ApplicantID:  appl.ID,
                        GazetteID:    patUrl.GazetteID,
                        PubDate:      patent.PubDate,
                        PriorityDate: patent.PriorityDate,
                        DeadlineDate: patent.PriorityDate.AddDate(r.patDeadline.Years, r.patDeadline.Months, r.patDeadline.Days),
                        Countries:    patent.Countries,
                })
                if err != nil {
                        return r.error(ctx, err, "failed to create or get patent", map[string]any{
                                "number":        patent.Number,
                                "name":          patent.Name,
                                "applicant_id":  appl.ID,
                                "gazette_id":    patUrl.GazetteID,
                                "pub_date":      patent.PubDate,
                                "priority_date": patent.PriorityDate,
                                "countries":     patent.Countries,
                        })
                }

                for _, agent := range agents {
                        if agent.Name == "" || agent.Address == "" {
                                return r.error(ctx, fmt.Errorf("supplied agent's critical field is empty"), "validation error", nil)
                        }
                        a, err := q.CreateOrGetAgent(ctx, sqlcgen.CreateOrGetAgentParams{
                                Name:     agent.Name,
                                Address:  agent.Address,
                                Postcode: agent.Postcode,
                                Country:  agent.Country,
                        })
                        if err != nil {
                                return r.error(ctx, err, "failed to create or get agent", map[string]any{
                                        "name":     agent.Name,
                                        "address":  agent.Address,
                                        "postcode": agent.Postcode,
                                        "country":  agent.Country,
                                })
                        }

                        err = q.CreatePatentAgent(ctx, sqlcgen.CreatePatentAgentParams{
                                PatentID: pat.ID,
                                AgentID:  a.ID,
                        })
                        if err != nil {
                                return r.error(ctx, err, "failed to create patent-agent relation", map[string]any{
                                        "patent_id": pat.ID,
                                        "agent_id":  a.ID,
                                })
                        }
                }

                err = q.SetPatentUrlStatus(ctx, sqlcgen.SetPatentUrlStatusParams{
                        ID:     patUrl.ID,
                        Status: 2,
                })
                if err != nil {
                        return r.error(ctx, err, "failed to set patent url status", map[string]any{
                                "patent_url_id": patUrl.ID,
                                "status":        2,
                        })
                }

                return nil
        })
}

func (r *JobRunner) markPatentUrlFailed(ctx context.Context, patUrl *types.PatentUrl) error {
        err := r.execTx(ctx, func(q RunnerQueries) error {
                if err := q.IncrementPatentUrlsAttempts(ctx, patUrl.ID); err != nil {
                        return fmt.Errorf("increment attempts: %w", err)
                }
                if err := q.SetPatentUrlStatus(ctx, sqlcgen.SetPatentUrlStatusParams{
                        ID:     patUrl.ID,
                        Status: 0,
                }); err != nil {
                        return fmt.Errorf("reset status: %w", err)
                }
                return nil
        })
        if err != nil {
                return err
        }

        // reflect changed attempts state on the instance
        patUrl.Attempts++

        return nil
}

func (r *JobRunner) Subscribe() (<-chan StreamMessage, func()) {
        sub := &subscription{
                ch:   make(chan StreamMessage, 10),
                done: make(chan struct{}),
        }

        r.mu.Lock()
        r.subscribers = append(r.subscribers, sub)
        r.mu.Unlock()

        // NOTE: sub struct is GC'ed after this closure execs and goes out of consumer's scope
        unsubscribe := func() {
                r.mu.Lock()
                subs := r.subscribers
                for i, s := range subs {
                        if s == sub {
                                r.subscribers = append(subs[:i], subs[i+1:]...)
                                break
                        }
                }
                r.mu.Unlock()
                close(sub.done)
        }

        return sub.ch, unsubscribe
}

func (r *JobRunner) GetFullStreamMessage() StreamMessage {
        r.mu.RLock()
        msg := r.getFullMessage()
        r.mu.RUnlock()

        event := StreamMessage{
                Type: EventFull,
                Data: msg,
        }

        return event
}

// uses RLock
func (r *JobRunner) broadcast(eventType StreamEventType) {
        r.mu.RLock()
        var msg any
        switch eventType {
        case EventProgress:
                msg = r.getProgressMessage()
        case EventOrder:
                msg = r.getOrderMessage()
        case EventStatus:
                msg = r.getStatusMessage()
        default:
                msg = r.getFullMessage()
        }

        event := StreamMessage{
                Type: eventType,
                Data: msg,
        }

        subs := append([]*subscription(nil), r.subscribers...)
        r.mu.RUnlock()

        for _, sub := range subs {
                select {
                case <-sub.done:
                default:
                        select {
                        case sub.ch <- event:
                        default:
                        }
                }
        }
}

// caller must hold Lock/RLock
func (r *JobRunner) filterDeleted(order []int32) []int32 {
        if len(r.deleteJobs) > 0 && len(order) > 0 {
                filtered := make([]int32, 0, len(order))
                for _, id := range order {
                        if _, del := r.deleteJobs[id]; !del {
                                filtered = append(filtered, id)
                        }
                }
                order = filtered
        }

        return order
}

// caller must hold Lock/RLock
func (r *JobRunner) getFullMessage() FullMessage {
        jobsCopy := make(map[int32]Job, len(r.currentPats))
        orderCopy := r.filterDeleted(r.order) // safe since underlying array is never mutated, only replaced (copy will hold old descriptor)
        statusCopy := r.status
        msgCopy := r.msg
        dlCopy := r.patDeadline

        for k, v := range r.jobs {
                j := Job{
                        GazetteID: v.GazetteID,
                        Name:      v.Name,
                        Completed: v.Completed,
                        Failed:    v.Failed,
                        Total:     v.Total,
                }
                jobsCopy[k] = j
        }

        pMsg := r.getProgressMessage()

        msg := FullMessage{
                ProgressMessage: pMsg,
                Status:          statusCopy,
                Order:           orderCopy,
                Msg:             msgCopy,
                AllJobs:         jobsCopy,
                Deadline:        dlCopy,
        }

        return msg
}

// caller must hold Lock/RLock
func (r *JobRunner) getProgressMessage() ProgressMessage {
        job, ok := r.jobs[r.runningID]
        if !ok {
                job = &Job{}
        }

        currPatCopy := make(map[int32][]string, len(r.currentPats))

        for k, v := range r.currentPats {
                s := make([]string, len(v))
                copy(s, v)
                currPatCopy[k] = s
        }

        msg := ProgressMessage{
                CurrentJob: Job{
                        GazetteID: job.GazetteID,
                        Name:      job.Name,
                        Completed: job.Completed,
                        Failed:    job.Failed,
                        Total:     job.Total,
                },
                CurrentPatents: currPatCopy,
                Msg:            r.msg,
        }

        return msg
}

// caller must hold Lock/RLock
func (r *JobRunner) getOrderMessage() OrderMessage {
        job, ok := r.jobs[r.runningID]
        if !ok {
                job = &Job{}
        }

        orderCopy := r.filterDeleted(r.order)

        msg := OrderMessage{
                Order: orderCopy,
                CurrentJob: Job{
                        GazetteID: job.GazetteID,
                        Name:      job.Name,
                        Completed: job.Completed,
                        Failed:    job.Failed,
                        Total:     job.Total,
                },
        }

        return msg
}

// caller must hold Lock/RLock
func (r *JobRunner) getStatusMessage() StatusMessage {
        msg := StatusMessage{
                Status: r.status,
                Msg:    r.msg,
        }

        return msg
}

func (r *JobRunner) Add(ctx context.Context, gazetteID int32, gazetteName string, total, completed, failed int64) (int32, error) {
        r.mu.RLock()
        _, ok := r.jobs[gazetteID]
        if ok {
                r.mu.RUnlock()
                return 0, errors.New("job already exists")
        }
        r.mu.RUnlock()

        r.mu.Lock()

        job := &Job{
                GazetteID: gazetteID,
                Name:      gazetteName,
                Total:     total,
                Completed: completed,
                Failed:    failed,
        }

        r.jobs[gazetteID] = job
        r.order = append(r.order, gazetteID)
        metrics.JobQueueDepth.Inc()

        startRunner := len(r.order) == 1
        if startRunner {
                r.status = StatusRunning
        }

        p := int32(len(r.order) - 1)

        r.mu.Unlock()

        if startRunner {
                r.startRun()
        }

        r.broadcast(EventFull)

        return p, nil
}

func (r *JobRunner) Delete(gazetteID int32) error {
        r.mu.Lock()
        _, ok := r.jobs[gazetteID]
        if !ok {
                r.mu.Unlock()
                return errors.New("job not found")
        }
        r.deleteJobs[gazetteID] = struct{}{}
        r.mu.Unlock()

        go r.cleanup(gazetteID)

        return nil
}

func (r *JobRunner) Pause() error {
        r.mu.Lock()

        if r.status != StatusRunning && r.status != StatusFailed {
                r.mu.Unlock()
                return errors.New("runner is not running")
        }

        r.mu.Unlock()

        r.pauseCh <- struct{}{}

        return nil
}

func (r *JobRunner) Resume() error {
        r.mu.Lock()

        if r.status != StatusPaused {
                r.mu.Unlock()
                return errors.New("runner is not paused")
        }

        r.mu.Unlock()

        r.resumeCh <- struct{}{}

        return nil
}

func (r *JobRunner) Start() error {
        r.mu.Lock()

        if r.status != StatusStopped {
                r.mu.Unlock()
                return errors.New("runner is not stopped")
        }

        if len(r.order) == 0 {
                r.mu.Unlock()
                return errors.New("nothing to run")
        }

        r.status = StatusRunning
        r.mu.Unlock()

        r.startRun()

        return nil
}

func (r *JobRunner) Stop() error {
        r.mu.Lock()
        if r.status == StatusStopped {
                r.mu.Unlock()
                return fmt.Errorf("runner is already stopped")
        }

        cancel := r.cancel
        r.cancel = nil
        isRunning := r.runningID != 0
        r.mu.Unlock()

        if isRunning && cancel != nil {
                cancel()
        }

        return nil
}

func (r *JobRunner) Wait() {
        r.wg.Wait()
}

func (r *JobRunner) SetOrder(order []int32) error {
        // copy to have owned backing array and not rely on descriptor of caller's array
        // (not modified by single call from handler in this particular case, just good practice)

        r.mu.RLock()
        rOrder := append([]int32(nil), r.filterDeleted(r.order)...)
        r.mu.RUnlock()

        if len(order) != len(rOrder) {
                return fmt.Errorf("input order length differs from internal")
        }

        m := make(map[int32]int)

        for _, v := range order {
                m[v]++
        }

        for _, v := range rOrder {
                if m[v] == 0 {
                        return fmt.Errorf("input order values differ from internal")
                }
                m[v]--
        }

        r.mu.Lock()
        r.order = order
        r.mu.Unlock()

        r.broadcast(EventOrder)

        return nil
}

func (r *JobRunner) SetPatentDeadline(days int, months int, years int) error {
        dl := deadline{days, months, years}

        r.mu.Lock()
        r.patDeadline = dl
        r.mu.Unlock()

        r.broadcast(EventFull)

        return nil
}

func (r *JobRunner) startRun() {
        r.wg.Add(1)
        go r.run()
}

func (r *JobRunner) run() {
        defer r.wg.Done()

        r.mu.Lock()
        if len(r.order) < 1 {
                r.status = StatusStopped
                r.mu.Unlock()
                r.broadcast(EventFull)

                return
        }

        ctx, cancel := context.WithCancel(r.rootCtx)
        gazetteID := r.order[0]
        r.runningID = gazetteID
        r.cancel = cancel
        r.mu.Unlock()

        var wg sync.WaitGroup

        defer func() {
                cancel()
                wg.Wait()
                r.mu.Lock()
                r.runningID = 0
                r.mu.Unlock()

                if r.status != StatusStopped {
                        r.rerun()
                } else {
                        r.broadcast(EventFull)
                }
        }()

        for {
                select {
                case <-r.pauseCh:
                        r.mu.Lock()
                        r.status = StatusPaused
                        r.mu.Unlock()

                        r.broadcast(EventStatus)
                        select {
                        case <-r.resumeCh:
                                r.mu.Lock()
                                r.status = StatusRunning
                                r.mu.Unlock()

                                r.broadcast(EventStatus)
                        case <-ctx.Done():
                                r.mu.Lock()
                                r.status = StatusStopped
                                r.mu.Unlock()

                                return
                        }
                case <-ctx.Done():
                        r.mu.Lock()
                        r.status = StatusStopped
                        r.mu.Unlock()

                        return
                default:
                }

                r.mu.Lock()
                if len(r.order) < 1 {
                        r.status = StatusStopped
                        r.mu.Unlock()
                        break
                }

                if gazetteID != r.order[0] {
                        gazetteID = r.order[0]
                        r.runningID = gazetteID
                        // broadcast not needed, SetOrder() broadcasts
                }

                // Delete() adds job's GazetteID to deleteJobs map and launches cleanup() in goroutine
                if _, ok := r.deleteJobs[gazetteID]; ok {
                        for r.jobs[gazetteID] != nil {
                                r.cond.Wait()
                        }
                        r.mu.Unlock()
                        continue
                }
                r.mu.Unlock()

                select {
                case r.sem <- struct{}{}:
                case <-r.pauseCh:
                        r.pauseCh <- struct{}{}
                        continue
                case <-ctx.Done():
                        continue
                }
                // patUrl status change in db (0 -> 1) happens here
                dbCtx, dbCancel := context.WithTimeout(context.Background(), 10*time.Second)
                patUrl, err := r.getUnprocessedPatentUrlByGazette(dbCtx, gazetteID)
                dbCancel()
                if err != nil {
                        <-r.sem
                        r.mu.Lock()
                        r.status = StatusStopped
                        r.msg = fmt.Sprintf("failed to get patent_url from db: %s", err.Error())
                        r.mu.Unlock()
                        r.broadcast(EventFull)

                        return
                }
                if patUrl == (types.PatentUrl{}) {
                        <-r.sem
                        r.cleanup(gazetteID)

                        continue
                }

                proxy, err := r.proxyPool.GetRandom()
                if err != nil {
                        <-r.sem
                        r.mu.Lock()
                        r.status = StatusStopped
                        r.msg = fmt.Sprintf("failed to get proxy: %v", err)
                        r.mu.Unlock()
                        r.broadcast(EventFull)

                        return
                }

                grpcReq := &pbmodel.PatentURL{
                        Url:   patUrl.URL,
                        Proxy: proxy,
                }

                wg.Add(1)

                r.mu.Lock()
                r.currentPats[gazetteID] = append(r.currentPats[gazetteID], patUrl.Name)
                r.mu.Unlock()
                r.broadcast(EventProgress)

                go func(ctx context.Context, grpcReq *pbmodel.PatentURL, patUrl types.PatentUrl, gazetteID int32) {
                        workerCtx, workerCancel := context.WithTimeout(ctx, 600*time.Second)

                        defer func() {
                                <-r.sem
                                wg.Done()
                                r.removeFromCurrent(gazetteID, patUrl.Name)
                                r.broadcast(EventProgress)
                                workerCancel()
                        }()

                        // TODO: Log here
                        pauseWithMessage := func(msg string) {
                                dbCtx, dbCancel := context.WithTimeout(context.Background(), 10*time.Second)
                                // errP to not shadow captured err variable
                                errP := r.queries.SetPatentUrlStatus(dbCtx, sqlcgen.SetPatentUrlStatusParams{
                                        ID:     patUrl.ID,
                                        Status: 0,
                                })
                                dbCancel()
                                if errP != nil {
                                        msg = msg + fmt.Sprintf("; failed to reset status: %v", err)
                                }

                                select {
                                case r.pauseCh <- struct{}{}:
                                        r.mu.Lock()
                                        r.msg = msg
                                        r.mu.Unlock()
                                default:
                                }
                        }

                        stream, err := r.grpc.GetPatent(workerCtx, grpcReq)
                        if err != nil {
                                metrics.GrpcClientConnFailed.WithLabelValues("GetPatent").Inc()

                                pauseWithMessage(fmt.Sprintf("failed to connect to scraper: %v", err))

                                return
                        }
                        for {
                                event, err := stream.Recv()
                                if err == io.EOF {
                                        break
                                }
                                if err != nil {
                                        pauseWithMessage(fmt.Sprintf("failed to read stream: %v", err))

                                        return
                                }

                                switch payload := event.Payload.(type) {
                                case *pbmodel.PatentEvent_Progress:
                                        // r.mu.Lock()
                                        // if job, ok := r.jobs[gazetteID]; ok {
                                        //         job.Msg = payload.Progress.Stage
                                        // }
                                        // r.mu.Unlock()
                                case *pbmodel.PatentEvent_Package:
                                        pkg := payload.Package

                                        patent, err := mapper.PatentFromProtoPackage(pkg)
                                        if err != nil {
                                                pauseWithMessage(fmt.Sprintf("failed to convert package: %v", err))

                                                return
                                        }

                                        // only pkg validation for these 2 are nil checks,
                                        // which was already checked when converting patent
                                        applicant, _ := mapper.ApplicantFromProtoPackage(pkg)
                                        agents, _ := mapper.AgentsFromProtoPackage(pkg)

                                        dbCtx, dbCancel := context.WithTimeout(context.Background(), 10*time.Second)
                                        err = r.storeProcessedPatentFn(
                                                dbCtx,
                                                patUrl,
                                                patent,
                                                applicant,
                                                agents,
                                        )
                                        dbCancel()
                                        if err != nil {
                                                pauseWithMessage(fmt.Sprintf("failed to store processed patent: %v", err))

                                                return
                                        }

                                        r.mu.Lock()
                                        if job, ok := r.jobs[gazetteID]; ok {
                                                r.msg = "Finished " + patUrl.Name
                                                job.Completed++
                                        }
                                        r.mu.Unlock()

                                        metrics.JobsTotal.WithLabelValues(strconv.Itoa(int(gazetteID)), "success").Inc()

                                        r.broadcast(EventProgress)
                                case *pbmodel.PatentEvent_Error:
                                        if payload.Error.Code == "JOB_MISSED" {
                                                dbCtx, dbCancel := context.WithTimeout(context.Background(), 10*time.Second)
                                                err = r.markPatentUrlFailedFn(dbCtx, &patUrl)
                                                dbCancel()

                                                if err != nil {
                                                        pauseWithMessage(fmt.Sprintf("failed to increment attempts and reset status: %v", err))

                                                        return
                                                }

                                                if patUrl.Attempts >= patUrl.MaxAttempts {
                                                        r.mu.Lock()
                                                        if job, ok := r.jobs[gazetteID]; ok {
                                                                job.Failed++
                                                                r.msg = "Failed " + patUrl.Name
                                                        }
                                                        r.mu.Unlock()

                                                        dbCtx, dbCancel := context.WithTimeout(context.Background(), 10*time.Second)
                                                        err = r.queries.SetPatentUrlFailed(dbCtx, patUrl.ID)
                                                        dbCancel()
                                                        if err != nil {
                                                                // TODO: replace this type of calls with abstract error for client, log the error instead
                                                                pauseWithMessage(fmt.Sprintf("failed to mark processed patent as failed: %v", err))

                                                                return
                                                        }

                                                        metrics.JobsTotal.WithLabelValues(strconv.Itoa(int(gazetteID)), "error").Inc()

                                                        r.broadcast(EventProgress)
                                                } else {
                                                        r.mu.Lock()
                                                        r.msg = "Miss on " + patUrl.Name + ". Will retry"
                                                        r.mu.Unlock()
                                                        metrics.JobsTotal.WithLabelValues(strconv.Itoa(int(gazetteID)), "missed").Inc()
                                                        r.broadcast(EventProgress)
                                                }
                                        }

                                        if payload.Error.Code == "JOB_FAILED" {
                                                dbCtx, dbCancel := context.WithTimeout(context.Background(), 10*time.Second)
                                                err = r.queries.SetPatentUrlFailed(dbCtx, patUrl.ID)
                                                dbCancel()
                                                if err != nil {
                                                        pauseWithMessage(fmt.Sprintf("failed to mark processed patent as failed: %v", err))

                                                        return
                                                }

                                                r.mu.Lock()
                                                if job, ok := r.jobs[gazetteID]; ok {
                                                        job.Failed++
                                                        r.msg = "Failed " + patUrl.Name
                                                }
                                                r.mu.Unlock()

                                                metrics.JobsTotal.WithLabelValues(strconv.Itoa(int(gazetteID)), "error").Inc()

                                                r.broadcast(EventProgress)
                                        }

                                        if payload.Error.Code == "INVALID_REQUEST" {
                                                pauseWithMessage("invalid data supplied to scraper")

                                                return
                                        }
                                }
                        }
                }(ctx, grpcReq, patUrl, gazetteID)
        }
}

func (r *JobRunner) removeFromCurrent(gazetteID int32, name string) {
        r.mu.Lock()
        defer r.mu.Unlock()

        for i, v := range r.currentPats[gazetteID] {
                if v == name {
                        r.currentPats[gazetteID] = append(r.currentPats[gazetteID][:i], r.currentPats[gazetteID][i+1:]...)
                        break
                }
        }

        r.cond.Broadcast()
}

func (r *JobRunner) cleanup(gazetteID int32) {
        r.mu.Lock()
        for len(r.currentPats[gazetteID]) > 0 {
                r.cond.Wait() // atomically releases mu, re-acquires on wake
        }

        r.order = filterOrder(r.order, gazetteID)
        metrics.JobQueueDepth.Dec()

        delete(r.jobs, gazetteID)
        delete(r.currentPats, gazetteID)
        delete(r.deleteJobs, gazetteID)
        r.mu.Unlock()

        r.cond.Broadcast() // for run() waiting on running job marked deleted
        r.broadcast(EventFull)
}

func (r *JobRunner) rerun() {
        r.mu.Lock()
        if len(r.order) > 0 {
                r.status = StatusRunning
                r.mu.Unlock()
                r.startRun()
        } else {
                r.mu.Unlock()
        }
}

func filterOrder(order []int32, exclude int32) []int32 {
        out := make([]int32, 0, len(order))
        for _, id := range order {
                if id != exclude {
                        out = append(out, id)
                }
        }

        return out
}