diff --git a/.vscode/launch.json b/.vscode/launch.json index 4cd7ee9..8358ffb 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -30,6 +30,19 @@ "showLog": true, "console": "integratedTerminal" }, + { + "name": "Launch API Service", + "type": "go", + "request": "launch", + "mode": "debug", + "program": "${workspaceFolder}/api/main.go", + "env": { + "GO_ENV": "development" + }, + "args": [], + "showLog": true, + "console": "integratedTerminal" + }, { "name": "Launch Current File", "type": "go", diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..038411a --- /dev/null +++ b/Makefile @@ -0,0 +1,48 @@ +# Jobs Scraper API Makefile + +# Variables +APP_NAME=jobs-scraper +API_DIR=./api +DOCS_DIR=./docs +BIN_DIR=./bin +MAIN_FILE=$(API_DIR)/main.go + +# Default target + +# NATS server with JetStream +nats-server: + @echo "Starting NATS server with JetStream..." + docker run --rm -p 4222:4222 -p 8222:8222 nats:latest -js + +.DEFAULT_GOAL := run + +.PHONY: run +run: ## Run the application + @echo "Running $(APP_NAME)..." + @go run $(MAIN_FILE) + +.PHONY: clean +clean: ## Clean build artifacts + @echo "Cleaning build artifacts..." + @rm -rf $(BIN_DIR) + @echo "Clean complete" + +# Swagger documentation +.PHONY: swagger +swagger: ## Generate Swagger documentation + @echo "Generating Swagger documentation..." + @go run github.com/swaggo/swag/cmd/swag@latest init -g $(MAIN_FILE) -o $(DOCS_DIR) + @echo "Swagger documentation generated in $(DOCS_DIR)/" + +# Docker operations (if needed in the future) +.PHONY: docker-build +docker-build: ## Build Docker image + @echo "Building Docker image..." + @docker build -t $(APP_NAME) . + @echo "Docker image built: $(APP_NAME)" + +.PHONY: docker-run +docker-run: ## Run Docker container + @echo "Running Docker container..." + @docker run -p 8080:8080 $(APP_NAME) + diff --git a/api/app.go b/api/app/app.go similarity index 53% rename from api/app.go rename to api/app/app.go index edf52e4..8ceb326 100644 --- a/api/app.go +++ b/api/app/app.go @@ -1,25 +1,32 @@ -package main +package app import ( "database/sql" "github.com/gorilla/mux" "github.com/jobs-scraper/api/commands/job" - "github.com/jobs-scraper/application" + "github.com/jobs-scraper/internal/ports" "github.com/jobs-scraper/internal/repo" "github.com/jobs-scraper/internal/server" ) -func NewApplication(router *mux.Router, db *sql.DB) *application.Application { +type Application struct { + Router *mux.Router + DB *sql.DB + Server *server.AppServer + JobCommands *ports.JobCommands +} + +func NewApplication(router *mux.Router, db *sql.DB) *Application { s := server.NewServer(router) jobRepo := repo.NewJobRepository(db) - return &application.Application{ + return &Application{ Server: s, Router: router, DB: db, - JobCommands: &application.JobCommands{ + JobCommands: &ports.JobCommands{ CreateJob: job.NewCreateJobHandler(jobRepo), }, } diff --git a/api/main.go b/api/main.go index 7905807..efa9a62 100644 --- a/api/main.go +++ b/api/main.go @@ -1,5 +1,120 @@ package main -func main() { +import ( + "context" + "encoding/json" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + "github.com/gorilla/mux" + "github.com/jobs-scraper/api/app" + _ "github.com/jobs-scraper/docs" + "github.com/jobs-scraper/infrastructure" + httpHandler "github.com/jobs-scraper/infrastructure/http" + "github.com/jobs-scraper/infrastructure/nats" + "github.com/jobs-scraper/internal/domain" + "github.com/joho/godotenv" + httpSwagger "github.com/swaggo/http-swagger" +) + +// @title Jobs Scraper API +// @version 1.0 +// @description API for managing job scraping operations +// @BasePath / +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") + + nc, err := nats.NewNatsClient() + + if err != nil { + log.Fatal("Failed to create nats client") + } + + defer nc.Close() + + // Run database migrations + if err := infrastructure.RunMigrations(db); err != nil { + log.Fatalf("Failed to run migrations: %v", err) + } + + router := mux.NewRouter() + + // Swagger endpoint + router.PathPrefix("/swagger/").Handler(httpSwagger.WrapHandler) + + router.Use(httpHandler.CORSMiddleware) + router.Use(httpHandler.LogsMiddleware) + + app := app.NewApplication(router, db) + + jobHandler := httpHandler.NewJobHandler(app.JobCommands) + + jobHandler.RegisterRoutes(router) + + jobRequest := domain.SearchQuery{ + Keywords: "Javascript", + Location: "US", + FWT: "2,3", + } + + // Marshal to JSON with error handling + jsonData, err := json.Marshal(jobRequest) + if err != nil { + log.Printf("Error marshaling JSON: %v", err) + } else { + // Publish with error handling + if err := nc.Publish(nats.LinkedInSubTopic, jsonData); err != nil { + log.Printf("Error publishing message: %v", err) + } else { + log.Printf("Published job request: %s", string(jsonData)) + } + } + + go func() { + log.Printf("Server running on http %s\n", "8080") + if err := app.Server.Start(); err != nil && err != http.ErrServerClosed { + log.Fatalf("Could not listen on http %s: %v\n", "8080", err) + } + }() + + stopChan := make(chan os.Signal, 1) + signal.Notify( + stopChan, + os.Interrupt, + syscall.SIGTERM, + ) + + <-stopChan + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + app.Server.Stop(ctx) + + log.Println("Server shutting down") + os.Exit(0) } diff --git a/application/app.go b/application/app.go deleted file mode 100644 index 88cb41a..0000000 --- a/application/app.go +++ /dev/null @@ -1,20 +0,0 @@ -package application - -import ( - "database/sql" - - "github.com/gorilla/mux" - "github.com/jobs-scraper/internal/ports" - "github.com/jobs-scraper/internal/server" -) - -type Application struct { - Router *mux.Router - DB *sql.DB - Server *server.AppServer - JobCommands *JobCommands -} - -type JobCommands struct { - CreateJob ports.JobCommandHandler -} diff --git a/bin/jobs-scraper b/bin/jobs-scraper new file mode 100755 index 0000000..cca6de9 Binary files /dev/null and b/bin/jobs-scraper differ diff --git a/bin/scraper b/bin/scraper new file mode 100755 index 0000000..ec09b4f Binary files /dev/null and b/bin/scraper differ diff --git a/cv/main.go b/cv/main.go index 17f014d..0b139b8 100644 --- a/cv/main.go +++ b/cv/main.go @@ -36,13 +36,6 @@ func main() { log.Println("Successfully connected to db") - // Run database migrations - if err := infrastructure.RunMigrations(db); err != nil { - log.Fatalf("Failed to run migrations: %v", err) - } - - log.Println("Database migrations completed successfully") - jobRepo := repo.NewJobRepository(db) jobDescriptionRepo := repo.NewJobDescriptionRepository(db) openRouterService := services.NewOpenRouterService(model, apiKey) diff --git a/docs/docs.go b/docs/docs.go new file mode 100644 index 0000000..eec547e --- /dev/null +++ b/docs/docs.go @@ -0,0 +1,130 @@ +// Package docs Code generated by swaggo/swag. DO NOT EDIT +package docs + +import "github.com/swaggo/swag" + +const docTemplate = `{ + "schemes": {{ marshal .Schemes }}, + "swagger": "2.0", + "info": { + "description": "{{escape .Description}}", + "title": "{{.Title}}", + "contact": {}, + "version": "{{.Version}}" + }, + "host": "{{.Host}}", + "basePath": "{{.BasePath}}", + "paths": { + "/jobs": { + "post": { + "description": "Creates a new job with the provided specifications", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "jobs" + ], + "summary": "Create a new job", + "parameters": [ + { + "description": "Job creation data", + "name": "job", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/ports.CreateJobCommand" + } + } + ], + "responses": { + "201": { + "description": "Successfully created job", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "400": { + "description": "Invalid request data", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "500": { + "description": "Internal server error", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + } + }, + "definitions": { + "domain.JobProvider": { + "type": "integer", + "enum": [ + 0, + 1, + 2, + 3, + 4, + 5 + ], + "x-enum-varnames": [ + "LinkedIn", + "Indeed", + "Glassdoor", + "Bayt", + "TokyoDev", + "JapanDev" + ] + }, + "ports.CreateJobCommand": { + "type": "object", + "properties": { + "fwt": { + "type": "string" + }, + "keywords": { + "type": "string" + }, + "location": { + "type": "string" + }, + "provider": { + "$ref": "#/definitions/domain.JobProvider" + } + } + } + } +}` + +// SwaggerInfo holds exported Swagger Info so clients can modify it +var SwaggerInfo = &swag.Spec{ + Version: "1.0", + Host: "", + BasePath: "/", + Schemes: []string{}, + Title: "Jobs Scraper API", + Description: "API for managing job scraping operations", + InfoInstanceName: "swagger", + SwaggerTemplate: docTemplate, + LeftDelim: "{{", + RightDelim: "}}", +} + +func init() { + swag.Register(SwaggerInfo.InstanceName(), SwaggerInfo) +} diff --git a/docs/swagger.json b/docs/swagger.json new file mode 100644 index 0000000..98d21fb --- /dev/null +++ b/docs/swagger.json @@ -0,0 +1,105 @@ +{ + "swagger": "2.0", + "info": { + "description": "API for managing job scraping operations", + "title": "Jobs Scraper API", + "contact": {}, + "version": "1.0" + }, + "basePath": "/", + "paths": { + "/jobs": { + "post": { + "description": "Creates a new job with the provided specifications", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "jobs" + ], + "summary": "Create a new job", + "parameters": [ + { + "description": "Job creation data", + "name": "job", + "in": "body", + "required": true, + "schema": { + "$ref": "#/definitions/ports.CreateJobCommand" + } + } + ], + "responses": { + "201": { + "description": "Successfully created job", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "400": { + "description": "Invalid request data", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "500": { + "description": "Internal server error", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + } + }, + "definitions": { + "domain.JobProvider": { + "type": "integer", + "enum": [ + 0, + 1, + 2, + 3, + 4, + 5 + ], + "x-enum-varnames": [ + "LinkedIn", + "Indeed", + "Glassdoor", + "Bayt", + "TokyoDev", + "JapanDev" + ] + }, + "ports.CreateJobCommand": { + "type": "object", + "properties": { + "fwt": { + "type": "string" + }, + "keywords": { + "type": "string" + }, + "location": { + "type": "string" + }, + "provider": { + "$ref": "#/definitions/domain.JobProvider" + } + } + } + } +} \ No newline at end of file diff --git a/docs/swagger.yaml b/docs/swagger.yaml new file mode 100644 index 0000000..ecf9833 --- /dev/null +++ b/docs/swagger.yaml @@ -0,0 +1,72 @@ +basePath: / +definitions: + domain.JobProvider: + enum: + - 0 + - 1 + - 2 + - 3 + - 4 + - 5 + type: integer + x-enum-varnames: + - LinkedIn + - Indeed + - Glassdoor + - Bayt + - TokyoDev + - JapanDev + ports.CreateJobCommand: + properties: + fwt: + type: string + keywords: + type: string + location: + type: string + provider: + $ref: '#/definitions/domain.JobProvider' + type: object +info: + contact: {} + description: API for managing job scraping operations + title: Jobs Scraper API + version: "1.0" +paths: + /jobs: + post: + consumes: + - application/json + description: Creates a new job with the provided specifications + parameters: + - description: Job creation data + in: body + name: job + required: true + schema: + $ref: '#/definitions/ports.CreateJobCommand' + produces: + - application/json + responses: + "201": + description: Successfully created job + schema: + additionalProperties: + type: string + type: object + "400": + description: Invalid request data + schema: + additionalProperties: + type: string + type: object + "500": + description: Internal server error + schema: + additionalProperties: + type: string + type: object + summary: Create a new job + tags: + - jobs +swagger: "2.0" diff --git a/go.mod b/go.mod index 940aefb..e403e97 100644 --- a/go.mod +++ b/go.mod @@ -11,12 +11,33 @@ require ( github.com/gorilla/mux v1.8.1 github.com/joho/godotenv v1.5.1 github.com/lib/pq v1.10.9 + github.com/nats-io/nats.go v1.46.1 + github.com/swaggo/http-swagger v1.3.4 + github.com/swaggo/swag v1.16.3 +) + +require ( + github.com/KyleBanks/depth v1.2.1 // indirect + github.com/go-openapi/jsonpointer v0.19.5 // indirect + github.com/go-openapi/jsonreference v0.20.0 // indirect + github.com/go-openapi/spec v0.20.6 // indirect + github.com/go-openapi/swag v0.19.15 // indirect + github.com/josharian/intern v1.0.0 // indirect + github.com/mailru/easyjson v0.7.6 // indirect + github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe // indirect + golang.org/x/tools v0.24.0 // indirect + gopkg.in/yaml.v2 v2.4.0 // indirect ) require ( github.com/andybalholm/cascadia v1.3.3 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-multierror v1.1.1 // indirect + github.com/klauspost/compress v1.18.0 // indirect + github.com/nats-io/nkeys v0.4.11 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/orsinium-labs/enum v1.4.0 // indirect + golang.org/x/crypto v0.42.0 // indirect golang.org/x/net v0.44.0 // indirect + golang.org/x/sys v0.36.0 // indirect ) diff --git a/go.sum b/go.sum index 72c8d8f..28192fe 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0= github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc= +github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/PuerkitoBio/goquery v1.10.3 h1:pFYcNSqHxBD06Fpj/KsbStFRsgRATgnf3LeXiUkhzPo= @@ -10,6 +12,8 @@ github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dhui/dktest v0.4.6 h1:+DPKyScKSEp3VLtbMDHcUq6V5Lm5zfZZVb0Sk7Ahom4= @@ -30,6 +34,16 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg= +github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY= +github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg= +github.com/go-openapi/jsonreference v0.20.0 h1:MYlu0sBgChmCfJxxUKZ8g1cPWFOB37YSZqewK7OKeyA= +github.com/go-openapi/jsonreference v0.20.0/go.mod h1:Ag74Ico3lPc+zR+qjn4XBUmXymS4zJbYVCZmcgkasdo= +github.com/go-openapi/spec v0.20.6 h1:ich1RQ3WDbfoeTqTAb+5EIxNmpKVJZWBNah9RAT0jIQ= +github.com/go-openapi/spec v0.20.6/go.mod h1:2OpW+JddWPrpXSCIX8eOx7lZ5iyuWj3RYR6VaaBKcWA= +github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk= +github.com/go-openapi/swag v0.19.15 h1:D2NRCBzS9/pEY3gP9Nl8aDqGUcPFrwG2p+CNFrLyrCM= +github.com/go-openapi/swag v0.19.15/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang-migrate/migrate/v4 v4.19.0 h1:RcjOnCGz3Or6HQYEJ/EEVLfWnmw9KnoigPSjzhCuaSE= @@ -44,8 +58,21 @@ github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+l github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= +github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc= +github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc= +github.com/mailru/easyjson v0.7.6 h1:8yTIVnZgCoiM1TgqoeTl+LfU5Jg6/xL3QhGQnimLYnA= +github.com/mailru/easyjson v0.7.6/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= github.com/matryer/is v1.4.1 h1:55ehd8zaGABKLXQUe2awZ99BD/PTc2ls+KV/dXphgEQ= github.com/matryer/is v1.4.1/go.mod h1:8I/i5uYgLzgsgEloJE1U6xx5HkBQpAZvepWuujKwMRU= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= @@ -54,6 +81,14 @@ github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0= github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/nats-io/nats.go v1.46.1 h1:bqQ2ZcxVd2lpYI97xYASeRTY3I5boe/IVmuUDPitHfo= +github.com/nats-io/nats.go v1.46.1/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g= +github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0= +github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs= +github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug= @@ -64,8 +99,17 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe h1:K8pHPVoTgxFJt1lXuIzzOX7zZhZFldJQK/CgKx9BFIc= +github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe/go.mod h1:lKJPbtWzJ9JhsTN1k1gZgleJWY/cqq0psdoMmaThG3w= +github.com/swaggo/http-swagger v1.3.4 h1:q7t/XLx0n15H1Q9/tk3Y9L4n210XzJF5WtnDX64a5ww= +github.com/swaggo/http-swagger v1.3.4/go.mod h1:9dAh0unqMBAlbp1uE2Uc2mQTxNMU/ha4UbucIg1MFkQ= +github.com/swaggo/swag v1.16.3 h1:PnCYjPCah8FK4I26l2F/KQ4yz3sILcVUN3cTlBFA9Pg= +github.com/swaggo/swag v1.16.3/go.mod h1:DImHIuOFXKpMFAQjcC7FG4m3Dg4+QuUgUzJmKjI/gRk= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= @@ -83,13 +127,18 @@ golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliY golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8= golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= +golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI= +golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/mod v0.21.0 h1:vvrHzRwRfVKSiLrG+d4FMl/Qi4ukBCE6kZlTUkDYRT0= +golang.org/x/mod v0.21.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= @@ -106,8 +155,11 @@ golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw= +golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -130,6 +182,7 @@ golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY= golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= @@ -143,6 +196,17 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58= golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= +golang.org/x/tools v0.24.0 h1:J1shsA93PJUEVaUSaay7UXAyE8aimq3GW0pjlolpa24= +golang.org/x/tools v0.24.0/go.mod h1:YhNqVBIfWHdzvTLs0d8LCuMhkKUgSUKldakyV7W/WDQ= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU= +gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= +gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/infrastructure/http/job.go b/infrastructure/http/job.go new file mode 100644 index 0000000..cccbe5a --- /dev/null +++ b/infrastructure/http/job.go @@ -0,0 +1,65 @@ +package http + +import ( + "encoding/json" + "log/slog" + "net/http" + + "github.com/jobs-scraper/internal/ports" + "github.com/jobs-scraper/internal/utils" + + "github.com/gorilla/mux" +) + +// JobHandler handles HTTP requests related to cars +type JobHandler struct { + jobCommands *ports.JobCommands +} + +// NewVehicleHandler creates a new instance of VehicleHandler +func NewJobHandler(jobCommands *ports.JobCommands) *JobHandler { + return &JobHandler{ + jobCommands: jobCommands, + } +} + +// RegisterRoutes registers all routes to the router +func (h *JobHandler) RegisterRoutes(router *mux.Router) { + vehicles := router.PathPrefix("/jobs").Subrouter() + + //vehicles.Use(AuthMiddleware) + + vehicles.HandleFunc("", utils.Make(h.CreateJob)).Methods("POST") + // vehicles.HandleFunc("", utils.Make(h.ListVehicles)).Methods("GET") + // vehicles.HandleFunc("/{id}", utils.Make(h.GetVehicle)).Methods("GET") + // vehicles.HandleFunc("/buy", utils.Make(h.BuyVehicle)).Methods("POST") + // vehicles.HandleFunc("/sell", utils.Make(h.SellVehicle)).Methods("POST") +} + +// CreateJob handles the creation of a new job +// @Summary Create a new job +// @Description Creates a new job with the provided specifications +// @Tags jobs +// @Accept json +// @Produce json +// @Param job body ports.CreateJobCommand true "Job creation data" +// @Success 201 {object} map[string]string "Successfully created job" +// @Failure 400 {object} map[string]string "Invalid request data" +// @Failure 500 {object} map[string]string "Internal server error" +// @Router /jobs [post] +func (h *JobHandler) CreateJob(w http.ResponseWriter, r *http.Request) error { + var cmd ports.CreateJobCommand + if err := json.NewDecoder(r.Body).Decode(&cmd); err != nil { + slog.Error(err.Error()) + return utils.NewAPIError(http.StatusBadRequest, utils.InvalidJSON()) + } + + err := h.jobCommands.CreateJob.Handle(cmd) + if err != nil { + slog.Error(err.Error()) + return utils.NewAPIError(http.StatusInternalServerError, err) + } + + utils.WriteJSON(w, http.StatusCreated, map[string]string{"message": "Success"}) + return nil +} diff --git a/infrastructure/http/middlewares.go b/infrastructure/http/middlewares.go new file mode 100644 index 0000000..3b6e0da --- /dev/null +++ b/infrastructure/http/middlewares.go @@ -0,0 +1,143 @@ +package http + +import ( + "context" + "encoding/json" + "log" + "net/http" + "strings" + "time" +) + +// CORSMiddleware handles Cross-Origin Resource Sharing +func CORSMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Set CORS headers + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With") + w.Header().Set("Access-Control-Allow-Credentials", "true") + w.Header().Set("Access-Control-Max-Age", "86400") + + // Handle preflight OPTIONS requests + if r.Method == "OPTIONS" { + w.WriteHeader(http.StatusOK) + return + } + + next.ServeHTTP(w, r) + }) +} + +// LogsMiddleware logs HTTP requests with timing information +func LogsMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + next.ServeHTTP(w, r) + log.Printf("%s %s %s", r.Method, r.RequestURI, time.Since(start)) + }) +} + +// AuthMiddleware handles JWT token validation, API keys, and basic auth +func AuthMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Skip auth for swagger endpoints and health checks + if strings.HasPrefix(r.URL.Path, "/swagger/") || r.URL.Path == "/health" { + next.ServeHTTP(w, r) + return + } + + // Try Bearer token authentication first + authHeader := r.Header.Get("Authorization") + if authHeader != "" { + if strings.HasPrefix(authHeader, "Bearer ") { + token := strings.TrimPrefix(authHeader, "Bearer ") + if token != "" { + userID, err := validateJWTToken(token) + if err == nil { + ctx := context.WithValue(r.Context(), "user_id", userID) + next.ServeHTTP(w, r.WithContext(ctx)) + return + } + } + } + } + + // Try API Key authentication + apiKey := r.Header.Get("X-API-Key") + if apiKey != "" { + if isValidAPIKey(apiKey) { + ctx := context.WithValue(r.Context(), "user_id", getUserIDFromAPIKey(apiKey)) + next.ServeHTTP(w, r.WithContext(ctx)) + return + } + } + + // Try Basic authentication + username, password, ok := r.BasicAuth() + if ok { + if isValidCredentials(username, password) { + ctx := context.WithValue(r.Context(), "username", username) + next.ServeHTTP(w, r.WithContext(ctx)) + return + } + } + + // Fallback to cookie-based auth (existing behavior) + token, err := r.Cookie("token") + if err == nil && token.Value == "123" { + next.ServeHTTP(w, r) + return + } + + // No valid authentication found + response := map[string]interface{}{ + "error": "Unauthorized", + "message": "Valid authentication required. Use Bearer token, API key, or basic auth.", + "code": http.StatusUnauthorized, + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusUnauthorized) + json.NewEncoder(w).Encode(response) + }) +} + +// Helper functions for authentication validation +func validateJWTToken(token string) (string, error) { + // TODO: Implement JWT validation logic with your preferred JWT library + // For demonstration, accept a demo token + if token == "valid-jwt-token" { + return "user-123", nil + } + return "", http.ErrAbortHandler +} + +func isValidAPIKey(apiKey string) bool { + // TODO: Implement API key validation logic + // For demonstration, accept a demo API key + return apiKey == "demo-api-key-123" +} + +func getUserIDFromAPIKey(apiKey string) string { + // TODO: Implement logic to get user ID from API key + // For demonstration, return a mock user ID + return "api-user-123" +} + +func isValidCredentials(username, password string) bool { + // TODO: Implement credential validation logic + // For demonstration, accept demo credentials + return username == "admin" && password == "password" +} + +// GetUserIDFromContext extracts user ID from request context +func GetUserIDFromContext(ctx context.Context) (string, bool) { + userID, ok := ctx.Value("user_id").(string) + return userID, ok +} + +// GetUsernameFromContext extracts username from request context +func GetUsernameFromContext(ctx context.Context) (string, bool) { + username, ok := ctx.Value("username").(string) + return username, ok +} diff --git a/infrastructure/nats/nats.go b/infrastructure/nats/nats.go new file mode 100644 index 0000000..dcb1b03 --- /dev/null +++ b/infrastructure/nats/nats.go @@ -0,0 +1,129 @@ +package nats + +import ( + "fmt" + "log" + "strings" + "time" + + "github.com/nats-io/nats.go" +) + +const ( + ScraperTopic = "scraper" + LinkedInSubTopic = "scraper.linkedin" + IndeedSubTopic = "scraper.indeed" + BaytSubTopic = "scraper.bayt" + TokyoDevSubTopic = "scraper.tokyodev" + JapanDevSubTopic = "scraper.japandev" +) + +type NatsClient struct { + Conn *nats.Conn + JetStream nats.JetStreamContext + StreamName string +} + +func NewNatsClient() (*NatsClient, error) { + // Connect to NATS + nc, err := nats.Connect(nats.DefaultURL, nats.RetryOnFailedConnect(true), + nats.MaxReconnects(5), + nats.ReconnectWait(5*time.Second)) + if err != nil { + return nil, fmt.Errorf("error connecting to NATS: %v", err) + } + + // Create JetStream Context + js, err := nc.JetStream() + if err != nil { + return nil, fmt.Errorf("error getting JetStream context: %v", err) + } + + // Create Stream + streamName := "SCRAPER_STREAM" + _, err = js.StreamInfo(streamName) + if err != nil { + // Stream doesn't exist, let's create it + _, err := js.AddStream(&nats.StreamConfig{ + Name: streamName, + Subjects: []string{ + LinkedInSubTopic, + IndeedSubTopic, + BaytSubTopic, + TokyoDevSubTopic, + JapanDevSubTopic, + }, + Storage: nats.FileStorage, + MaxAge: 24 * time.Hour, + Retention: nats.WorkQueuePolicy, + }) + if err != nil { + return nil, fmt.Errorf("error creating stream: %v", err) + } + log.Printf("Created new stream: %s", streamName) + } + + return &NatsClient{ + Conn: nc, + JetStream: js, + StreamName: streamName, + }, nil +} + +// Publish publishes a message to a specific topic +func (n *NatsClient) Publish(topic string, data []byte) error { + _, err := n.JetStream.Publish(topic, data) + if err != nil { + return fmt.Errorf("error publishing message: %v", err) + } + return nil +} + +// Subscribe creates a subscription to a specific topic with production-ready error handling +func (n *NatsClient) Subscribe(topic, consumerName string, handler nats.MsgHandler) (*nats.Subscription, error) { + // Define subscription options + subscribeOptions := []nats.SubOpt{ + nats.Durable(consumerName), // Durable consumer name + nats.ManualAck(), // Manual acknowledgment + nats.AckExplicit(), // Explicit acknowledgment required + nats.DeliverAll(), // Deliver all messages + } + + // First attempt to subscribe + sub, err := n.JetStream.Subscribe(topic, handler, subscribeOptions...) + if err != nil { + // Check if error is due to consumer already being bound + if strings.Contains(err.Error(), "already bound") || + strings.Contains(err.Error(), "consumer is already bound") { + + log.Printf("Consumer %s is already bound, attempting to delete and recreate", consumerName) + + // Delete the existing consumer + deleteErr := n.JetStream.DeleteConsumer(n.StreamName, consumerName) + if deleteErr != nil { + log.Printf("Warning: Failed to delete existing consumer: %v", deleteErr) + } + + // Retry subscription after deleting consumer + sub, err = n.JetStream.Subscribe(topic, handler, subscribeOptions...) + if err != nil { + return nil, fmt.Errorf("error subscribing to topic after consumer deletion: %v", err) + } + + log.Printf("Successfully recreated consumer %s and subscribed to %s", consumerName, topic) + } else { + return nil, fmt.Errorf("error subscribing to topic: %v", err) + } + } else { + log.Printf("Successfully subscribed to %s using existing consumer %s", topic, consumerName) + } + + return sub, nil +} + +// Close closes the NATS connection +func (n *NatsClient) Close() { + if n.Conn != nil { + n.Conn.Close() + } +} diff --git a/internal/models/Job.go b/internal/models/Job.go deleted file mode 100644 index e45054f..0000000 --- a/internal/models/Job.go +++ /dev/null @@ -1,19 +0,0 @@ -package models - -import "github.com/jobs-scraper/internal/domain" - -type Job struct { - ID int64 - Title string - Company string - CompanyLink string - Location string - JobLink string - Provider domain.JobProvider -} - -type SearchQuery struct { - Keywords string `json:"keywords"` - Location string `json:"location"` - FWT string `json:"f_WT"` // Work type filter (1=onsite, 2=remote, 3=hybrid) -} diff --git a/internal/models/JobDescription.go b/internal/models/JobDescription.go deleted file mode 100644 index 35b106a..0000000 --- a/internal/models/JobDescription.go +++ /dev/null @@ -1,7 +0,0 @@ -package models - -type JobDescription struct { - JobID int64 - Description string - Criteria map[string]string -} diff --git a/internal/models/JobWithDescription.go b/internal/models/JobWithDescription.go deleted file mode 100644 index f819926..0000000 --- a/internal/models/JobWithDescription.go +++ /dev/null @@ -1,6 +0,0 @@ -package models - -type JobWithDescription struct { - Job Job - JobDescription JobDescription -} diff --git a/internal/ports/commands.go b/internal/ports/job-commands.go similarity index 86% rename from internal/ports/commands.go rename to internal/ports/job-commands.go index b188ec9..89c0b1c 100644 --- a/internal/ports/commands.go +++ b/internal/ports/job-commands.go @@ -2,6 +2,10 @@ package ports import "github.com/jobs-scraper/internal/domain" +type JobCommands struct { + CreateJob JobCommandHandler +} + // CreateJobCommand represents the command to create a job type CreateJobCommand struct { Location string diff --git a/internal/utils/http.go b/internal/utils/http.go new file mode 100644 index 0000000..7ba79da --- /dev/null +++ b/internal/utils/http.go @@ -0,0 +1,87 @@ +package utils + +import ( + "encoding/json" + "fmt" + "log/slog" + "net/http" +) + +// APIError represents an API error with status code and message +type APIError struct { + StatusCode int `json:"statusCode"` + Msg any `json:"msg"` +} + +// Error implements the error interface for APIError +func (e APIError) Error() string { + return fmt.Sprintf("api error: %d", e.StatusCode) +} + +// NewAPIError creates a new APIError with the given status code and error message +func NewAPIError(statusCode int, err error) APIError { + return APIError{ + StatusCode: statusCode, + Msg: err.Error(), + } +} + +// InvalidRequestData creates an APIError for validation errors with a map of field errors +func InvalidRequestData(msg string) APIError { + return APIError{ + StatusCode: http.StatusUnprocessableEntity, + Msg: msg, + } +} + +// InvalidJSON creates an APIError for invalid JSON format +func InvalidJSON() APIError { + return NewAPIError(http.StatusBadRequest, fmt.Errorf("invalid JSON request data")) +} + +// APIResponse represents a standardized API response wrapper +type APIResponse struct { + Success bool `json:"success"` + Data any `json:"data,omitempty"` + Error any `json:"error,omitempty"` +} + +// WriteJSON writes the response as JSON with given status code, wrapped in a data object +func WriteJSON(w http.ResponseWriter, status int, data any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + + response := APIResponse{ + Success: status >= 200 && status < 300, + } + + if response.Success { + response.Data = data + } else { + response.Error = data + } + + json.NewEncoder(w).Encode(response) +} + +// APIFunc is the signature for API handler functions +type APIFunc func(w http.ResponseWriter, r *http.Request) error + +// Make wraps an APIFunc and handles errors consistently +func Make(h APIFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if err := h(w, r); err != nil { + if apiErr, ok := err.(APIError); ok { + WriteJSON(w, apiErr.StatusCode, apiErr) + } else { + errResp := map[string]any{ + "statusCode": http.StatusInternalServerError, + "msg": "internal server error", + } + WriteJSON(w, http.StatusInternalServerError, errResp) + // You could add logging here + slog.Error("HTTP API error", "err", err.Error(), "path", r.URL.Path) + } + } + } +} diff --git a/nats-checker.go b/nats-checker.go new file mode 100644 index 0000000..9605831 --- /dev/null +++ b/nats-checker.go @@ -0,0 +1,80 @@ +package main + +import ( + "fmt" + "log" + "time" + + "github.com/nats-io/nats.go" +) + +func main() { + // Connect to NATS server + nc, err := nats.Connect(nats.DefaultURL) + if err != nil { + log.Fatal("Failed to connect to NATS:", err) + } + defer nc.Close() + + // Get JetStream context + js, err := nc.JetStream() + if err != nil { + log.Fatal("Failed to get JetStream context:", err) + } + + // Check if the stream exists + streamName := "SCRAPER_STREAM" + streamInfo, err := js.StreamInfo(streamName) + if err != nil { + log.Fatal("Failed to get stream info:", err) + } + + fmt.Printf("Stream: %s\n", streamInfo.Config.Name) + fmt.Printf("Messages: %d\n", streamInfo.State.Msgs) + fmt.Printf("Subjects: %v\n", streamInfo.Config.Subjects) + fmt.Printf("Storage: %s\n", streamInfo.Config.Storage) + fmt.Printf("Max Age: %s\n", streamInfo.Config.MaxAge) + fmt.Println("---") + + if streamInfo.State.Msgs == 0 { + fmt.Println("No messages found in the stream") + return + } + + // Subscribe to get messages + sub, err := js.PullSubscribe("", "temp-checker", nats.BindStream(streamName)) + if err != nil { + log.Fatal("Failed to create subscription:", err) + } + defer sub.Unsubscribe() + + fmt.Println("Messages in the stream:") + fmt.Println("======================") + + // Fetch messages + msgs, err := sub.Fetch(int(streamInfo.State.Msgs), nats.MaxWait(5*time.Second)) + if err != nil { + log.Printf("Error fetching messages: %v", err) + return + } + + for i, msg := range msgs { + fmt.Printf("Message %d:\n", i+1) + fmt.Printf(" Subject: %s\n", msg.Subject) + fmt.Printf(" Data: %s\n", string(msg.Data)) + + // Get metadata + metadata, err := msg.Metadata() + if err == nil { + fmt.Printf(" Timestamp: %s\n", time.Unix(0, metadata.Timestamp.UnixNano()).Format(time.RFC3339)) + fmt.Printf(" Sequence: %d\n", metadata.Sequence.Stream) + } + fmt.Println(" ---") + + // Acknowledge the message + msg.Ack() + } + + // Clean up the temporary consumer + js.DeleteConsumer(streamName, "temp-checker") +} diff --git a/scraper/main.go b/scraper/main.go index 4fcfc1a..a2adf9a 100644 --- a/scraper/main.go +++ b/scraper/main.go @@ -1,14 +1,23 @@ package main import ( - "context" + // "context" + "encoding/json" "log" - "time" + "os" + "os/signal" + "syscall" + + // "time" "github.com/jobs-scraper/infrastructure" + localNats "github.com/jobs-scraper/infrastructure/nats" "github.com/jobs-scraper/internal/domain" - "github.com/jobs-scraper/internal/pipeline" - "github.com/jobs-scraper/internal/repo" + "github.com/nats-io/nats.go" + + // "github.com/jobs-scraper/internal/domain" + // "github.com/jobs-scraper/internal/pipeline" + // "github.com/jobs-scraper/internal/repo" "github.com/joho/godotenv" ) @@ -34,36 +43,61 @@ func main() { log.Println("Successfully connected to db") - // Run database migrations - if err := infrastructure.RunMigrations(db); err != nil { - log.Fatalf("Failed to run migrations: %v", err) - } - - scraper := pipeline.NewScraper(pipeline.Config{ - SortBy: "R", - MaxRetries: 3, - BaseDelay: 1 * time.Second, - MaxDelay: 30 * time.Second, - RequestTimeout: 30 * time.Second, - }) - - jobRepo := repo.NewJobRepository(db) - jobDescriptionRepo := repo.NewJobDescriptionRepository(db) - - jobPipeline := pipeline.NewJobPipeline(scraper, 5, 1*time.Second) // 5 workers, 1 second rate limit - - ctx := context.Background() - - searchParams := domain.SearchQuery{ - Keywords: "Javascript", - Location: "US", - FWT: "2,3", - } - - err = jobPipeline.ProcessJobsStreaming(ctx, 10, jobRepo, jobDescriptionRepo, searchParams) + nc, err := localNats.NewNatsClient() if err != nil { - log.Fatalf("Pipeline processing failed: %v", err) + log.Fatal("Error connecting to nats") } - log.Println("Jobs inserted successfully") + // scraper := pipeline.NewScraper(pipeline.Config{ + // SortBy: "R", + // MaxRetries: 3, + // BaseDelay: 1 * time.Second, + // MaxDelay: 30 * time.Second, + // RequestTimeout: 30 * time.Second, + // }) + + // jobRepo := repo.NewJobRepository(db) + // jobDescriptionRepo := repo.NewJobDescriptionRepository(db) + + // jobPipeline := pipeline.NewJobPipeline(scraper, 5, 1*time.Second) // 5 workers, 1 second rate limit + + // ctx := context.Background() + + // searchParams := domain.SearchQuery{ + // Keywords: "Javascript", + // Location: "US", + // FWT: "2,3", + // } + + // Subscribe to LinkedIn topic + sub, err := nc.Subscribe(localNats.LinkedInSubTopic, "scraper-consumer", func(msg *nats.Msg) { + var data domain.SearchQuery + err := json.Unmarshal(msg.Data, &data) + + if err != nil { + log.Printf("Error processing message: %v", err) + return + + } + log.Printf("Received message on %s: %s", msg.Subject, string(msg.Data)) + // err = jobPipeline.ProcessJobsStreaming(ctx, 10, jobRepo, jobDescriptionRepo, searchParams) + + msg.Ack() + }) + if err != nil { + log.Fatal("Error subscribing to LinkedIn topic:", err) + } + + log.Printf("Successfully subscribed to %s", localNats.LinkedInSubTopic) + + // Keep the program running to listen for messages + log.Println("Scraper is running. Press Ctrl+C to stop...") + + // Wait for interrupt signal + c := make(chan os.Signal, 1) + signal.Notify(c, os.Interrupt, syscall.SIGTERM) + <-c + sub.Unsubscribe() + nc.Close() + log.Println("Shutting down scraper...") }