[#15] Keyword CSV Upload And Scrape Asynq Worker

Dockerfile

@@ -8,6 +8,7 @@ RUN go mod download
 COPY . .
 
 RUN go build -ldflags '-w -s' -a -o ./bin/api ./cmd/api \
+    && go build -ldflags '-w -s' -a -o ./bin/worker ./cmd/worker \
     && go build -ldflags '-w -s' -a -o ./bin/dbmigrate ./cmd/dbmigrate
 
 EXPOSE 8080

cmd/api/main.go

@@ -6,8 +6,10 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 
+	"github.com/hibiken/asynq"
 	"gorm.io/driver/postgres"
 	"gorm.io/gorm"
 	gormlogger "gorm.io/gorm/logger"
@@ -42,7 +44,15 @@ func main() {
 		l.Fatal().Err(err).Msg("DB connection start failure")
 	}
 
-	r := router.New(c.Server.TimeoutRead, c.Server.TimeoutWrite, db, ml, l, v)
+	redisClusterAddresses := strings.Split(c.RedisHosts, ",")
+	if len(redisClusterAddresses) == 0 {
+		l.Fatal().Msg("Redis cluster connection failure")
+	}
+	asyq := asynq.NewClient(asynq.RedisClusterClientOpt{
+		Addrs: redisClusterAddresses,
+	})
+
+	r := router.New(c.Server.TimeoutRead, c.Server.TimeoutWrite, db, ml, l, v, asyq)
 
 	s := &http.Server{
 		Addr:         fmt.Sprintf(":%d", c.Server.Port),

cmd/worker/main.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+
+	"github.com/hibiken/asynq"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+	gormlogger "gorm.io/gorm/logger"
+
+	"web-scraper.dev/internal/config"
+	"web-scraper.dev/internal/utils/logger"
+	"web-scraper.dev/internal/workers"
+)
+
+const fmtDBString = "host=%s user=%s password=%s dbname=%s port=%d sslmode=disable"
+
+func main() {
+	c := config.NewWorkerConf()
+	l := logger.New(true)
+	db, err := gormDB(&c.DB)
+	if err != nil {
+		l.Fatal().Err(err).Msg("DB connection start failure")
+	}
+
+	redisClusterAddresses := strings.Split(c.RedisHosts, ",")
+	if len(redisClusterAddresses) == 0 {
+		l.Fatal().Msg("Redis cluster connection failure")
+	}
+	redisConnOpt := asynq.RedisClusterClientOpt{
+		Addrs: redisClusterAddresses,
+	}
+
+	scrapeWorker := workers.NewScrapeWorker(redisConnOpt, db, l)
+
+	closed := make(chan struct{})
+	go func() {
+		sigint := make(chan os.Signal, 1)
+		signal.Notify(sigint, os.Interrupt, syscall.SIGTERM)
+		<-sigint
+
+		l.Info().Msg("Shutting down worker")
+
+		if err := scrapeWorker.Stop(); err != nil {
+			l.Error().Err(err).Msg("Server shutdown failure")
+		}
+
+		sqlDB, err := db.DB()
+		if err == nil {
+			if err = sqlDB.Close(); err != nil {
+				l.Error().Err(err).Msg("DB connection closing failure")
+			}
+		}
+
+		close(closed)
+	}()
+
+	l.Info().Msg("Starting worker")
+	if err := scrapeWorker.Start(); err != nil {
+		l.Fatal().Err(err).Msg("Worker start failure")
+	}
+
+	<-closed
+	l.Info().Msgf("Worker shutdown successfully")
+}
+
+func gormDB(conf *config.ConfDB) (*gorm.DB, error) {
+	var logLevel gormlogger.LogLevel
+	if conf.Debug {
+		logLevel = gormlogger.Info
+	} else {
+		logLevel = gormlogger.Error
+	}
+
+	dbString := fmt.Sprintf(fmtDBString, conf.Host, conf.Username, conf.Password, conf.DBName, conf.Port)
+	return gorm.Open(postgres.Open(dbString), &gorm.Config{Logger: gormlogger.Default.LogMode(logLevel)})
+}

compose.yml

@@ -10,9 +10,22 @@ services:
         condition: service_healthy
       mailhog:
         condition: service_started
+      redis:
+        condition: service_started
     command: [ "sh", "-c", "/app/bin/dbmigrate up && /app/bin/api" ]
     restart: always
 
+  worker:
+    build: .
+    env_file: .env
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_started
+    command: [ "/app/bin/worker" ]
+    restart: always
+
   db:
     image: postgres:17-alpine
     environment:
@@ -39,3 +52,20 @@ services:
       - "1025:1025"
       - "8025:8025"
     restart: always
+
+  redis:
+    image: neohq/redis-cluster:latest # for amd64 use grokzen/redis-cluster:latest
+    volumes:
+      - ./redis-cluster.tmpl:/redis-conf/redis-cluster.tmpl
+    environment:
+      # IP: 0.0.0.0 # if needed to connect from a local Go app(not running Docker)
+      MASTERS: 3
+      SLAVES_PER_MASTER: 1
+    ports:
+      - "7000-7005:7000-7005"
+    healthcheck:
+      test: [ "CMD-SHELL", "redis-cli -p 7005 ping" ]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    restart: always

go.mod

@@ -7,8 +7,10 @@ require (
 	github.com/go-chi/chi/v5 v5.2.2
 	github.com/go-chi/cors v1.2.2
 	github.com/go-playground/validator/v10 v10.27.0
+	github.com/gocolly/colly/v2 v2.2.0
 	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/google/uuid v1.6.0
+	github.com/hibiken/asynq v0.25.1
 	github.com/jackc/pgx/v5 v5.7.5
 	github.com/pressly/goose/v3 v3.24.3
 	github.com/rs/xid v1.6.0
@@ -19,22 +21,43 @@ require (
 )
 
 require (
+	github.com/PuerkitoBio/goquery v1.10.3 // indirect
+	github.com/andybalholm/cascadia v1.3.3 // indirect
+	github.com/antchfx/htmlquery v1.3.4 // indirect
+	github.com/antchfx/xmlquery v1.4.4 // indirect
+	github.com/antchfx/xpath v1.3.4 // indirect
+	github.com/bits-and-blooms/bitset v1.22.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/gabriel-vasile/mimetype v1.4.9 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
+	github.com/gobwas/glob v0.2.3 // indirect
+	github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/kennygrant/sanitize v1.2.4 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mfridman/interpolate v0.0.2 // indirect
+	github.com/nlnwa/whatwg-url v0.6.2 // indirect
+	github.com/redis/go-redis/v9 v9.11.0 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
+	github.com/saintfish/chardet v0.0.0-20230101081208-5e3ef4b5456d // indirect
 	github.com/sethvargo/go-retry v0.3.0 // indirect
+	github.com/spf13/cast v1.9.2 // indirect
+	github.com/temoto/robotstxt v1.1.2 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.41.0 // indirect
 	golang.org/x/sync v0.15.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/text v0.26.0 // indirect
+	golang.org/x/time v0.12.0 // indirect
+	google.golang.org/appengine v1.6.8 // indirect
+	google.golang.org/protobuf v1.36.6 // indirect
 )