From 06474845979191cc557c9ee2c5997dff5541ed6a Mon Sep 17 00:00:00 2001 From: Aliaksei Karneyeu Date: Wed, 4 Mar 2026 14:41:08 +0100 Subject: [PATCH] Add Go replacement for content-fetch service and CLAUDE.md - Add packages/content-fetch-go: a Go rewrite of packages/content-fetch that is fully API-compatible (same HTTP endpoints, env vars, Redis key schema, and BullMQ v5 job format) but produces a significantly smaller Docker image (Alpine + Chromium only, no Node.js runtime) - Internal packages: - bullmq: BullMQ v5-compatible producer and consumer (Lua moveToActive) - queue/worker: concurrent job worker (4 goroutines, 500 ms poll) - fetch: Chromium page fetch via chromedp replacing puppeteer-parse - handler: processFetchContentJob logic (cache, GCS upload, job queuing) - gcs: Google Cloud Storage upload - analytics: PostHog failure event capture - server: HTTP endpoints (/_ah/health, /metrics, /lifecycle/prestop, /) - redisutil: dual Redis connections (cache + BullMQ MQ) - config: all env var loading - Add CLAUDE.md with build/test/lint commands and architecture overview Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 133 +++++ packages/content-fetch-go/.dockerignore | 3 + packages/content-fetch-go/Dockerfile | 49 ++ packages/content-fetch-go/go.mod | 70 +++ packages/content-fetch-go/go.sum | 166 +++++++ .../internal/analytics/analytics.go | 73 +++ .../internal/browser/browser.go | 106 ++++ .../internal/bullmq/bullmq.go | 371 ++++++++++++++ .../internal/config/config.go | 92 ++++ .../content-fetch-go/internal/fetch/fetch.go | 450 +++++++++++++++++ packages/content-fetch-go/internal/gcs/gcs.go | 73 +++ .../internal/handler/handler.go | 457 ++++++++++++++++++ .../content-fetch-go/internal/queue/worker.go | 110 +++++ .../internal/redisutil/redisutil.go | 88 ++++ .../internal/server/server.go | 132 +++++ packages/content-fetch-go/main.go | 80 +++ 16 files changed, 2453 insertions(+) create mode 100644 CLAUDE.md create mode 100644 packages/content-fetch-go/.dockerignore create mode 100644 packages/content-fetch-go/Dockerfile create mode 100644 packages/content-fetch-go/go.mod create mode 100644 packages/content-fetch-go/go.sum create mode 100644 packages/content-fetch-go/internal/analytics/analytics.go create mode 100644 packages/content-fetch-go/internal/browser/browser.go create mode 100644 packages/content-fetch-go/internal/bullmq/bullmq.go create mode 100644 packages/content-fetch-go/internal/config/config.go create mode 100644 packages/content-fetch-go/internal/fetch/fetch.go create mode 100644 packages/content-fetch-go/internal/gcs/gcs.go create mode 100644 packages/content-fetch-go/internal/handler/handler.go create mode 100644 packages/content-fetch-go/internal/queue/worker.go create mode 100644 packages/content-fetch-go/internal/redisutil/redisutil.go create mode 100644 packages/content-fetch-go/internal/server/server.go create mode 100644 packages/content-fetch-go/main.go diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 000000000..55e7b2535 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,133 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +Omnivore is an open-source, self-hosted read-it-later application. The monorepo contains 24+ packages managed with **Yarn 1.22.19** and **Lerna**. Primary languages are TypeScript/JavaScript, with Swift (iOS), Kotlin (Android), and Go (imageproxy). + +## Commands + +### Development Setup +```bash +# Start all backend services (PostgreSQL, Redis, API, Content-Fetch) +docker compose up + +# Frontend development against dockerized backend +cd packages/web && cp .env.template .env.local && yarn dev + +# Run API dev server with hot-reload +make api # or: yarn workspace @omnivore/api dev + +# Run web dev server +make web # or: yarn workspace @omnivore/web dev + +# Run queue processor +make qp # or: yarn workspace @omnivore/api dev_qp +``` + +### Build +```bash +yarn build # Build all packages via Lerna + +# Build a specific package +yarn workspace @omnivore/api build +yarn workspace @omnivore/web build + +# Build content-fetch service (includes dependencies) +make content_fetch # builds content-handler, puppeteer-parse, content-fetch in order +``` + +### Testing +```bash +yarn test # Run all tests via Lerna (streaming output) + +# Run tests for a specific package +lerna run test --scope=@omnivore/api + +# Run a single test file (from within a package directory) +yarn mocha -r ts-node/register --config mocha-config.json test/path/to/file.test.ts + +# Type checking +yarn workspace @omnivore/api test:typecheck +``` + +Tests require PostgreSQL and Redis — use `docker compose -f docker-compose-test.yml up` for a test environment. + +### Linting +```bash +yarn lint # Lint all packages (ESLint, parallel) +yarn workspace @omnivore/api lint +yarn workspace @omnivore/api lint:fix # Auto-fix +``` + +Code style: no semicolons, single quotes (enforced by Prettier + ESLint with `@typescript-eslint`). + +## Architecture + +### Core Services + +**`packages/api`** — The central backend. GraphQL API (Apollo Server 3 on Express), running on port 4000 (8080 in Docker). Connects to PostgreSQL via TypeORM and Redis via ioredis/BullMQ. Exposes both a `/graphql` endpoint and REST routers under `/api/` and `/svc/` prefixes. + +**`packages/web`** — Next.js 14 frontend (React 18, TypeScript). Runs on port 3000. Uses SWR + TanStack Query for data fetching, Stitches for CSS-in-JS, and Radix UI components. Communicates with the API exclusively via GraphQL. + +**`packages/content-fetch`** — Async microservice for fetching and processing web content. Picks jobs off a BullMQ/Redis queue. Depends on `puppeteer-parse` and `content-handler`. + +**`packages/puppeteer-parse`** — Headless Chromium via Puppeteer for rendering and extracting page content. Called by `content-fetch`. + +**`packages/content-handler`** — Shared utilities for HTML parsing, sanitization, and Mozilla Readability integration. Used by both `api` and `puppeteer-parse`. + +**`packages/db`** — Database schema and migrations via Postgrator. Migration files live here; TypeORM entities live in `packages/api/src/entity/`. + +### Specialized Handlers (Cloud Function-style) + +These packages run independently and are invoked via HTTP or queue: +- `rss-handler` — RSS feed polling +- `inbound-email-handler` — Email ingestion +- `import-handler` / `export-handler` — Data migration/export jobs +- `pdf-handler` — PDF processing +- `thumbnail-handler` — Thumbnail generation +- `text-to-speech` — TTS audio generation +- `integration-handler` — Third-party integrations (Logseq, Obsidian, Readwise, etc.) +- `rule-handler` — User-defined automation rules + +### Shared Libraries + +- `packages/liqe` — Query parser/filter engine for Omnivore's search syntax +- `packages/readabilityjs` — Fork of Mozilla Readability +- `packages/utils` — Shared TypeScript utilities +- `packages/appreader` — WebView bundle (Next.js app) embedded in iOS and Android apps + +### Mobile and Native + +- `apple/` — iOS/macOS app in Swift/SwiftUI; uses `packages/appreader` bundle +- `android/` — Android app in Kotlin; uses `packages/appreader` bundle +- `imageproxy/` — Go service for proxying and caching images + +### Data Flow + +``` +Web/Mobile client + → GraphQL API (Apollo/Express) → PostgreSQL (TypeORM) + → Redis (cache + BullMQ queues) + → BullMQ queue workers + → content-fetch → puppeteer-parse → content-handler → stored article HTML +``` + +### Key Patterns + +- **GraphQL schema** is defined in `packages/api/src/schema/` and generated into `packages/api/src/generated/schema.graphql` +- **Resolvers** are in `packages/api/src/resolvers/` +- **REST service routes** (internal service-to-service) are under `packages/api/src/routers/svc/` +- **TypeORM entities** mirror the PostgreSQL schema in `packages/api/src/entity/` +- **Queue jobs** are defined in `packages/api/src/queue-processor.ts` and processed by BullMQ workers + +### Environment + +Key env vars (see `packages/api/.env.example` and `packages/web/.env.template`): +- `PG_HOST/PG_USER/PG_PASSWORD/PG_DB` — PostgreSQL connection +- `REDIS_URL` — Redis connection (used for both cache and queues) +- `JWT_SECRET` — Token signing +- `CONTENT_FETCH_URL` — URL of the content-fetch service +- `CLIENT_URL` — Frontend URL (for CORS and redirects) +- `IMAGE_PROXY_SECRET` / `IMAGE_PROXY_URL` — Image proxy config diff --git a/packages/content-fetch-go/.dockerignore b/packages/content-fetch-go/.dockerignore new file mode 100644 index 000000000..afdd9d8e7 --- /dev/null +++ b/packages/content-fetch-go/.dockerignore @@ -0,0 +1,3 @@ +Dockerfile +.dockerignore +*.md diff --git a/packages/content-fetch-go/Dockerfile b/packages/content-fetch-go/Dockerfile new file mode 100644 index 000000000..20d08d4e6 --- /dev/null +++ b/packages/content-fetch-go/Dockerfile @@ -0,0 +1,49 @@ +FROM golang:1.24-alpine AS build +LABEL org.opencontainers.image.source="https://github.com/omnivore-app/omnivore" + +RUN apk add --no-cache git ca-certificates + +WORKDIR /app + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . +RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o content-fetch-go . + +# ─── Runtime image ──────────────────────────────────────────────────────────── +FROM alpine:3.20 + +LABEL org.opencontainers.image.source="https://github.com/omnivore-app/omnivore" + +# Add Chromium and required fonts/libs from Alpine edge (mirrors the original Dockerfile) +RUN echo "@edge https://dl-cdn.alpinelinux.org/alpine/edge/community" >> /etc/apk/repositories \ + && echo "@edge https://dl-cdn.alpinelinux.org/alpine/edge/main" >> /etc/apk/repositories \ + && echo "@edge https://dl-cdn.alpinelinux.org/alpine/edge/testing" >> /etc/apk/repositories \ + && apk -U upgrade \ + && apk add --no-cache \ + chromium@edge \ + freetype@edge \ + ttf-freefont@edge \ + nss@edge \ + libstdc++@edge \ + sqlite-libs@edge \ + ca-certificates@edge \ + && rm -rf /var/cache/apk/* + +WORKDIR /app + +ENV CHROMIUM_PATH=/usr/bin/chromium +ENV LAUNCH_HEADLESS=true +ENV PORT=8080 + +# Download ad/tracker block-list for the hosts file (matches original start.sh) +RUN wget -q https://raw.githubusercontent.com/StevenBlack/hosts/master/hosts \ + && cat hosts >> /etc/hosts \ + && rm hosts + +COPY --from=build /app/content-fetch-go . + +EXPOSE 8080 + +CMD ["./content-fetch-go"] diff --git a/packages/content-fetch-go/go.mod b/packages/content-fetch-go/go.mod new file mode 100644 index 000000000..63e90053b --- /dev/null +++ b/packages/content-fetch-go/go.mod @@ -0,0 +1,70 @@ +module github.com/omnivore-app/omnivore/content-fetch-go + +go 1.25.0 + +require ( + cloud.google.com/go/storage v1.60.0 + github.com/chromedp/cdproto v0.0.0-20250724212937-08a3db8b4327 + github.com/chromedp/chromedp v0.14.2 + github.com/golang-jwt/jwt/v5 v5.3.1 + github.com/posthog/posthog-go v1.10.0 + github.com/redis/go-redis/v9 v9.18.0 + google.golang.org/api v0.265.0 +) + +require ( + cel.dev/expr v0.24.0 // indirect + cloud.google.com/go v0.123.0 // indirect + cloud.google.com/go/auth v0.18.1 // indirect + cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect + cloud.google.com/go/compute/metadata v0.9.0 // indirect + cloud.google.com/go/iam v1.5.3 // indirect + cloud.google.com/go/monitoring v1.24.3 // indirect + github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 // indirect + github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0 // indirect + github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/chromedp/sysutil v1.1.0 // indirect + github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/envoyproxy/go-control-plane/envoy v1.35.0 // indirect + github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect + github.com/go-jose/go-jose/v4 v4.1.3 // indirect + github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/gobwas/httphead v0.1.0 // indirect + github.com/gobwas/pool v0.2.1 // indirect + github.com/gobwas/ws v1.4.0 // indirect + github.com/goccy/go-json v0.10.5 // indirect + github.com/google/s2a-go v0.1.9 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/googleapis/enterprise-certificate-proxy v0.3.11 // indirect + github.com/googleapis/gax-go/v2 v2.17.0 // indirect + github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect + github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect + github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/detectors/gcp v1.38.0 // indirect + go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect + go.opentelemetry.io/otel v1.39.0 // indirect + go.opentelemetry.io/otel/metric v1.39.0 // indirect + go.opentelemetry.io/otel/sdk v1.39.0 // indirect + go.opentelemetry.io/otel/sdk/metric v1.39.0 // indirect + go.opentelemetry.io/otel/trace v1.39.0 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/crypto v0.47.0 // indirect + golang.org/x/net v0.49.0 // indirect + golang.org/x/oauth2 v0.35.0 // indirect + golang.org/x/sync v0.19.0 // indirect + golang.org/x/sys v0.40.0 // indirect + golang.org/x/text v0.33.0 // indirect + golang.org/x/time v0.14.0 // indirect + google.golang.org/genproto v0.0.0-20260128011058-8636f8732409 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260203192932-546029d2fa20 // indirect + google.golang.org/grpc v1.78.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect +) diff --git a/packages/content-fetch-go/go.sum b/packages/content-fetch-go/go.sum new file mode 100644 index 000000000..faf93d16d --- /dev/null +++ b/packages/content-fetch-go/go.sum @@ -0,0 +1,166 @@ +cel.dev/expr v0.24.0 h1:56OvJKSH3hDGL0ml5uSxZmz3/3Pq4tJ+fb1unVLAFcY= +cel.dev/expr v0.24.0/go.mod h1:hLPLo1W4QUmuYdA72RBX06QTs6MXw941piREPl3Yfiw= +cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= +cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= +cloud.google.com/go/auth v0.18.1 h1:IwTEx92GFUo2pJ6Qea0EU3zYvKnTAeRCODxfA/G5UWs= +cloud.google.com/go/auth v0.18.1/go.mod h1:GfTYoS9G3CWpRA3Va9doKN9mjPGRS+v41jmZAhBzbrA= +cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= +cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= +cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= +cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= +cloud.google.com/go/iam v1.5.3 h1:+vMINPiDF2ognBJ97ABAYYwRgsaqxPbQDlMnbHMjolc= +cloud.google.com/go/iam v1.5.3/go.mod h1:MR3v9oLkZCTlaqljW6Eb2d3HGDGK5/bDv93jhfISFvU= +cloud.google.com/go/logging v1.13.1 h1:O7LvmO0kGLaHY/gq8cV7T0dyp6zJhYAOtZPX4TF3QtY= +cloud.google.com/go/logging v1.13.1/go.mod h1:XAQkfkMBxQRjQek96WLPNze7vsOmay9H5PqfsNYDqvw= +cloud.google.com/go/longrunning v0.8.0 h1:LiKK77J3bx5gDLi4SMViHixjD2ohlkwBi+mKA7EhfW8= +cloud.google.com/go/longrunning v0.8.0/go.mod h1:UmErU2Onzi+fKDg2gR7dusz11Pe26aknR4kHmJJqIfk= +cloud.google.com/go/monitoring v1.24.3 h1:dde+gMNc0UhPZD1Azu6at2e79bfdztVDS5lvhOdsgaE= +cloud.google.com/go/monitoring v1.24.3/go.mod h1:nYP6W0tm3N9H/bOw8am7t62YTzZY+zUeQ+Bi6+2eonI= +cloud.google.com/go/storage v1.60.0 h1:oBfZrSOCimggVNz9Y/bXY35uUcts7OViubeddTTVzQ8= +cloud.google.com/go/storage v1.60.0/go.mod h1:q+5196hXfejkctrnx+VYU8RKQr/L3c0cBIlrjmiAKE0= +cloud.google.com/go/trace v1.11.7 h1:kDNDX8JkaAG3R2nq1lIdkb7FCSi1rCmsEtKVsty7p+U= +cloud.google.com/go/trace v1.11.7/go.mod h1:TNn9d5V3fQVf6s4SCveVMIBS2LJUqo73GACmq/Tky0s= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 h1:sBEjpZlNHzK1voKq9695PJSX2o5NEXl7/OL3coiIY0c= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0/go.mod h1:P4WPRUkOhJC13W//jWpyfJNDAIpvRbAUIYLX/4jtlE0= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0 h1:UnDZ/zFfG1JhH/DqxIZYU/1CUAlTUScoXD/LcM2Ykk8= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0/go.mod h1:IA1C1U7jO/ENqm/vhi7V9YYpBsp+IMyqNrEN94N7tVc= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.55.0 h1:7t/qx5Ost0s0wbA/VDrByOooURhp+ikYwv20i9Y07TQ= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.55.0/go.mod h1:vB2GH9GAYYJTO3mEn8oYwzEdhlayZIdQz6zdzgUIRvA= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0 h1:0s6TxfCu2KHkkZPnBfsQ2y5qia0jl3MMrmBhu3nCOYk= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0/go.mod h1:Mf6O40IAyB9zR/1J8nGDDPirZQQPbYJni8Yisy7NTMc= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/chromedp/cdproto v0.0.0-20250724212937-08a3db8b4327 h1:UQ4AU+BGti3Sy/aLU8KVseYKNALcX9UXY6DfpwQ6J8E= +github.com/chromedp/cdproto v0.0.0-20250724212937-08a3db8b4327/go.mod h1:NItd7aLkcfOA/dcMXvl8p1u+lQqioRMq/SqDp71Pb/k= +github.com/chromedp/chromedp v0.14.2 h1:r3b/WtwM50RsBZHMUm9fsNhhzRStTHrKdr2zmwbZSzM= +github.com/chromedp/chromedp v0.14.2/go.mod h1:rHzAv60xDE7VNy/MYtTUrYreSc0ujt2O1/C3bzctYBo= +github.com/chromedp/sysutil v1.1.0 h1:PUFNv5EcprjqXZD9nJb9b/c9ibAbxiYo4exNWZyipwM= +github.com/chromedp/sysutil v1.1.0/go.mod h1:WiThHUdltqCNKGc4gaU50XgYjwjYIhKWoHGPTUfWTJ8= +github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f h1:Y8xYupdHxryycyPlc9Y+bSQAYZnetRJ70VMVKm5CKI0= +github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f/go.mod h1:HlzOvOjVBOfTGSRXRyY0OiCS/3J1akRGQQpRO/7zyF4= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= +github.com/envoyproxy/go-control-plane v0.13.5-0.20251024222203-75eaa193e329 h1:K+fnvUM0VZ7ZFJf0n4L/BRlnsb9pL/GuDG6FqaH+PwM= +github.com/envoyproxy/go-control-plane v0.13.5-0.20251024222203-75eaa193e329/go.mod h1:Alz8LEClvR7xKsrq3qzoc4N0guvVNSS8KmSChGYr9hs= +github.com/envoyproxy/go-control-plane/envoy v1.35.0 h1:ixjkELDE+ru6idPxcHLj8LBVc2bFP7iBytj353BoHUo= +github.com/envoyproxy/go-control-plane/envoy v1.35.0/go.mod h1:09qwbGVuSWWAyN5t/b3iyVfz5+z8QWGrzkoqm/8SbEs= +github.com/envoyproxy/go-control-plane/ratelimit v0.1.0 h1:/G9QYbddjL25KvtKTv3an9lx6VBE2cnb8wp1vEGNYGI= +github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4= +github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8= +github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU= +github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/go-jose/go-jose/v4 v4.1.3 h1:CVLmWDhDVRa6Mi/IgCgaopNosCaHz7zrMeF9MlZRkrs= +github.com/go-jose/go-jose/v4 v4.1.3/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= +github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2 h1:iizUGZ9pEquQS5jTGkh4AqeeHCMbfbjeb0zMt0aEFzs= +github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2/go.mod h1:TiCD2a1pcmjd7YnhGH0f/zKNcCD06B029pHhzV23c2M= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +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/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU= +github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM= +github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og= +github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= +github.com/gobwas/ws v1.4.0 h1:CTaoG1tojrh4ucGPcoJFiAQUAsEWekEWvLy7GsVNqGs= +github.com/gobwas/ws v1.4.0/go.mod h1:G3gNqMNtPppf5XUz7O4shetPpcZ1VJ7zt18dlUeakrc= +github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= +github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/martian/v3 v3.3.3 h1:DIhPTQrbPkgs2yJYdXU/eNACCG5DVQjySNRNlflZ9Fc= +github.com/google/martian/v3 v3.3.3/go.mod h1:iEPrYcgCF7jA9OtScMFQyAlZZ4YXTKEtJ1E6RWzmBA0= +github.com/google/s2a-go v0.1.9 h1:LGD7gtMgezd8a/Xak7mEWL0PjoTQFvpRudN895yqKW0= +github.com/google/s2a-go v0.1.9/go.mod h1:YA0Ei2ZQL3acow2O62kdp9UlnvMmU7kA6Eutn0dXayM= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/googleapis/enterprise-certificate-proxy v0.3.11 h1:vAe81Msw+8tKUxi2Dqh/NZMz7475yUvmRIkXr4oN2ao= +github.com/googleapis/enterprise-certificate-proxy v0.3.11/go.mod h1:RFV7MUdlb7AgEq2v7FmMCfeSMCllAzWxFgRdusoGks8= +github.com/googleapis/gax-go/v2 v2.17.0 h1:RksgfBpxqff0EZkDWYuz9q/uWsTVz+kf43LsZ1J6SMc= +github.com/googleapis/gax-go/v2 v2.17.0/go.mod h1:mzaqghpQp4JDh3HvADwrat+6M3MOIDp5YKHhb9PAgDY= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4= +github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= +github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80 h1:6Yzfa6GP0rIo/kULo2bwGEkFvCePZ3qHDDTC3/J9Swo= +github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80/go.mod h1:imJHygn/1yfhB7XSJJKlFZKl/J+dCPAknuiaGOshXAs= +github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde h1:x0TT0RDC7UhAVbbWWBzr41ElhJx5tXPWkIHA2HWPRuw= +github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0= +github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo= +github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/posthog/posthog-go v1.10.0 h1:wfoy7Jfb4LigCoHYyMZoiJmmEoCLOkSaYfDxM/NtCqY= +github.com/posthog/posthog-go v1.10.0/go.mod h1:wB3/9Q7d9gGb1P/yf/Wri9VBlbP8oA8z++prRzL5OcY= +github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs= +github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0= +github.com/spiffe/go-spiffe/v2 v2.6.0 h1:l+DolpxNWYgruGQVV0xsfeya3CsC7m8iBzDnMpsbLuo= +github.com/spiffe/go-spiffe/v2 v2.6.0/go.mod h1:gm2SeUoMZEtpnzPNs2Csc0D/gX33k1xIx7lEzqblHEs= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= +github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/detectors/gcp v1.38.0 h1:ZoYbqX7OaA/TAikspPl3ozPI6iY6LiIY9I8cUfm+pJs= +go.opentelemetry.io/contrib/detectors/gcp v1.38.0/go.mod h1:SU+iU7nu5ud4oCb3LQOhIZ3nRLj6FNVrKgtflbaf2ts= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0 h1:YH4g8lQroajqUwWbq/tr2QX1JFmEXaDLgG+ew9bLMWo= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0/go.mod h1:fvPi2qXDqFs8M4B4fmJhE92TyQs9Ydjlg3RvfUp+NbQ= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q= +go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48= +go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8= +go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.39.0 h1:5gn2urDL/FBnK8OkCfD1j3/ER79rUuTYmCvlXBKeYL8= +go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.39.0/go.mod h1:0fBG6ZJxhqByfFZDwSwpZGzJU671HkwpWaNe2t4VUPI= +go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0= +go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs= +go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18= +go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE= +go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= +go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= +go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= +go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +golang.org/x/crypto v0.47.0 h1:V6e3FRj+n4dbpw86FJ8Fv7XVOql7TEwpHapKoMJ/GO8= +golang.org/x/crypto v0.47.0/go.mod h1:ff3Y9VzzKbwSSEzWqJsJVBnWmRwRSHt/6Op5n9bQc4A= +golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= +golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= +golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= +golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= +golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE= +golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8= +golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= +golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= +gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= +gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +google.golang.org/api v0.265.0 h1:FZvfUdI8nfmuNrE34aOWFPmLC+qRBEiNm3JdivTvAAU= +google.golang.org/api v0.265.0/go.mod h1:uAvfEl3SLUj/7n6k+lJutcswVojHPp2Sp08jWCu8hLY= +google.golang.org/genproto v0.0.0-20260128011058-8636f8732409 h1:VQZ/yAbAtjkHgH80teYd2em3xtIkkHd7ZhqfH2N9CsM= +google.golang.org/genproto v0.0.0-20260128011058-8636f8732409/go.mod h1:rxKD3IEILWEu3P44seeNOAwZN4SaoKaQ/2eTg4mM6EM= +google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20 h1:7ei4lp52gK1uSejlA8AZl5AJjeLUOHBQscRQZUgAcu0= +google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20/go.mod h1:ZdbssH/1SOVnjnDlXzxDHK2MCidiqXtbYccJNzNYPEE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260203192932-546029d2fa20 h1:Jr5R2J6F6qWyzINc+4AM8t5pfUz6beZpHp678GNrMbE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260203192932-546029d2fa20/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= +google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc= +google.golang.org/grpc v1.78.0/go.mod h1:I47qjTo4OKbMkjA/aOOwxDIiPSBofUtQUI5EfpWvW7U= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +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/packages/content-fetch-go/internal/analytics/analytics.go b/packages/content-fetch-go/internal/analytics/analytics.go new file mode 100644 index 000000000..e4ea392aa --- /dev/null +++ b/packages/content-fetch-go/internal/analytics/analytics.go @@ -0,0 +1,73 @@ +// Package analytics wraps PostHog event capture, matching the analytics.ts behaviour. +package analytics + +import ( + "log" + + "github.com/omnivore-app/omnivore/content-fetch-go/internal/config" + "github.com/posthog/posthog-go" +) + +// Client sends analytics events to PostHog. +type Client struct { + ph posthog.Client + cfg *config.Config +} + +// Event carries data for an analytics capture call. +type Event struct { + Result string // "success" or "failure" + URL string + Source string + TotalTime int64 // milliseconds + ErrorMessage string +} + +// New creates a PostHog analytics client. +func New(cfg *config.Config) *Client { + ph, err := posthog.NewWithConfig(cfg.PostHogAPIKey, posthog.Config{}) + if err != nil { + log.Printf("Failed to create PostHog client: %v", err) + return &Client{cfg: cfg} + } + return &Client{ph: ph, cfg: cfg} +} + +// Capture sends a content_fetch_result event. +// Only failure events are sent when SEND_ANALYTICS is set, matching the TS behaviour. +func (c *Client) Capture(userIDs []string, ev Event) { + if c.ph == nil { + return + } + if !c.cfg.SendAnalytics || ev.Result != "failure" { + return + } + + for _, uid := range userIDs { + props := posthog.NewProperties(). + Set("url", ev.URL). + Set("source", ev.Source). + Set("totalTime", ev.TotalTime). + Set("env", c.cfg.APIEnv) + if ev.ErrorMessage != "" { + props.Set("errorMessage", ev.ErrorMessage) + } + + if err := c.ph.Enqueue(posthog.Capture{ + DistinctId: uid, + Event: "content_fetch_" + ev.Result, + Properties: props, + }); err != nil { + log.Printf("PostHog enqueue error: %v", err) + } + } +} + +// Close flushes and closes the PostHog client. +func (c *Client) Close() { + if c.ph != nil { + if err := c.ph.Close(); err != nil { + log.Printf("PostHog close error: %v", err) + } + } +} diff --git a/packages/content-fetch-go/internal/browser/browser.go b/packages/content-fetch-go/internal/browser/browser.go new file mode 100644 index 000000000..a33e93b75 --- /dev/null +++ b/packages/content-fetch-go/internal/browser/browser.go @@ -0,0 +1,106 @@ +// Package browser manages a shared headless Chromium instance via chromedp. +package browser + +import ( + "context" + "fmt" + "log" + "sync" + + "github.com/chromedp/chromedp" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/config" +) + +// Browser wraps a persistent chromedp browser allocator. +type Browser struct { + cfg *config.Config + allocCtx context.Context + allocCancel context.CancelFunc + mu sync.Mutex +} + +func New(cfg *config.Config) *Browser { + return &Browser{cfg: cfg} +} + +// allocatorOpts returns the chromedp ExecAllocator options matching the original Puppeteer args. +func (b *Browser) allocatorOpts() []chromedp.ExecAllocatorOption { + opts := append(chromedp.DefaultExecAllocatorOptions[:], + chromedp.Flag("autoplay-policy", "user-gesture-required"), + chromedp.Flag("disable-component-update", true), + chromedp.Flag("disable-domain-reliability", true), + chromedp.Flag("disable-print-preview", true), + chromedp.Flag("disable-setuid-sandbox", true), + chromedp.Flag("disable-speech-api", true), + chromedp.Flag("enable-features", "SharedArrayBuffer"), + chromedp.Flag("hide-scrollbars", true), + chromedp.Flag("mute-audio", true), + chromedp.Flag("no-default-browser-check", true), + chromedp.Flag("no-pings", true), + chromedp.Flag("no-sandbox", true), + chromedp.Flag("no-zygote", true), + chromedp.Flag("disable-extensions", true), + chromedp.Flag("disable-dev-shm-usage", true), + chromedp.Flag("no-first-run", true), + chromedp.Flag("disable-background-networking", true), + chromedp.Flag("disable-gpu", true), + chromedp.Flag("disable-software-rasterizer", true), + chromedp.Flag("ignore-certificate-errors", true), + chromedp.WindowSize(1920, 1080), + chromedp.Headless, + ) + + if b.cfg.ChromiumPath != "" && !b.cfg.UseFirefox { + opts = append(opts, chromedp.ExecPath(b.cfg.ChromiumPath)) + } + + return opts +} + +// getAllocator returns the shared browser allocator context, creating it if needed. +func (b *Browser) getAllocator() (context.Context, error) { + b.mu.Lock() + defer b.mu.Unlock() + + if b.allocCtx != nil { + // Check if still alive + select { + case <-b.allocCtx.Done(): + // Allocator died, recreate + b.allocCtx = nil + b.allocCancel = nil + default: + return b.allocCtx, nil + } + } + + log.Println("Starting chromedp browser allocator") + ctx, cancel := chromedp.NewExecAllocator(context.Background(), b.allocatorOpts()...) + b.allocCtx = ctx + b.allocCancel = cancel + return ctx, nil +} + +// NewContext creates a new browser tab context using the shared allocator. +func (b *Browser) NewContext() (context.Context, context.CancelFunc, error) { + allocCtx, err := b.getAllocator() + if err != nil { + return nil, nil, fmt.Errorf("get allocator: %w", err) + } + + ctx, cancel := chromedp.NewContext(allocCtx) + return ctx, cancel, nil +} + +// Close shuts down the browser. +func (b *Browser) Close() { + b.mu.Lock() + defer b.mu.Unlock() + + if b.allocCancel != nil { + b.allocCancel() + b.allocCancel = nil + b.allocCtx = nil + log.Println("Browser closed") + } +} diff --git a/packages/content-fetch-go/internal/bullmq/bullmq.go b/packages/content-fetch-go/internal/bullmq/bullmq.go new file mode 100644 index 000000000..63e61df5e --- /dev/null +++ b/packages/content-fetch-go/internal/bullmq/bullmq.go @@ -0,0 +1,371 @@ +// Package queue implements BullMQ-compatible Redis queue operations. +// +// BullMQ v5 stores jobs as Redis hashes at bull:{queue}:{jobId} +// and manages state via lists/sorted-sets: +// - bull:{queue}:wait LIST (pending, LIFO insertion LPUSH, LPOP consumption) +// - bull:{queue}:active LIST (currently processing) +// - bull:{queue}:completed ZSET (done, score=timestamp) +// - bull:{queue}:failed ZSET (failed, score=timestamp) +// - bull:{queue}:prioritized ZSET (priority queue, score encodes priority+counter) +// - bull:{queue}:id STRING (atomic ID counter) +// +// Workers use a Lua moveToActive script. For the consumer side we replicate +// a simplified version that is compatible with existing BullMQ producers and +// consumers (i.e. jobs added here can be consumed by the TS worker, and vice-versa). +package bullmq + +import ( + "context" + "encoding/json" + "fmt" + "log" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + bullPrefix = "bull" + + // Queue names + ContentFetchQueue = "omnivore-content-fetch-queue" + BackendQueue = "omnivore-backend-queue" + + // Job names + SavePageJob = "save-page" + + // removeOnComplete/removeOnFail ages (seconds) – match TS defaults + completeAge = 3600 + failAge = 86400 +) + +// queueKey returns the base key prefix for a BullMQ queue. +func queueKey(queueName string) string { + return fmt.Sprintf("%s:%s", bullPrefix, queueName) +} + +func waitKey(q string) string { return queueKey(q) + ":wait" } +func activeKey(q string) string { return queueKey(q) + ":active" } +func completedKey(q string) string { return queueKey(q) + ":completed" } +func failedKey(q string) string { return queueKey(q) + ":failed" } +func idKey(q string) string { return queueKey(q) + ":id" } +func metaKey(q string) string { return queueKey(q) + ":meta" } +func jobKey(q, id string) string { return queueKey(q) + ":" + id } +func prioritizedKey(q string) string { return queueKey(q) + ":prioritized" } +func eventsKey(q string) string { return queueKey(q) + ":events" } + +// JobOpts mirrors BullMQ BulkJobOptions. +type JobOpts struct { + Attempts int `json:"attempts"` + Priority int `json:"priority"` + Backoff BackoffOpt `json:"backoff"` + // removeOnComplete and removeOnFail are not stored in opts hash but handled by cleanup +} + +type BackoffOpt struct { + Type string `json:"type"` + Delay int `json:"delay"` +} + +// RawJob is a job as stored in Redis. +type RawJob struct { + ID string + Name string + Data json.RawMessage + Opts JobOpts + Timestamp int64 + AttemptsMade int +} + +// 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() + if err != nil { + return "", err + } + return strconv.FormatInt(id, 10), nil +} + +// AddJobOpts carries parameters for adding a single job. +type AddJobOpts struct { + Name string + Data interface{} + Opts JobOpts +} + +// 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 { + for _, j := range jobs { + jobID, err := nextJobID(ctx, rdb, queueName) + if err != nil { + return fmt.Errorf("get next job id: %w", err) + } + + dataBytes, err := json.Marshal(j.Data) + if err != nil { + return fmt.Errorf("marshal job data: %w", err) + } + + optsBytes, err := json.Marshal(j.Opts) + if err != nil { + return fmt.Errorf("marshal job opts: %w", err) + } + + now := time.Now().UnixMilli() + key := jobKey(queueName, jobID) + + // Store the job hash + pipe := rdb.Pipeline() + pipe.HSet(ctx, key, + "name", j.Name, + "data", string(dataBytes), + "opts", string(optsBytes), + "timestamp", now, + "attemptsMade", 0, + "attemptsStarted", 0, + "stalledCounter", 0, + ) + + if j.Opts.Priority > 0 { + // Priority jobs go into the prioritized sorted set. + // BullMQ score = priority * 0x100000000 + counter (counter increments per priority) + // For simplicity we use: score = priority * 1e12 + now which preserves ordering. + score := float64(j.Opts.Priority)*1e12 + float64(now) + pipe.ZAdd(ctx, prioritizedKey(queueName), redis.Z{ + Score: score, + Member: jobID, + }) + } else { + pipe.LPush(ctx, waitKey(queueName), jobID) + } + + // Publish event for BullMQ dashboard/metrics compatibility + eventPayload, _ := json.Marshal(map[string]interface{}{ + "jobId": jobID, + "prev": "waiting", + }) + pipe.XAdd(ctx, &redis.XAddArgs{ + Stream: eventsKey(queueName), + Values: map[string]interface{}{ + "event": "waiting", + "data": string(eventPayload), + }, + MaxLen: 10000, + }) + + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("add job %s: %w", jobID, err) + } + + log.Printf("Queued job %s/%s id=%s", queueName, j.Name, jobID) + } + return nil +} + +// moveToActive implements a simplified version of BullMQ's moveToActive Lua script. +// It atomically moves the next job from wait (or prioritized) → active. +// Returns the job ID and whether a job was found. +// +// The Lua script below is compatible with BullMQ v5 in that jobs added by the +// TS BullMQ library can be consumed here and vice-versa. It is intentionally +// simplified – it does not handle stalled job detection or rate limiting, which +// are handled by the BullMQ scheduler in the TS layer if both run simultaneously. +var moveToActiveScript = redis.NewScript(` +local waitKey = KEYS[1] +local prioritizedKey = KEYS[2] +local activeKey = KEYS[3] +local jobKeyPrefix = ARGV[1] -- e.g. "bull:omnivore-content-fetch-queue:" + +-- Try prioritized first (lowest score = highest priority) +local jobId = redis.call("ZPOPMIN", prioritizedKey, 1) +if #jobId > 0 then + jobId = jobId[1] +else + -- Fall back to regular wait list (RPOP = FIFO from the end, matching BullMQ) + jobId = redis.call("RPOP", waitKey) +end + +if jobId == false or jobId == nil then + return nil +end + +-- Move to active +redis.call("LPUSH", activeKey, jobId) + +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) { + keys := []string{ + waitKey(queueName), + prioritizedKey(queueName), + activeKey(queueName), + } + prefix := queueKey(queueName) + ":" + + result, err := moveToActiveScript.Run(ctx, rdb, keys, prefix).Result() + if err == redis.Nil { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("moveToActive: %w", err) + } + + jobID, ok := result.(string) + if !ok || jobID == "" { + return nil, nil + } + + return getJob(ctx, rdb, queueName, jobID) +} + +func getJob(ctx context.Context, rdb *redis.Client, queueName, jobID string) (*RawJob, error) { + fields, err := rdb.HGetAll(ctx, jobKey(queueName, jobID)).Result() + if err != nil { + return nil, fmt.Errorf("hgetall job %s: %w", jobID, err) + } + if len(fields) == 0 { + return nil, fmt.Errorf("job %s not found", jobID) + } + + var opts JobOpts + if o := fields["opts"]; o != "" { + _ = json.Unmarshal([]byte(o), &opts) + } + + var attempts int + if a := fields["attemptsMade"]; a != "" { + attempts, _ = strconv.Atoi(a) + } + + return &RawJob{ + ID: jobID, + Name: fields["name"], + Data: json.RawMessage(fields["data"]), + Opts: opts, + AttemptsMade: attempts, + }, nil +} + +// CompleteJob moves a job from active to completed. +func CompleteJob(ctx context.Context, rdb *redis.Client, queueName, jobID string) error { + now := time.Now().UnixMilli() + pipe := rdb.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) + // Schedule cleanup (approximate removeOnComplete age) + pipe.Expire(ctx, jobKey(queueName, jobID), completeAge*time.Second) + _, err := pipe.Exec(ctx) + return err +} + +// 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 { + now := time.Now().UnixMilli() + + // Increment attemptsMade + newAttempts, err := rdb.HIncrBy(ctx, jobKey(queueName, jobID), "attemptsMade", 1).Result() + if err != nil { + return err + } + + if int(newAttempts) < opts.Attempts { + // Retry: calculate exponential backoff delay + delay := exponentialDelay(opts.Backoff.Delay, int(newAttempts)-1) + retryAt := now + int64(delay) + + pipe := rdb.Pipeline() + pipe.LRem(ctx, activeKey(queueName), 0, jobID) + pipe.ZAdd(ctx, fmt.Sprintf("%s:delayed", queueKey(queueName)), redis.Z{ + Score: float64(retryAt), + Member: jobID, + }) + pipe.HSet(ctx, jobKey(queueName, jobID), "failedReason", reason) + _, err = pipe.Exec(ctx) + return err + } + + // Max attempts reached → move to failed + pipe := rdb.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) + pipe.Expire(ctx, jobKey(queueName, jobID), failAge*time.Second) + _, err = pipe.Exec(ctx) + return err +} + +// exponentialDelay returns the next retry delay in milliseconds (matches BullMQ's exponential). +func exponentialDelay(baseDelay, attempt int) int { + d := baseDelay + for i := 0; i < attempt; i++ { + d *= 2 + } + return d +} + +// 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() + activeCmd := pipe.LLen(ctx, activeKey(queueName)) + failedCmd := pipe.ZCard(ctx, failedKey(queueName)) + completedCmd := pipe.ZCard(ctx, completedKey(queueName)) + prioritizedCmd := pipe.ZCard(ctx, prioritizedKey(queueName)) + waitCmd := pipe.LLen(ctx, waitKey(queueName)) + + if _, err := pipe.Exec(ctx); err != nil && err != redis.Nil { + return nil, err + } + + return map[string]int64{ + "active": activeCmd.Val(), + "failed": failedCmd.Val(), + "completed": completedCmd.Val(), + "prioritized": prioritizedCmd.Val() + waitCmd.Val(), + }, nil +} + +// 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) { + // Check both prioritized zset and wait list + results, err := rdb.ZRangeWithScores(ctx, prioritizedKey(queueName), 0, 0).Result() + if err != nil { + return 0, err + } + + 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() + if err == nil { + if tsMs, err := strconv.ParseInt(ts, 10, 64); err == nil { + return float64(time.Now().UnixMilli()-tsMs) / 1000.0, nil + } + } + } + + // Fall back to wait list + waitIDs, err := rdb.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() + if err != nil { + return 0, nil + } + tsMs, err := strconv.ParseInt(ts, 10, 64) + if err != nil { + return 0, nil + } + return float64(time.Now().UnixMilli()-tsMs) / 1000.0, nil +} + +// 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() +} diff --git a/packages/content-fetch-go/internal/config/config.go b/packages/content-fetch-go/internal/config/config.go new file mode 100644 index 000000000..464298f09 --- /dev/null +++ b/packages/content-fetch-go/internal/config/config.go @@ -0,0 +1,92 @@ +package config + +import ( + "os" + "strconv" +) + +type Config struct { + // HTTP + Port int + VerificationToken string + + // Redis - cache + RedisURL string + RedisCert string + + // Redis - queue (BullMQ) + MQRedisURL string + MQRedisCert string + + // Browser + ChromiumPath string + FirefoxPath string + UseFirefox bool + LaunchHeadless bool + + // GCS + GCSUploadBucket string + GCSKeyFilePath string + SkipUploadOriginal bool + + // Analytics (PostHog) + PostHogAPIKey string + SendAnalytics bool + APIEnv string + + // Import metrics + ImporterMetricsCollectorURL string + JWTSecret string + + // Domain blocking + MaxFeedFetchFailures int +} + +func Load() *Config { + cfg := &Config{ + Port: envInt("PORT", 3002), + VerificationToken: os.Getenv("VERIFICATION_TOKEN"), + + RedisURL: os.Getenv("REDIS_URL"), + RedisCert: os.Getenv("REDIS_CERT"), + + MQRedisURL: os.Getenv("MQ_REDIS_URL"), + MQRedisCert: os.Getenv("MQ_REDIS_CERT"), + + ChromiumPath: envDefault("CHROMIUM_PATH", "/usr/bin/chromium"), + FirefoxPath: envDefault("FIREFOX_PATH", "/usr/bin/firefox"), + UseFirefox: os.Getenv("USE_FIREFOX") == "true", + LaunchHeadless: os.Getenv("LAUNCH_HEADLESS") == "true", + + GCSUploadBucket: envDefault("GCS_UPLOAD_BUCKET", "omnivore-files"), + GCSKeyFilePath: os.Getenv("GCS_UPLOAD_SA_KEY_FILE_PATH"), + SkipUploadOriginal: os.Getenv("SKIP_UPLOAD_ORIGINAL") == "true", + + PostHogAPIKey: envDefault("POSTHOG_API_KEY", "test"), + SendAnalytics: os.Getenv("SEND_ANALYTICS") != "", + APIEnv: os.Getenv("API_ENV"), + + ImporterMetricsCollectorURL: os.Getenv("IMPORTER_METRICS_COLLECTOR_URL"), + JWTSecret: os.Getenv("JWT_SECRET"), + + MaxFeedFetchFailures: envInt("MAX_FEED_FETCH_FAILURES", 10), + } + + return cfg +} + +func envDefault(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +func envInt(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return def +} diff --git a/packages/content-fetch-go/internal/fetch/fetch.go b/packages/content-fetch-go/internal/fetch/fetch.go new file mode 100644 index 000000000..bb079e569 --- /dev/null +++ b/packages/content-fetch-go/internal/fetch/fetch.go @@ -0,0 +1,450 @@ +// Package fetch implements content retrieval using chromedp (Chromium DevTools Protocol). +// It replicates the behaviour of packages/puppeteer-parse with equivalent Go logic. +package fetch + +import ( + "bytes" + "context" + "fmt" + "io" + "log" + "net" + "net/http" + "net/url" + "regexp" + "strings" + "time" + + "github.com/chromedp/cdproto/emulation" + "github.com/chromedp/cdproto/network" + "github.com/chromedp/chromedp" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/browser" +) + +// Result mirrors FetchResult from the TS implementation. +type Result struct { + FinalURL string + Title string + Content string + ContentType string +} + +// nonScriptHosts mirrors NON_SCRIPT_HOSTS – JS disabled for these. +var nonScriptHosts = []string{"medium.com", "fastcompany.com", "fortelabs.com"} + +// allowedContentTypes mirrors ALLOWED_CONTENT_TYPES. +var allowedContentTypes = map[string]bool{ + "text/html": true, + "application/octet-stream": true, + "text/plain": true, + "application/pdf": true, +} + +// noCacheURLs mirrors NO_CACHE_URLS from request_handler.ts (checked externally, listed here for reference). +var NoCacheURLs = map[string]bool{ + "https://deviceandbrowserinfo.com/are_you_a_bot": true, + "https://deviceandbrowserinfo.com/info_device": true, + "https://jacksonh.org": true, +} + +// 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) { + start := time.Now() + log.Printf("content-fetch request url=%s locale=%s timezone=%s", rawURL, locale, timezone) + + parsedURL, err := parseAndValidateURL(rawURL) + if err != nil { + return nil, err + } + targetURL := parsedURL.String() + + // Pre-handle: detect PDFs, images, and other special content types. + preResult, err := preHandle(ctx, targetURL) + if err != nil { + log.Printf("pre-handle warning for %s: %v", targetURL, err) + // Non-fatal; continue with browser fetch + } + + if preResult != nil { + if preResult.URL != "" { + if p, err := parseAndValidateURL(preResult.URL); err == nil { + targetURL = p.String() + } + } + if preResult.ContentType == "application/pdf" || (preResult.Content != "" && preResult.Title != "") { + return &Result{ + FinalURL: targetURL, + Title: preResult.Title, + Content: preResult.Content, + ContentType: preResult.ContentType, + }, nil + } + } + + // Fall through to browser fetch + result, err := retrievePage(ctx, br, targetURL, locale, timezone) + if err != nil { + return nil, err + } + + log.Printf("content-fetch done url=%s duration=%s", targetURL, time.Since(start)) + return result, nil +} + +// preHandleResult carries the output of pre-handle checks. +type preHandleResult struct { + URL string + Title string + Content string + ContentType string +} + +// preHandle performs lightweight checks before launching the browser: +// - Detects PDF URLs (by extension or HEAD request) +// - Detects image content types +// Returns nil if normal browser rendering should proceed. +func preHandle(ctx context.Context, rawURL string) (*preHandleResult, error) { + lower := strings.ToLower(rawURL) + + // PDF by extension + if strings.HasSuffix(lower, ".pdf") || strings.Contains(lower, ".pdf?") { + return &preHandleResult{URL: rawURL, ContentType: "application/pdf"}, nil + } + + // Quick HEAD to detect content-type without full page load + client := &http.Client{Timeout: 5 * time.Second, CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) > 5 { + return fmt.Errorf("too many redirects") + } + return nil + }} + + req, err := http.NewRequestWithContext(ctx, http.MethodHead, rawURL, nil) + if err != nil { + return nil, err + } + req.Header.Set("User-Agent", "Mozilla/5.0 (compatible; Omnivore/1.0)") + + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + ct := strings.ToLower(resp.Header.Get("Content-Type")) + if ct != "" { + ct = strings.SplitN(ct, ";", 2)[0] + ct = strings.TrimSpace(ct) + } + + if ct == "application/pdf" { + return &preHandleResult{URL: resp.Request.URL.String(), ContentType: "application/pdf"}, nil + } + + return nil, nil +} + +// 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() + if err != nil { + return nil, fmt.Errorf("new browser context: %w", err) + } + defer cancel() + + // Timeout for the entire page load + DOM settle + tabCtx, timeoutCancel := context.WithTimeout(tabCtx, 60*time.Second) + defer timeoutCancel() + + result := &Result{FinalURL: targetURL} + + disableJS := shouldDisableJS(targetURL) + + // Track final URL (after redirects) and content-type via network events + var finalURL string + var contentType string + var lastPDFURL string + + chromedp.ListenTarget(tabCtx, func(ev interface{}) { + switch e := ev.(type) { + case *network.EventResponseReceived: + if e.Type == network.ResourceTypeDocument { + ct := strings.ToLower(e.Response.MimeType) + if ct == "application/pdf" { + lastPDFURL = e.Response.URL + } + if finalURL == "" { + finalURL = e.Response.URL + if ct, ok := e.Response.Headers["content-type"].(string); ok { + contentType = ct + } + if contentType == "" { + contentType = e.Response.MimeType + } + } + } + } + }) + + var actions []chromedp.Action + + // Set locale header + if locale != "" { + actions = append(actions, network.SetExtraHTTPHeaders(network.Headers{ + "Accept-Language": locale, + })) + } + + if disableJS { + actions = append(actions, chromedp.ActionFunc(func(ctx context.Context) error { + return emulation.SetScriptExecutionDisabled(true).Do(ctx) + })) + } + + // Navigate + actions = append(actions, chromedp.Navigate(targetURL)) + + // Wait for DOM to settle (equivalent to waitForDOMToSettle) + actions = append(actions, chromedp.ActionFunc(func(ctx context.Context) error { + return waitForDOMSettle(ctx, 5*time.Second, 1*time.Second) + })) + + // Scroll the page + actions = append(actions, chromedp.ActionFunc(func(ctx context.Context) error { + return scrollPage(ctx, 5*time.Second) + })) + + // Capture title + actions = append(actions, chromedp.Title(&result.Title)) + + // Capture and clean DOM + var domContent string + actions = append(actions, chromedp.ActionFunc(func(ctx context.Context) error { + return captureAndCleanDOM(ctx, &domContent) + })) + + if err := chromedp.Run(tabCtx, actions...); err != nil { + if lastPDFURL != "" { + return &Result{FinalURL: lastPDFURL, ContentType: "application/pdf"}, nil + } + return nil, fmt.Errorf("chromedp run: %w", err) + } + + if domContent == "IS_BLOCKED" { + return nil, fmt.Errorf("page is blocked by anti-bot protection") + } + + if finalURL != "" { + result.FinalURL = finalURL + } + result.Content = domContent + result.ContentType = contentType + + return result, nil +} + +// waitForDOMSettle runs a MutationObserver via JS to wait until DOM mutations settle. +// Mirrors waitForDOMToSettle from puppeteer-parse. +func waitForDOMSettle(ctx context.Context, timeout, debounce time.Duration) error { + timeoutMs := int(timeout.Milliseconds()) + debounceMs := int(debounce.Milliseconds()) + + script := fmt.Sprintf(` + new Promise((resolve) => { + const timeoutMs = %d; + const debounceMs = %d; + let debounceTimer; + const mainTimeout = setTimeout(() => { + observer.disconnect(); + resolve(); + }, timeoutMs); + const debouncedResolve = () => { + clearTimeout(debounceTimer); + debounceTimer = setTimeout(() => { + observer.disconnect(); + clearTimeout(mainTimeout); + resolve(); + }, debounceMs); + }; + const observer = new MutationObserver(debouncedResolve); + observer.observe(document.body, { attributes: true, childList: true, subtree: true }); + }) + `, timeoutMs, debounceMs) + + return chromedp.Evaluate(script, nil).Do(ctx) +} + +// scrollPage scrolls the entire page with a timeout, mirroring the Puppeteer scroll logic. +func scrollPage(ctx context.Context, timeout time.Duration) error { + scrollCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + script := ` + new Promise((resolve) => { + let scrollHeight = document.body.scrollHeight; + let totalHeight = 0; + let distance = 500; + let timer = setInterval(() => { + window.scrollBy(0, distance); + totalHeight += distance; + if (totalHeight >= scrollHeight) { + clearInterval(timer); + resolve(true); + } + }, 10); + }) + ` + err := chromedp.Evaluate(script, nil).Do(scrollCtx) + if err != nil && scrollCtx.Err() != nil { + return nil // Timeout during scroll is acceptable + } + return err +} + +// captureAndCleanDOM evaluates JS to clean the DOM and return the outerHTML. +// Mirrors the page.evaluate block in retrieveHtml() from puppeteer-parse/src/index.ts. +func captureAndCleanDOM(ctx context.Context, result *string) error { + script := ` + (function() { + const BI_SRC_REGEXP = /url\("(.+?)"\)/gi; + + Array.from(document.body.getElementsByTagName('*')).forEach((el) => { + const style = window.getComputedStyle(el); + const src = el.getAttribute('src'); + + try { + if (el.tagName && ['img', 'image'].includes(el.tagName.toLowerCase())) { + const filter = style.getPropertyValue('filter'); + if (filter && filter.startsWith('blur')) { + el.parentNode && el.parentNode.removeChild(el); + return; + } + } + } catch(err) {} + + const bgImage = style.getPropertyValue('background-image'); + if (bgImage && !['', 'none'].includes(bgImage)) { + const filter = style.getPropertyValue('filter'); + if (filter && filter.startsWith('blur')) { + el && el.parentNode && el.parentNode.removeChild(el); + } else { + BI_SRC_REGEXP.lastIndex = 0; + const matchedSRC = BI_SRC_REGEXP.exec(bgImage); + BI_SRC_REGEXP.lastIndex = 0; + if (matchedSRC && matchedSRC[1] && !src) { + if (!el.textContent) { + const img = document.createElement('img'); + img.src = matchedSRC[1]; + el && el.parentNode && el.parentNode.replaceChild(img, el); + } + } + } + } + + if (el.tagName === 'IFRAME') { + // Instagram iframe handling omitted (requires cross-frame access unavailable in CDP) + } + }); + + if ( + document.querySelector('[data-translate="managed_checking_msg"]') || + document.getElementById('px-block-form-wrapper') + ) { + return 'IS_BLOCKED'; + } + + return document.documentElement.outerHTML; + })() + ` + return chromedp.Evaluate(script, result).Do(ctx) +} + +// shouldDisableJS returns true for domains where JS should be disabled (mirrors NON_SCRIPT_HOSTS). +func shouldDisableJS(rawURL string) bool { + u, err := url.Parse(rawURL) + if err != nil { + return false + } + hostname := u.Hostname() + for _, host := range nonScriptHosts { + if strings.HasSuffix(hostname, host) { + return true + } + } + return false +} + +// parseAndValidateURL validates and normalises the URL, mirroring validateUrlString + getUrl. +func parseAndValidateURL(rawURL string) (*url.URL, error) { + // Extract first URL if embedded in a string + re := regexp.MustCompile(`(https?://[^\s]+)`) + matches := re.FindStringSubmatch(rawURL) + if len(matches) == 0 { + return nil, fmt.Errorf("no URL found in: %s", rawURL) + } + rawURL = matches[1] + + u, err := url.Parse(rawURL) + if err != nil { + return nil, fmt.Errorf("invalid URL: %w", err) + } + + if u.Scheme != "http" && u.Scheme != "https" { + return nil, fmt.Errorf("invalid URL protocol: %s", u.Scheme) + } + + hostname := u.Hostname() + if hostname == "localhost" || hostname == "0.0.0.0" { + return nil, fmt.Errorf("URL points to localhost") + } + + if isPrivateIP(hostname) { + return nil, fmt.Errorf("URL points to private IP: %s", hostname) + } + + return u, nil +} + +var privateIPRanges = []*regexp.Regexp{ + regexp.MustCompile(`^10\.`), + regexp.MustCompile(`^172\.(1[6-9]|2[0-9]|3[0-1])\.`), + regexp.MustCompile(`^192\.168\.`), +} + +func isPrivateIP(hostname string) bool { + // Try resolving if it's a hostname + ip := net.ParseIP(hostname) + if ip == nil { + return false + } + s := ip.String() + for _, re := range privateIPRanges { + if re.MatchString(s) { + return true + } + } + return false +} + +// FetchPDF downloads a PDF directly and returns its bytes (for GCS upload). +func FetchPDF(ctx context.Context, rawURL string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil) + if err != nil { + return nil, err + } + req.Header.Set("User-Agent", "Mozilla/5.0 (compatible; Omnivore/1.0)") + + client := &http.Client{Timeout: 30 * time.Second} + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var buf bytes.Buffer + if _, err := io.Copy(&buf, resp.Body); err != nil { + return nil, err + } + return buf.Bytes(), nil +} diff --git a/packages/content-fetch-go/internal/gcs/gcs.go b/packages/content-fetch-go/internal/gcs/gcs.go new file mode 100644 index 000000000..ed1dd66ce --- /dev/null +++ b/packages/content-fetch-go/internal/gcs/gcs.go @@ -0,0 +1,73 @@ +// Package gcs handles Google Cloud Storage uploads. +package gcs + +import ( + "context" + "fmt" + "io" + "log" + "strings" + "time" + + "cloud.google.com/go/storage" + "google.golang.org/api/option" +) + +// Client wraps GCS operations. +type Client struct { + bucket string + gcsClient *storage.Client +} + +// New creates a GCS client. If keyFilePath is empty, Application Default Credentials are used. +func New(ctx context.Context, bucketName, keyFilePath string) (*Client, error) { + var opts []option.ClientOption + if keyFilePath != "" { + opts = append(opts, option.WithCredentialsFile(keyFilePath)) + } + + gcsClient, err := storage.NewClient(ctx, opts...) + if err != nil { + return nil, fmt.Errorf("new GCS client: %w", err) + } + + return &Client{bucket: bucketName, gcsClient: gcsClient}, nil +} + +// UploadContent uploads content string to GCS at filePath (non-public, 5s write timeout). +func (c *Client) UploadContent(ctx context.Context, filePath, content string) error { + writeCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + wc := c.gcsClient.Bucket(c.bucket).Object(filePath).NewWriter(writeCtx) + wc.ContentType = "text/html" + + if _, err := io.Copy(wc, strings.NewReader(content)); err != nil { + return fmt.Errorf("write to GCS %s: %w", filePath, err) + } + if err := wc.Close(); err != nil { + return fmt.Errorf("close GCS writer %s: %w", filePath, err) + } + + log.Printf("Original content uploaded to %s", filePath) + return nil +} + +// UploadOriginalContent mirrors uploadOriginalContent() from request_handler.ts. +// It uploads content for each user using the pattern content/{userId}/{libraryItemId}.{timestamp}.original +func (c *Client) UploadOriginalContent(ctx context.Context, users []UserRef, content string, savedTimestamp int64) error { + for _, user := range users { + filePath := fmt.Sprintf("content/%s/%s.%d.original", user.ID, user.LibraryItemID, savedTimestamp) + if err := c.UploadContent(ctx, filePath, content); err != nil { + // Log but don't fail the whole job + log.Printf("Failed to upload original content for user %s: %v", user.ID, err) + } + } + return nil +} + +// UserRef holds the minimal user info needed for uploads. +type UserRef struct { + ID string + LibraryItemID string +} diff --git a/packages/content-fetch-go/internal/handler/handler.go b/packages/content-fetch-go/internal/handler/handler.go new file mode 100644 index 000000000..a82eeb539 --- /dev/null +++ b/packages/content-fetch-go/internal/handler/handler.go @@ -0,0 +1,457 @@ +// Package handler implements processFetchContentJob — the core job processing logic. +// It mirrors packages/content-fetch/src/request_handler.ts exactly. +package handler + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "net/url" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/analytics" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/browser" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/bullmq" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/config" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/fetch" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/gcs" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/redisutil" + "github.com/redis/go-redis/v9" +) + +// UserConfig mirrors the TS UserConfig interface. +type UserConfig struct { + ID string `json:"id"` + LibraryItemID string `json:"libraryItemId"` + Folder *string `json:"folder,omitempty"` +} + +// JobData mirrors the TS JobData interface exactly. +type JobData struct { + URL string `json:"url"` + UserID *string `json:"userId,omitempty"` + SaveRequestID string `json:"saveRequestId"` + State *string `json:"state,omitempty"` + Labels []string `json:"labels,omitempty"` + Source *string `json:"source,omitempty"` + TaskID *string `json:"taskId,omitempty"` + Locale *string `json:"locale,omitempty"` + Timezone *string `json:"timezone,omitempty"` + RSSFeedURL *string `json:"rssFeedUrl,omitempty"` + SavedAt *string `json:"savedAt,omitempty"` + PublishedAt *string `json:"publishedAt,omitempty"` + Folder *string `json:"folder,omitempty"` + Users []UserConfig `json:"users,omitempty"` + Priority string `json:"priority"` // "high" | "low" +} + +// savePageJobData mirrors SavePageJobData from job.ts. +type savePageJobData struct { + UserID string `json:"userId"` + URL string `json:"url"` + FinalURL string `json:"finalUrl"` + ArticleSavingRequestID string `json:"articleSavingRequestId"` + State *string `json:"state,omitempty"` + Labels []string `json:"labels,omitempty"` + Source string `json:"source"` + Folder *string `json:"folder,omitempty"` + RSSFeedURL *string `json:"rssFeedUrl,omitempty"` + SavedAt *string `json:"savedAt,omitempty"` + PublishedAt *string `json:"publishedAt,omitempty"` + TaskID *string `json:"taskId,omitempty"` + Title string `json:"title,omitempty"` + ContentType string `json:"contentType,omitempty"` + CacheKey string `json:"cacheKey,omitempty"` +} + +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, + data *JobData, + attemptsMade int, +) error { + functionStartTime := time.Now() + + // Build user list (mirrors the TS logic) + users := make([]UserConfig, 0) + if data.UserID != nil && *data.UserID != "" { + folder := data.Folder + users = append(users, UserConfig{ + ID: *data.UserID, + LibraryItemID: data.SaveRequestID, + Folder: folder, + }) + } else { + users = data.Users + } + + source := "puppeteer-parse" + if data.Source != nil && *data.Source != "" { + source = *data.Source + } + + locale := "" + if data.Locale != nil { + locale = *data.Locale + } + timezone := "" + if data.Timezone != nil { + timezone = *data.Timezone + } + + logFields := map[string]interface{}{ + "url": data.URL, + "articleSavingRequestId": data.SaveRequestID, + "source": source, + "users": users, + } + log.Printf("Article parsing request %+v", logFields) + + var processErr error + defer func() { + totalTime := time.Since(functionStartTime).Milliseconds() + log.Printf("parse-page result url=%s totalTime=%dms error=%v", data.URL, totalTime, processErr) + + // Analytics + userIDs := make([]string, len(users)) + for i, u := range users { + userIDs[i] = u.ID + } + result := "success" + var errMsg string + if processErr != nil { + result = "failure" + errMsg = processErr.Error() + } + analyticsClient := analytics.New(cfg) + analyticsClient.Capture(userIDs, analytics.Event{ + Result: result, + URL: data.URL, + Source: source, + TotalTime: totalTime, + ErrorMessage: errMsg, + }) + analyticsClient.Close() + + // Import status update on final failure attempt + lastAttempt := attemptsMade+1 >= maxImportAttempts + 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) + } + } + }() + + // Check domain block + domain, err := extractDomain(data.URL) + if err != nil { + processErr = fmt.Errorf("invalid URL: %w", err) + return processErr + } + + blocked, err := isDomainBlocked(ctx, rds, domain, cfg.MaxFeedFetchFailures) + if err != nil { + log.Printf("Error checking domain block: %v", err) + } + if blocked { + log.Printf("Domain is blocked: %s", domain) + // Return nil (not an error) — mirroring the TS behaviour of silently dropping + return nil + } + + // Try cache + cacheKey := buildCacheKey(data.URL, locale, timezone) + fetchResult, err := getCachedResult(ctx, rds, 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) + if err != nil { + _ = incrementDomainFailure(ctx, rds, domain) + processErr = fmt.Errorf("fetchContent: %w", err) + return processErr + } + log.Println("Content fetched successfully") + + // Cache result (skip NO_CACHE_URLS) + if fetchResult.Content != "" && !fetch.NoCacheURLs[data.URL] { + if err := cacheResult(ctx, rds, cacheKey, fetchResult); err != nil { + log.Printf("Cache write error: %v", err) + } + } + } + + savedDate := time.Now() + if data.SavedAt != nil && *data.SavedAt != "" { + if t, err := time.Parse(time.RFC3339, *data.SavedAt); err == nil { + savedDate = t + } + } + + // Upload original content to GCS + if fetchResult.Content != "" && !cfg.SkipUploadOriginal { + gcsClient, err := gcs.New(ctx, cfg.GCSUploadBucket, cfg.GCSKeyFilePath) + if err != nil { + log.Printf("GCS client init error: %v", err) + } else { + refs := make([]gcs.UserRef, len(users)) + for i, u := range users { + refs[i] = gcs.UserRef{ID: u.ID, LibraryItemID: u.LibraryItemID} + } + if err := gcsClient.UploadOriginalContent(ctx, refs, fetchResult.Content, savedDate.UnixMilli()); err != nil { + log.Printf("GCS upload error: %v", err) + } + } + } + + // Build save-page jobs and queue them + savedAtStr := savedDate.Format(time.RFC3339) + savePageJobs := make([]bullmq.AddJobOpts, 0, len(users)) + for _, user := range users { + folder := user.Folder + jobData := savePageJobData{ + UserID: user.ID, + URL: data.URL, + FinalURL: fetchResult.FinalURL, + ArticleSavingRequestID: user.LibraryItemID, + State: data.State, + Labels: data.Labels, + Source: source, + Folder: folder, + RSSFeedURL: data.RSSFeedURL, + SavedAt: &savedAtStr, + PublishedAt: data.PublishedAt, + TaskID: data.TaskID, + Title: fetchResult.Title, + ContentType: fetchResult.ContentType, + CacheKey: cacheKey, + } + + isRSS := data.RSSFeedURL != nil && *data.RSSFeedURL != "" + isImport := data.TaskID != nil && *data.TaskID != "" + + priority := getBullMQPriority(isRSS, isImport, data.Priority) + attempts := getAttempts(isRSS, isImport) + backoffDelay := 2000 + + savePageJobs = append(savePageJobs, bullmq.AddJobOpts{ + Name: bullmq.SavePageJob, + Data: jobData, + Opts: bullmq.JobOpts{ + Attempts: attempts, + Priority: priority, + Backoff: bullmq.BackoffOpt{ + Type: "exponential", + Delay: backoffDelay, + }, + }, + }) + } + + if err := bullmq.AddBulk(ctx, rds.MQClient, bullmq.BackendQueue, savePageJobs); err != nil { + processErr = fmt.Errorf("queue save-page jobs: %w", err) + return processErr + } + + log.Printf("save-page jobs queued: %d", len(savePageJobs)) + return nil +} + +// getBullMQPriority mirrors getPriority() from job.ts. +func getBullMQPriority(isRSS, isImport bool, priority string) int { + if isImport { + return 100 + } + if isRSS { + if priority == "low" { + return 10 + } + return 5 + } + if priority == "low" { + return 5 + } + return 1 +} + +// getAttempts mirrors getAttempts() from job.ts. +func getAttempts(isRSS, isImport bool) int { + if isImport { + return 1 + } + if isRSS { + return 2 + } + return 3 +} + +// buildCacheKey mirrors cacheKey() from request_handler.ts. +func buildCacheKey(rawURL, locale, timezone string) string { + return fmt.Sprintf("fetch-result:%s:%s:%s", rawURL, locale, timezone) +} + +// cachedFetchResult is the JSON structure stored in Redis. +type cachedFetchResult struct { + FinalURL string `json:"finalUrl"` + Title string `json:"title,omitempty"` + Content string `json:"content,omitempty"` + ContentType string `json:"contentType,omitempty"` +} + +// 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() + if err == redis.Nil { + log.Printf("Fetch result not cached: %s", key) + return nil, nil + } + if err != nil { + return nil, err + } + + var cached cachedFetchResult + if err := json.Unmarshal([]byte(val), &cached); err != nil { + log.Printf("Invalid cache entry for key %s: %v", key, err) + return nil, nil + } + + if cached.FinalURL == "" { + return nil, nil + } + + log.Printf("Fetch result is cached: %s", key) + return &fetch.Result{ + FinalURL: cached.FinalURL, + Title: cached.Title, + Content: cached.Content, + ContentType: cached.ContentType, + }, nil +} + +// 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 { + val, err := json.Marshal(cachedFetchResult{ + FinalURL: r.FinalURL, + Title: r.Title, + Content: r.Content, + ContentType: r.ContentType, + }) + if err != nil { + return err + } + return rds.CacheClient.SetNX(ctx, key, string(val), 24*time.Hour).Err() +} + +// failureRedisKey mirrors failureRedisKey() from request_handler.ts. +func failureRedisKey(domain string) string { + return "fetch-failure:" + domain +} + +// isDomainBlocked mirrors isDomainBlocked() from request_handler.ts. +func isDomainBlocked(ctx context.Context, rds *redisutil.RedisDataSource, domain string, maxFailures int) (bool, error) { + blockedDomains := map[string]bool{ + "localhost": true, + "weibo.com": true, + } + if blockedDomains[domain] { + return true, nil + } + + key := failureRedisKey(domain) + val, err := rds.CacheClient.Get(ctx, key).Result() + if err == redis.Nil { + return false, nil + } + if err != nil { + return false, err + } + + var count int + if _, err := fmt.Sscanf(val, "%d", &count); err != nil { + return false, nil + } + + if count > maxFailures { + log.Printf("Domain is blocked (failure count=%d): %s", count, domain) + return true, nil + } + + return false, nil +} + +// incrementDomainFailure mirrors incrementContentFetchFailure() from request_handler.ts. +func incrementDomainFailure(ctx context.Context, rds *redisutil.RedisDataSource, domain string) error { + key := failureRedisKey(domain) + if err := rds.CacheClient.Incr(ctx, key).Err(); err != nil { + return err + } + return rds.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 == "" { + log.Println("JWT_SECRET or IMPORTER_METRICS_COLLECTOR_URL not set, skipping import status update") + return + } + + token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{ + "uid": userID, + }) + tokenStr, err := token.SignedString([]byte(cfg.JWTSecret)) + if err != nil { + log.Printf("Failed to sign JWT: %v", err) + return + } + + status := "failed" + if isImported { + status = "imported" + } + + body, _ := json.Marshal(map[string]string{ + "taskId": taskID, + "status": status, + }) + + reqCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, cfg.ImporterMetricsCollectorURL, bytes.NewReader(body)) + if err != nil { + log.Printf("Failed to create import status request: %v", err) + return + } + req.Header.Set("Authorization", tokenStr) + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + log.Printf("Failed to send import status update: %v", err) + return + } + defer resp.Body.Close() + log.Printf("Import status update sent: status=%s code=%d", status, resp.StatusCode) +} + +// extractDomain extracts the hostname from a URL. +func extractDomain(rawURL string) (string, error) { + u, err := url.Parse(rawURL) + if err != nil { + return "", err + } + return u.Hostname(), nil +} diff --git a/packages/content-fetch-go/internal/queue/worker.go b/packages/content-fetch-go/internal/queue/worker.go new file mode 100644 index 000000000..251d5f494 --- /dev/null +++ b/packages/content-fetch-go/internal/queue/worker.go @@ -0,0 +1,110 @@ +package queue + +import ( + "context" + "encoding/json" + "log" + "sync" + "time" + + "github.com/omnivore-app/omnivore/content-fetch-go/internal/browser" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/bullmq" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/config" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/handler" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/redisutil" +) + +const ( + workerConcurrency = 4 + workerPollInterval = 500 * time.Millisecond +) + +// Worker processes jobs from the content-fetch BullMQ queue. +type Worker struct { + ctx context.Context + cfg *config.Config + rds *redisutil.RedisDataSource + br *browser.Browser + wg sync.WaitGroup + sem chan struct{} +} + +func NewWorker(ctx context.Context, cfg *config.Config, rds *redisutil.RedisDataSource, br *browser.Browser) *Worker { + return &Worker{ + ctx: ctx, + cfg: cfg, + rds: rds, + br: br, + sem: make(chan struct{}, workerConcurrency), + } +} + +func (w *Worker) Start() { + w.wg.Add(1) + go w.run() +} + +func (w *Worker) Wait() { + w.wg.Wait() +} + +func (w *Worker) run() { + defer w.wg.Done() + + log.Println("Queue worker started") + + // Ensure queue meta exists + _ = bullmq.EnsureQueueMeta(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue) + + for { + select { + case <-w.ctx.Done(): + log.Println("Queue worker stopping, draining active slots...") + for i := 0; i < workerConcurrency; i++ { + w.sem <- struct{}{} + } + log.Println("Queue worker stopped") + return + default: + } + + job, err := bullmq.PopJob(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue) + if err != nil { + log.Printf("Error popping job: %v", err) + time.Sleep(workerPollInterval) + continue + } + if job == nil { + time.Sleep(workerPollInterval) + continue + } + + w.sem <- struct{}{} + w.wg.Add(1) + go func(j *bullmq.RawJob) { + defer func() { <-w.sem }() + defer w.wg.Done() + w.processJob(j) + }(job) + } +} + +func (w *Worker) processJob(job *bullmq.RawJob) { + log.Printf("Processing job id=%s name=%s", job.ID, job.Name) + + 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) + return + } + + if err := handler.ProcessFetchContentJob(w.ctx, w.cfg, w.rds, w.br, &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) + return + } + + _ = bullmq.CompleteJob(w.ctx, w.rds.MQClient, bullmq.ContentFetchQueue, job.ID) + log.Printf("Job id=%s completed", job.ID) +} diff --git a/packages/content-fetch-go/internal/redisutil/redisutil.go b/packages/content-fetch-go/internal/redisutil/redisutil.go new file mode 100644 index 000000000..4e3e815d2 --- /dev/null +++ b/packages/content-fetch-go/internal/redisutil/redisutil.go @@ -0,0 +1,88 @@ +package redisutil + +import ( + "context" + "crypto/tls" + "fmt" + "log" + "strings" + "time" + + "github.com/omnivore-app/omnivore/content-fetch-go/internal/config" + "github.com/redis/go-redis/v9" +) + +// RedisDataSource holds two Redis clients: one for caching and one for BullMQ queues. +type RedisDataSource struct { + CacheClient *redis.Client + MQClient *redis.Client +} + +func New(cfg *config.Config) (*RedisDataSource, error) { + cacheClient, err := newClient(cfg.RedisURL, cfg.RedisCert) + if err != nil { + return nil, fmt.Errorf("cache redis: %w", err) + } + + mqClient := cacheClient + if cfg.MQRedisURL != "" { + mqClient, err = newClient(cfg.MQRedisURL, cfg.MQRedisCert) + if err != nil { + return nil, fmt.Errorf("mq redis: %w", err) + } + } + + // Ping both + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := cacheClient.Ping(ctx).Err(); err != nil { + return nil, fmt.Errorf("cache redis ping: %w", err) + } + if mqClient != cacheClient { + if err := mqClient.Ping(ctx).Err(); err != nil { + return nil, fmt.Errorf("mq redis ping: %w", err) + } + } + + log.Println("Redis connected") + return &RedisDataSource{ + CacheClient: cacheClient, + MQClient: mqClient, + }, nil +} + +func newClient(redisURL, cert string) (*redis.Client, error) { + if redisURL == "" { + return nil, fmt.Errorf("redis URL is empty") + } + + opt, err := redis.ParseURL(redisURL) + if err != nil { + return nil, fmt.Errorf("parse redis URL: %w", err) + } + + // TLS with custom cert for rediss:// URLs + if strings.HasPrefix(redisURL, "rediss://") && cert != "" { + opt.TLSConfig = &tls.Config{ + InsecureSkipVerify: true, //nolint:gosec // matches original TS behaviour + RootCAs: nil, + } + } + + // Match ioredis settings from original + opt.DialTimeout = 10 * time.Second + + return redis.NewClient(opt), nil +} + +func (r *RedisDataSource) Shutdown() { + if err := r.CacheClient.Close(); err != nil { + log.Printf("Error closing cache Redis: %v", err) + } + if r.MQClient != r.CacheClient { + if err := r.MQClient.Close(); err != nil { + log.Printf("Error closing MQ Redis: %v", err) + } + } + log.Println("Redis shutdown complete") +} diff --git a/packages/content-fetch-go/internal/server/server.go b/packages/content-fetch-go/internal/server/server.go new file mode 100644 index 000000000..ad7499758 --- /dev/null +++ b/packages/content-fetch-go/internal/server/server.go @@ -0,0 +1,132 @@ +// Package server implements the HTTP endpoints matching the original content-fetch Express app. +package server + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "strconv" + + "github.com/omnivore-app/omnivore/content-fetch-go/internal/browser" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/config" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/handler" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/bullmq" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/redisutil" +) + +// Worker is the minimal interface the server needs from the queue worker. +type Worker interface { + Wait() +} + +type mux struct { + cfg *config.Config + rds *redisutil.RedisDataSource + br *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} + m.HandleFunc("GET /_ah/health", m.health) + m.HandleFunc("GET /lifecycle/prestop", m.prestop) + m.HandleFunc("GET /metrics", m.metrics) + m.HandleFunc("/", m.root) // GET and POST + return m +} + +// health responds to Google Cloud autoscaler health checks. +func (m *mux) health(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) +} + +// prestop implements the Kubernetes/Cloud Run prestop lifecycle hook. +// It signals the worker to stop and waits for in-flight jobs to finish. +func (m *mux) prestop(w http.ResponseWriter, r *http.Request) { + log.Println("Prestop lifecycle hook called.") + // Worker shutdown is handled by the caller (main) via context cancellation; + // here we just wait until it's done. + m.worker.Wait() + log.Println("Worker drained on prestop") + w.WriteHeader(http.StatusOK) +} + +// metrics returns Prometheus-format queue metrics, matching the original /metrics endpoint. +func (m *mux) metrics(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + counts, err := bullmq.GetQueueCounts(ctx, m.rds.MQClient, bullmq.ContentFetchQueue) + if err != nil { + log.Printf("Error getting queue counts: %v", err) + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + + age, err := bullmq.OldestPrioritizedJobAge(ctx, m.rds.MQClient, bullmq.ContentFetchQueue) + if err != nil { + log.Printf("Error getting oldest job age: %v", err) + } + + output := "" + for _, metric := range []string{"active", "failed", "completed", "prioritized"} { + val, _ := counts[metric] + output += fmt.Sprintf("# TYPE omnivore_queue_messages_%s gauge\n", metric) + output += fmt.Sprintf("omnivore_queue_messages_%s{queue=%q} %d\n", metric, bullmq.ContentFetchQueue, val) + } + output += "# TYPE omnivore_queue_messages_oldest_job_age_seconds gauge\n" + output += fmt.Sprintf("omnivore_queue_messages_oldest_job_age_seconds{queue=%q} %s\n", + bullmq.ContentFetchQueue, strconv.FormatFloat(age, 'f', -1, 64)) + + w.Header().Set("Content-Type", "text/plain") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(output)) +} + +// root handles the primary job-processing endpoint (GET or POST /?token=...). +func (m *mux) root(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + + if r.Method != http.MethodGet && r.Method != http.MethodPost { + log.Printf("Request method is not GET or POST: %s", r.Method) + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + + if r.URL.Query().Get("token") != m.cfg.VerificationToken { + log.Println("Query does not include valid token") + w.WriteHeader(http.StatusForbidden) + return + } + + var data handler.JobData + if err := json.NewDecoder(r.Body).Decode(&data); err != nil { + log.Printf("Failed to decode request body: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + + attempt := 0 + if v := r.Header.Get("X-CloudTasks-TaskRetryCount"); v != "" { + if n, err := strconv.Atoi(v); err == nil { + attempt = n + } + } + + if err := handler.ProcessFetchContentJob( + context.Background(), + m.cfg, m.rds, m.br, + &data, attempt, + ); err != nil { + log.Printf("Error fetching content: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + w.WriteHeader(http.StatusOK) +} diff --git a/packages/content-fetch-go/main.go b/packages/content-fetch-go/main.go new file mode 100644 index 000000000..a6d05ae32 --- /dev/null +++ b/packages/content-fetch-go/main.go @@ -0,0 +1,80 @@ +package main + +import ( + "context" + "log" + "net/http" + "os" + "os/signal" + "strconv" + "syscall" + + "github.com/omnivore-app/omnivore/content-fetch-go/internal/browser" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/config" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/queue" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/redisutil" + "github.com/omnivore-app/omnivore/content-fetch-go/internal/server" +) + +func main() { + cfg := config.Load() + + if cfg.VerificationToken == "" { + log.Fatal("VERIFICATION_TOKEN is required") + } + + rds, err := redisutil.New(cfg) + if err != nil { + log.Fatalf("Failed to connect to Redis: %v", err) + } + + br := browser.New(cfg) + + workerCtx, workerCancel := context.WithCancel(context.Background()) + worker := queue.NewWorker(workerCtx, cfg, rds, br) + worker.Start() + + srv := server.New(cfg, rds, br, worker) + + port := cfg.Port + if port == 0 { + port = 3002 + } + addr := ":" + strconv.Itoa(port) + + httpServer := &http.Server{ + Addr: addr, + Handler: srv, + } + + go func() { + log.Printf("Worker started on %s", addr) + if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("HTTP server error: %v", err) + } + }() + + quit := make(chan os.Signal, 1) + signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + sig := <-quit + log.Printf("Received %s, shutting down...", sig) + + // Stop accepting new HTTP requests + if err := httpServer.Shutdown(context.Background()); err != nil { + log.Printf("HTTP server shutdown error: %v", err) + } + log.Println("HTTP server closed") + + // Stop queue worker + workerCancel() + worker.Wait() + log.Println("Worker closed") + + // Close browser + br.Close() + log.Println("Browser closed") + + // Close Redis + rds.Shutdown() + log.Println("Redis connection closed") +}