prehistoric-systems/omnivore
1
0
Fork 0
mirror of https://github.com/omnivore-app/omnivore.git synced 2026-03-11 08:54:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Rename abbreviated identifiers to full descriptive names

Browse source 
Replace short variable/field/parameter names with self-documenting ones:
- rds  → redisDS  (RedisDataSource)
- br   → browser
- cfg  → config
- rdb  → redisClient
- w    → worker  (in New() signatures; http.ResponseWriter stays as w)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Aliaksei Karneyeu 2026-03-04 15:05:26 +01:00
parent 25cbcddf01
commit 1ad2694861
9 changed files with 103 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
packages/content-fetch-go/internal/analytics/analytics.go
View file

@ -11,7 +11,7 @@ import (
// Client sends analytics events to PostHog.
type Client struct {
ph posthog.Client
cfg *config.Config
config *config.Config
}
// Event carries data for an analytics capture call.
@ -24,13 +24,13 @@ type Event struct {
}
// New creates a PostHog analytics client.
func New(cfg *config.Config) *Client {
ph, err := posthog.NewWithConfig(cfg.PostHogAPIKey, posthog.Config{})
func New(config *config.Config) *Client {
ph, err := posthog.NewWithConfig(config.PostHogAPIKey, posthog.Config{})
if err != nil {
log.Printf("Failed to create PostHog client: %v", err)
return &Client{cfg: cfg}
return &Client{config: config}
}
return &Client{ph: ph, cfg: cfg}
return &Client{ph: ph, config: config}
}
// Capture sends a content_fetch_result event.
@ -39,7 +39,7 @@ func (c *Client) Capture(userIDs []string, ev Event) {
if c.ph == nil {
return
}
if !c.cfg.SendAnalytics || ev.Result != "failure" {
if !c.config.SendAnalytics || ev.Result != "failure" {
return
}
@ -48,7 +48,7 @@ func (c *Client) Capture(userIDs []string, ev Event) {
Set("url", ev.URL).
Set("source", ev.Source).
Set("totalTime", ev.TotalTime).
Set("env", c.cfg.APIEnv)
Set("env", c.config.APIEnv)
if ev.ErrorMessage != "" {
props.Set("errorMessage", ev.ErrorMessage)
}

10
packages/content-fetch-go/internal/browser/browser.go
View file

@ -13,14 +13,14 @@ import (
// Browser wraps a persistent chromedp browser allocator.
type Browser struct {
cfg *config.Config
config *config.Config
allocCtx context.Context
allocCancel context.CancelFunc
mu sync.Mutex
}
func New(cfg *config.Config) *Browser {
return &Browser{cfg: cfg}
func New(config *config.Config) *Browser {
return &Browser{config: config}
}
// allocatorOpts returns the chromedp ExecAllocator options matching the original Puppeteer args.
@ -50,8 +50,8 @@ func (b *Browser) allocatorOpts() []chromedp.ExecAllocatorOption {
chromedp.Headless,
)
if b.cfg.ChromiumPath != "" && !b.cfg.UseFirefox {
opts = append(opts, chromedp.ExecPath(b.cfg.ChromiumPath))
if b.config.ChromiumPath != "" && !b.config.UseFirefox {
opts = append(opts, chromedp.ExecPath(b.config.ChromiumPath))
}
return opts

50
packages/content-fetch-go/internal/bullmq/bullmq.go
View file

@ -79,8 +79,8 @@ type RawJob struct {
}
// nextJobID atomically increments and returns a new job ID.
func nextJobID(ctx context.Context, rdb *redis.Client, queueName string) (string, error) {
id, err := rdb.Incr(ctx, idKey(queueName)).Result()
func nextJobID(ctx context.Context, redisClient *redis.Client, queueName string) (string, error) {
id, err := redisClient.Incr(ctx, idKey(queueName)).Result()
if err != nil {
return "", err
}
@ -96,9 +96,9 @@ type AddJobOpts struct {
// AddBulk adds multiple jobs to a BullMQ queue, replicating addBulk() semantics.
// Each job is stored as a hash and its ID appended to the appropriate list/zset.
func AddBulk(ctx context.Context, rdb *redis.Client, queueName string, jobs []AddJobOpts) error {
func AddBulk(ctx context.Context, redisClient *redis.Client, queueName string, jobs []AddJobOpts) error {
for _, j := range jobs {
jobID, err := nextJobID(ctx, rdb, queueName)
jobID, err := nextJobID(ctx, redisClient, queueName)
if err != nil {
return fmt.Errorf("get next job id: %w", err)
}
@ -117,7 +117,7 @@ func AddBulk(ctx context.Context, rdb *redis.Client, queueName string, jobs []Ad
key := jobKey(queueName, jobID)
// Store the job hash
pipe := rdb.Pipeline()
pipe := redisClient.Pipeline()
pipe.HSet(ctx, key,
"name", j.Name,
"data", string(dataBytes),
@ -199,7 +199,7 @@ return jobId
// PopJob atomically moves the next available job to the active list and returns it.
// Returns nil job if no job is available (non-blocking).
func PopJob(ctx context.Context, rdb *redis.Client, queueName string) (*RawJob, error) {
func PopJob(ctx context.Context, redisClient *redis.Client, queueName string) (*RawJob, error) {
keys := []string{
waitKey(queueName),
prioritizedKey(queueName),
@ -207,7 +207,7 @@ func PopJob(ctx context.Context, rdb *redis.Client, queueName string) (*RawJob,
}
prefix := queueKey(queueName) + ":"
result, err := moveToActiveScript.Run(ctx, rdb, keys, prefix).Result()
result, err := moveToActiveScript.Run(ctx, redisClient, keys, prefix).Result()
if err == redis.Nil {
return nil, nil
}
@ -220,11 +220,11 @@ func PopJob(ctx context.Context, rdb *redis.Client, queueName string) (*RawJob,
return nil, nil
}
return getJob(ctx, rdb, queueName, jobID)
return getJob(ctx, redisClient, queueName, jobID)
}
func getJob(ctx context.Context, rdb *redis.Client, queueName, jobID string) (*RawJob, error) {
fields, err := rdb.HGetAll(ctx, jobKey(queueName, jobID)).Result()
func getJob(ctx context.Context, redisClient *redis.Client, queueName, jobID string) (*RawJob, error) {
fields, err := redisClient.HGetAll(ctx, jobKey(queueName, jobID)).Result()
if err != nil {
return nil, fmt.Errorf("hgetall job %s: %w", jobID, err)
}
@ -252,9 +252,9 @@ func getJob(ctx context.Context, rdb *redis.Client, queueName, jobID string) (*R
}
// CompleteJob moves a job from active to completed.
func CompleteJob(ctx context.Context, rdb *redis.Client, queueName, jobID string) error {
func CompleteJob(ctx context.Context, redisClient *redis.Client, queueName, jobID string) error {
now := time.Now().UnixMilli()
pipe := rdb.Pipeline()
pipe := redisClient.Pipeline()
pipe.LRem(ctx, activeKey(queueName), 0, jobID)
pipe.ZAdd(ctx, completedKey(queueName), redis.Z{Score: float64(now), Member: jobID})
pipe.HSet(ctx, jobKey(queueName, jobID), "finishedOn", now)
@ -265,11 +265,11 @@ func CompleteJob(ctx context.Context, rdb *redis.Client, queueName, jobID string
}
// FailJob moves a job from active to failed (or re-queues it for retry).
func FailJob(ctx context.Context, rdb *redis.Client, queueName, jobID string, reason string, opts JobOpts) error {
func FailJob(ctx context.Context, redisClient *redis.Client, queueName, jobID string, reason string, opts JobOpts) error {
now := time.Now().UnixMilli()
// Increment attemptsMade
newAttempts, err := rdb.HIncrBy(ctx, jobKey(queueName, jobID), "attemptsMade", 1).Result()
newAttempts, err := redisClient.HIncrBy(ctx, jobKey(queueName, jobID), "attemptsMade", 1).Result()
if err != nil {
return err
}
@ -279,7 +279,7 @@ func FailJob(ctx context.Context, rdb *redis.Client, queueName, jobID string, re
delay := exponentialDelay(opts.Backoff.Delay, int(newAttempts)-1)
retryAt := now + int64(delay)
pipe := rdb.Pipeline()
pipe := redisClient.Pipeline()
pipe.LRem(ctx, activeKey(queueName), 0, jobID)
pipe.ZAdd(ctx, fmt.Sprintf("%s:delayed", queueKey(queueName)), redis.Z{
Score: float64(retryAt),
@ -291,7 +291,7 @@ func FailJob(ctx context.Context, rdb *redis.Client, queueName, jobID string, re
}
// Max attempts reached → move to failed
pipe := rdb.Pipeline()
pipe := redisClient.Pipeline()
pipe.LRem(ctx, activeKey(queueName), 0, jobID)
pipe.ZAdd(ctx, failedKey(queueName), redis.Z{Score: float64(now), Member: jobID})
pipe.HSet(ctx, jobKey(queueName, jobID), "failedReason", reason, "finishedOn", now)
@ -310,8 +310,8 @@ func exponentialDelay(baseDelay, attempt int) int {
}
// GetQueueCounts returns job counts for the metrics endpoint.
func GetQueueCounts(ctx context.Context, rdb *redis.Client, queueName string) (map[string]int64, error) {
pipe := rdb.Pipeline()
func GetQueueCounts(ctx context.Context, redisClient *redis.Client, queueName string) (map[string]int64, error) {
pipe := redisClient.Pipeline()
activeCmd := pipe.LLen(ctx, activeKey(queueName))
failedCmd := pipe.ZCard(ctx, failedKey(queueName))
completedCmd := pipe.ZCard(ctx, completedKey(queueName))
@ -331,9 +331,9 @@ func GetQueueCounts(ctx context.Context, rdb *redis.Client, queueName string) (m
}
// OldestPrioritizedJobAge returns the age in seconds of the oldest prioritized job, or 0.
func OldestPrioritizedJobAge(ctx context.Context, rdb *redis.Client, queueName string) (float64, error) {
func OldestPrioritizedJobAge(ctx context.Context, redisClient *redis.Client, queueName string) (float64, error) {
// Check both prioritized zset and wait list
results, err := rdb.ZRangeWithScores(ctx, prioritizedKey(queueName), 0, 0).Result()
results, err := redisClient.ZRangeWithScores(ctx, prioritizedKey(queueName), 0, 0).Result()
if err != nil {
return 0, err
}
@ -341,7 +341,7 @@ func OldestPrioritizedJobAge(ctx context.Context, rdb *redis.Client, queueName s
if len(results) > 0 {
// score encodes priority+counter, so we need the job's timestamp field
jobID := results[0].Member.(string)
ts, err := rdb.HGet(ctx, jobKey(queueName, jobID), "timestamp").Result()
ts, err := redisClient.HGet(ctx, jobKey(queueName, jobID), "timestamp").Result()
if err == nil {
if tsMs, err := strconv.ParseInt(ts, 10, 64); err == nil {
return float64(time.Now().UnixMilli()-tsMs) / 1000.0, nil
@ -350,11 +350,11 @@ func OldestPrioritizedJobAge(ctx context.Context, rdb *redis.Client, queueName s
}
// Fall back to wait list
waitIDs, err := rdb.LRange(ctx, waitKey(queueName), -1, -1).Result() // oldest = tail
waitIDs, err := redisClient.LRange(ctx, waitKey(queueName), -1, -1).Result() // oldest = tail
if err != nil || len(waitIDs) == 0 {
return 0, nil
}
ts, err := rdb.HGet(ctx, jobKey(queueName, waitIDs[0]), "timestamp").Result()
ts, err := redisClient.HGet(ctx, jobKey(queueName, waitIDs[0]), "timestamp").Result()
if err != nil {
return 0, nil
}
@ -366,6 +366,6 @@ func OldestPrioritizedJobAge(ctx context.Context, rdb *redis.Client, queueName s
}
// EnsureQueueMeta ensures the queue metadata key exists (BullMQ creates this on queue init).
func EnsureQueueMeta(ctx context.Context, rdb *redis.Client, queueName string) error {
return rdb.HSetNX(ctx, metaKey(queueName), "version", "5").Err()
func EnsureQueueMeta(ctx context.Context, redisClient *redis.Client, queueName string) error {
return redisClient.HSetNX(ctx, metaKey(queueName), "version", "5").Err()
}

8
packages/content-fetch-go/internal/fetch/fetch.go
View file

@ -48,7 +48,7 @@ var NoCacheURLs = map[string]bool{
}
// FetchContent is the Go equivalent of fetchContent() in puppeteer-parse/src/index.ts.
func FetchContent(ctx context.Context, br *browser.Browser, rawURL, locale, timezone string) (*Result, error) {
func FetchContent(ctx context.Context, browser *browser.Browser, rawURL, locale, timezone string) (*Result, error) {
start := time.Now()
log.Printf("content-fetch request url=%s locale=%s timezone=%s", rawURL, locale, timezone)
@ -82,7 +82,7 @@ func FetchContent(ctx context.Context, br *browser.Browser, rawURL, locale, time
}
// Fall through to browser fetch
result, err := retrievePage(ctx, br, targetURL, locale, timezone)
result, err := retrievePage(ctx, browser, targetURL, locale, timezone)
if err != nil {
return nil, err
}
@ -146,8 +146,8 @@ func preHandle(ctx context.Context, rawURL string) (*preHandleResult, error) {
// retrievePage navigates to the URL using the browser, waits for load and DOM settle,
// then captures the full document HTML.
func retrievePage(ctx context.Context, br *browser.Browser, targetURL, locale, timezone string) (*Result, error) {
tabCtx, cancel, err := br.NewContext()
func retrievePage(ctx context.Context, browser *browser.Browser, targetURL, locale, timezone string) (*Result, error) {
tabCtx, cancel, err := browser.NewContext()
if err != nil {
return nil, fmt.Errorf("new browser context: %w", err)
}

52
packages/content-fetch-go/internal/handler/handler.go
View file

@ -73,9 +73,9 @@ const maxImportAttempts = 1
// ProcessFetchContentJob is the Go equivalent of processFetchContentJob() from request_handler.ts.
func ProcessFetchContentJob(
ctx context.Context,
cfg *config.Config,
rds *redisutil.RedisDataSource,
br *browser.Browser,
config *config.Config,
redisDS *redisutil.RedisDataSource,
browser *browser.Browser,
data *JobData,
attemptsMade int,
) error {
@ -132,7 +132,7 @@ func ProcessFetchContentJob(
result = "failure"
errMsg = processErr.Error()
}
analyticsClient := analytics.New(cfg)
analyticsClient := analytics.New(config)
analyticsClient.Capture(userIDs, analytics.Event{
Result: result,
URL: data.URL,
@ -147,7 +147,7 @@ func ProcessFetchContentJob(
if processErr != nil && data.TaskID != nil && *data.TaskID != "" && lastAttempt {
log.Println("Sending import status update (failure)")
if len(users) > 0 {
sendImportStatusUpdate(ctx, cfg, users[0].ID, *data.TaskID, false)
sendImportStatusUpdate(ctx, config, users[0].ID, *data.TaskID, false)
}
}
}()
@ -159,7 +159,7 @@ func ProcessFetchContentJob(
return processErr
}
blocked, err := isDomainBlocked(ctx, rds, domain, cfg.MaxFeedFetchFailures)
blocked, err := isDomainBlocked(ctx, redisDS, domain, config.MaxFeedFetchFailures)
if err != nil {
log.Printf("Error checking domain block: %v", err)
}
@ -171,16 +171,16 @@ func ProcessFetchContentJob(
// Try cache
cacheKey := buildCacheKey(data.URL, locale, timezone)
fetchResult, err := getCachedResult(ctx, rds, cacheKey)
fetchResult, err := getCachedResult(ctx, redisDS, cacheKey)
if err != nil {
log.Printf("Cache read error: %v", err)
}
if fetchResult == nil {
log.Printf("Fetch result not in cache, fetching now: %s", data.URL)
fetchResult, err = fetch.FetchContent(ctx, br, data.URL, locale, timezone)
fetchResult, err = fetch.FetchContent(ctx, browser, data.URL, locale, timezone)
if err != nil {
_ = incrementDomainFailure(ctx, rds, domain)
_ = incrementDomainFailure(ctx, redisDS, domain)
processErr = fmt.Errorf("fetchContent: %w", err)
return processErr
}
@ -188,7 +188,7 @@ func ProcessFetchContentJob(
// Cache result (skip NO_CACHE_URLS)
if fetchResult.Content != "" && !fetch.NoCacheURLs[data.URL] {
if err := cacheResult(ctx, rds, cacheKey, fetchResult); err != nil {
if err := cacheResult(ctx, redisDS, cacheKey, fetchResult); err != nil {
log.Printf("Cache write error: %v", err)
}
}
@ -202,8 +202,8 @@ func ProcessFetchContentJob(
}
// Upload original content to GCS
if fetchResult.Content != "" && !cfg.SkipUploadOriginal {
gcsClient, err := gcs.New(ctx, cfg.GCSUploadBucket, cfg.GCSKeyFilePath)
if fetchResult.Content != "" && !config.SkipUploadOriginal {
gcsClient, err := gcs.New(ctx, config.GCSUploadBucket, config.GCSKeyFilePath)
if err != nil {
log.Printf("GCS client init error: %v", err)
} else {
@ -261,7 +261,7 @@ func ProcessFetchContentJob(
})
}
if err := bullmq.AddBulk(ctx, rds.MQClient, bullmq.BackendQueue, savePageJobs); err != nil {
if err := bullmq.AddBulk(ctx, redisDS.MQClient, bullmq.BackendQueue, savePageJobs); err != nil {
processErr = fmt.Errorf("queue save-page jobs: %w", err)
return processErr
}
@ -312,8 +312,8 @@ type cachedFetchResult struct {
}
// getCachedResult attempts to get a cached fetch result from Redis.
func getCachedResult(ctx context.Context, rds *redisutil.RedisDataSource, key string) (*fetch.Result, error) {
val, err := rds.CacheClient.Get(ctx, key).Result()
func getCachedResult(ctx context.Context, redisDS *redisutil.RedisDataSource, key string) (*fetch.Result, error) {
val, err := redisDS.CacheClient.Get(ctx, key).Result()
if err == redis.Nil {
log.Printf("Fetch result not cached: %s", key)
return nil, nil
@ -342,7 +342,7 @@ func getCachedResult(ctx context.Context, rds *redisutil.RedisDataSource, key st
}
// cacheResult stores a fetch result in Redis with a 24-hour TTL (NX = only if not exists).
func cacheResult(ctx context.Context, rds *redisutil.RedisDataSource, key string, r *fetch.Result) error {
func cacheResult(ctx context.Context, redisDS *redisutil.RedisDataSource, key string, r *fetch.Result) error {
val, err := json.Marshal(cachedFetchResult{
FinalURL: r.FinalURL,
Title: r.Title,
@ -352,7 +352,7 @@ func cacheResult(ctx context.Context, rds *redisutil.RedisDataSource, key string
if err != nil {
return err
}
return rds.CacheClient.SetNX(ctx, key, string(val), 24*time.Hour).Err()
return redisDS.CacheClient.SetNX(ctx, key, string(val), 24*time.Hour).Err()
}
// failureRedisKey mirrors failureRedisKey() from request_handler.ts.
@ -361,7 +361,7 @@ func failureRedisKey(domain string) string {
}
// isDomainBlocked mirrors isDomainBlocked() from request_handler.ts.
func isDomainBlocked(ctx context.Context, rds *redisutil.RedisDataSource, domain string, maxFailures int) (bool, error) {
func isDomainBlocked(ctx context.Context, redisDS *redisutil.RedisDataSource, domain string, maxFailures int) (bool, error) {
blockedDomains := map[string]bool{
"localhost": true,
"weibo.com": true,
@ -371,7 +371,7 @@ func isDomainBlocked(ctx context.Context, rds *redisutil.RedisDataSource, domain
}
key := failureRedisKey(domain)
val, err := rds.CacheClient.Get(ctx, key).Result()
val, err := redisDS.CacheClient.Get(ctx, key).Result()
if err == redis.Nil {
return false, nil
}
@ -393,17 +393,17 @@ func isDomainBlocked(ctx context.Context, rds *redisutil.RedisDataSource, domain
}
// incrementDomainFailure mirrors incrementContentFetchFailure() from request_handler.ts.
func incrementDomainFailure(ctx context.Context, rds *redisutil.RedisDataSource, domain string) error {
func incrementDomainFailure(ctx context.Context, redisDS *redisutil.RedisDataSource, domain string) error {
key := failureRedisKey(domain)
if err := rds.CacheClient.Incr(ctx, key).Err(); err != nil {
if err := redisDS.CacheClient.Incr(ctx, key).Err(); err != nil {
return err
}
return rds.CacheClient.Expire(ctx, key, time.Hour).Err()
return redisDS.CacheClient.Expire(ctx, key, time.Hour).Err()
}
// sendImportStatusUpdate mirrors sendImportStatusUpdate() from request_handler.ts.
func sendImportStatusUpdate(ctx context.Context, cfg *config.Config, userID, taskID string, isImported bool) {
if cfg.JWTSecret == "" || cfg.ImporterMetricsCollectorURL == "" {
func sendImportStatusUpdate(ctx context.Context, config *config.Config, userID, taskID string, isImported bool) {
if config.JWTSecret == "" || config.ImporterMetricsCollectorURL == "" {
log.Println("JWT_SECRET or IMPORTER_METRICS_COLLECTOR_URL not set, skipping import status update")
return
}
@ -411,7 +411,7 @@ func sendImportStatusUpdate(ctx context.Context, cfg *config.Config, userID, tas
token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
"uid": userID,
})
tokenStr, err := token.SignedString([]byte(cfg.JWTSecret))
tokenStr, err := token.SignedString([]byte(config.JWTSecret))
if err != nil {
log.Printf("Failed to sign JWT: %v", err)
return
@ -430,7 +430,7 @@ func sendImportStatusUpdate(ctx context.Context, cfg *config.Config, userID, tas
reqCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, cfg.ImporterMetricsCollectorURL, bytes.NewReader(body))
req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, config.ImporterMetricsCollectorURL, bytes.NewReader(body))
if err != nil {
log.Printf("Failed to create import status request: %v", err)
return

10
packages/content-fetch-go/internal/metrics/metrics.go
View file

@ -54,10 +54,10 @@ func init() {
// Handler returns an http.Handler that refreshes queue metrics from Redis on
// every request and then delegates to the standard promhttp handler.
func Handler(rdb *redis.Client, queueName string) http.Handler {
func Handler(redisClient *redis.Client, queueName string) http.Handler {
inner := promhttp.Handler()
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if err := refresh(r.Context(), rdb, queueName); err != nil {
if err := refresh(r.Context(), redisClient, queueName); err != nil {
log.Printf("Error refreshing queue metrics: %v", err)
http.Error(w, "internal error", http.StatusInternalServerError)
return
@ -67,8 +67,8 @@ func Handler(rdb *redis.Client, queueName string) http.Handler {
}
// refresh pulls the current queue counts from Redis and updates the gauges.
func refresh(ctx context.Context, rdb *redis.Client, queueName string) error {
counts, err := bullmq.GetQueueCounts(ctx, rdb, queueName)
func refresh(ctx context.Context, redisClient *redis.Client, queueName string) error {
counts, err := bullmq.GetQueueCounts(ctx, redisClient, queueName)
if err != nil {
return err
}
@ -79,7 +79,7 @@ func refresh(ctx context.Context, rdb *redis.Client, queueName string) error {
completedGauge.With(labels).Set(float64(counts["completed"]))
prioritizedGauge.With(labels).Set(float64(counts["prioritized"]))
age, err := bullmq.OldestPrioritizedJobAge(ctx, rdb, queueName)
age, err := bullmq.OldestPrioritizedJobAge(ctx, redisClient, queueName)
if err != nil {
log.Printf("Error getting oldest job age: %v", err)
age = 0

26
packages/content-fetch-go/internal/queue/worker.go
View file

@ -22,19 +22,19 @@ const (
// Worker processes jobs from the content-fetch BullMQ queue.
type Worker struct {
ctx context.Context
cfg *config.Config
rds *redisutil.RedisDataSource
br *browser.Browser
config *config.Config
redisDS *redisutil.RedisDataSource
browser *browser.Browser
wg sync.WaitGroup
sem chan struct{}
}
func NewWorker(ctx context.Context, cfg *config.Config, rds *redisutil.RedisDataSource, br *browser.Browser) *Worker {
func NewWorker(ctx context.Context, config *config.Config, redisDS *redisutil.RedisDataSource, browser *browser.Browser) *Worker {
return &Worker{
ctx: ctx,
cfg: cfg,
rds: rds,
br: br,
config: config,
redisDS: redisDS,
browser: browser,
sem: make(chan struct{}, workerConcurrency),
}
}
@ -54,7 +54,7 @@ func (w *Worker) run() {
log.Println("Queue worker started")
// Ensure queue meta exists
_ = bullmq.EnsureQueueMeta(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue)
_ = bullmq.EnsureQueueMeta(w.ctx, w.redisDS.MQClient, bullmq.ContentFetchQueue)
for {
select {
@ -68,7 +68,7 @@ func (w *Worker) run() {
default:
}
job, err := bullmq.PopJob(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue)
job, err := bullmq.PopJob(w.ctx, w.redisDS.MQClient, bullmq.ContentFetchQueue)
if err != nil {
log.Printf("Error popping job: %v", err)
time.Sleep(workerPollInterval)
@ -95,16 +95,16 @@ func (w *Worker) processJob(job *bullmq.RawJob) {
var data handler.JobData
if err := json.Unmarshal(job.Data, &data); err != nil {
log.Printf("Failed to unmarshal job data id=%s: %v", job.ID, err)
_ = bullmq.FailJob(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue, job.ID, err.Error(), job.Opts)
_ = bullmq.FailJob(w.ctx, w.redisDS.MQClient, bullmq.ContentFetchQueue, job.ID, err.Error(), job.Opts)
return
}
if err := handler.ProcessFetchContentJob(w.ctx, w.cfg, w.rds, w.br, &data, job.AttemptsMade); err != nil {
if err := handler.ProcessFetchContentJob(w.ctx, w.config, w.redisDS, w.browser, &data, job.AttemptsMade); err != nil {
log.Printf("Job id=%s failed: %v", job.ID, err)
_ = bullmq.FailJob(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue, job.ID, err.Error(), job.Opts)
_ = bullmq.FailJob(w.ctx, w.redisDS.MQClient, bullmq.ContentFetchQueue, job.ID, err.Error(), job.Opts)
return
}
_ = bullmq.CompleteJob(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue, job.ID)
_ = bullmq.CompleteJob(w.ctx, w.redisDS.MQClient, bullmq.ContentFetchQueue, job.ID)
log.Printf("Job id=%s completed", job.ID)
}

18
packages/content-fetch-go/internal/server/server.go
View file

@ -22,19 +22,19 @@ type Worker interface {
}
type mux struct {
cfg *config.Config
rds *redisutil.RedisDataSource
br *browser.Browser
worker Worker
config *config.Config
redisDS *redisutil.RedisDataSource
browser *browser.Browser
worker Worker
http.ServeMux
}
// New returns an http.Handler with all routes registered.
func New(cfg *config.Config, rds *redisutil.RedisDataSource, br *browser.Browser, w Worker) http.Handler {
m := &mux{cfg: cfg, rds: rds, br: br, worker: w}
func New(config *config.Config, redisDS *redisutil.RedisDataSource, browser *browser.Browser, worker Worker) http.Handler {
m := &mux{config: config, redisDS: redisDS, browser: browser, worker: worker}
m.HandleFunc("GET /_ah/health", m.health)
m.HandleFunc("GET /lifecycle/prestop", m.prestop)
m.Handle("GET /metrics", metrics.Handler(rds.MQClient, bullmq.ContentFetchQueue))
m.Handle("GET /metrics", metrics.Handler(redisDS.MQClient, bullmq.ContentFetchQueue))
m.HandleFunc("/", m.root) // GET and POST
return m
}
@ -66,7 +66,7 @@ func (m *mux) root(w http.ResponseWriter, r *http.Request) {
return
}
if r.URL.Query().Get("token") != m.cfg.VerificationToken {
if r.URL.Query().Get("token") != m.config.VerificationToken {
log.Println("Query does not include valid token")
w.WriteHeader(http.StatusForbidden)
return
@ -88,7 +88,7 @@ func (m *mux) root(w http.ResponseWriter, r *http.Request) {
if err := handler.ProcessFetchContentJob(
context.Background(),
m.cfg, m.rds, m.br,
m.config, m.redisDS, m.browser,
&data, attempt,
); err != nil {
log.Printf("Error fetching content: %v", err)

18
packages/content-fetch-go/main.go
View file

@ -17,26 +17,26 @@ import (
)
func main() {
cfg := config.Load()
config := config.Load()
if cfg.VerificationToken == "" {
if config.VerificationToken == "" {
log.Fatal("VERIFICATION_TOKEN is required")
}
rds, err := redisutil.New(cfg)
redisDS, err := redisutil.New(config)
if err != nil {
log.Fatalf("Failed to connect to Redis: %v", err)
}
br := browser.New(cfg)
browser := browser.New(config)
workerCtx, workerCancel := context.WithCancel(context.Background())
worker := queue.NewWorker(workerCtx, cfg, rds, br)
worker := queue.NewWorker(workerCtx, config, redisDS, browser)
worker.Start()
srv := server.New(cfg, rds, br, worker)
srv := server.New(config, redisDS, browser, worker)
port := cfg.Port
port := config.Port
if port == 0 {
port = 3002
}
@ -71,10 +71,10 @@ func main() {
log.Println("Worker closed")
// Close browser
br.Close()
browser.Close()
log.Println("Browser closed")
// Close Redis
rds.Shutdown()
redisDS.Shutdown()
log.Println("Redis connection closed")
}