diff --git a/apps/worker/.gitignore b/apps/worker/.gitignore new file mode 100644 index 0000000..91f29cc --- /dev/null +++ b/apps/worker/.gitignore @@ -0,0 +1,8 @@ +node_modules/ +.wrangler/ +worker-configuration.d.ts +# Build artifact of scripts/bundle-profile.ts (personal profile copies). +src/profile/ +# D1 import chunks generated by scripts/import-postings.ts (personal postings data). +.import/ +.dev.vars diff --git a/apps/worker/package.json b/apps/worker/package.json new file mode 100644 index 0000000..85e5236 --- /dev/null +++ b/apps/worker/package.json @@ -0,0 +1,23 @@ +{ + "name": "jobyssey-scrape", + "private": true, + "type": "module", + "description": "Weekly job scrape as a Cloudflare Workflow: portal search -> dedup against Twenty -> Claude scoring via AI Gateway -> Twenty opportunities.", + "scripts": { + "bundle:profile": "bun run scripts/bundle-profile.ts", + "typecheck": "bun run bundle:profile && wrangler types && tsc --noEmit && tsc --noEmit -p tsconfig.scripts.json", + "dev": "bun run bundle:profile && wrangler dev", + "deploy": "bun run bundle:profile && wrangler deploy", + "import:postings": "bun run scripts/import-postings.ts", + "test": "bun test" + }, + "dependencies": { + "@cloudflare/puppeteer": "^1.0.4" + }, + "devDependencies": { + "@cloudflare/workers-types": "^4", + "@types/bun": "1.3.14", + "typescript": "^5.4.0", + "wrangler": "^4.136.1" + } +} diff --git a/apps/worker/schema.sql b/apps/worker/schema.sql new file mode 100644 index 0000000..b274d28 --- /dev/null +++ b/apps/worker/schema.sql @@ -0,0 +1,47 @@ +-- D1 schema for the jobsdb snapshot (documents/postings/jobs.db). Applied first by +-- scripts/import-postings.ts as .import/0000-schema.sql; idempotent. +-- +-- `postings` mirrors the source table column-for-column (all TEXT). `companies` +-- carries only what src/portals/jobsdb.ts reads. `postings_fts` is a +-- self-contained FTS5 table (not external-content: those need the shadow +-- tables, which `wrangler d1 execute --file` cannot bulk-load sanely), joined +-- on posting_id rather than rowid. + +CREATE TABLE IF NOT EXISTS postings ( + posting_id TEXT PRIMARY KEY, + company_id TEXT, + company_name TEXT, + company_domain TEXT, + title TEXT, + department TEXT, + function TEXT, + seniority TEXT, + posted_date TEXT, + location_text TEXT, + remote_policy_stated TEXT, + location_class TEXT, + location_country TEXT, + salary_min TEXT, + salary_max TEXT, + salary_currency TEXT, + url TEXT, + source_ats TEXT, + source_tier TEXT, + scrape_timestamp TEXT, + classification_rationale TEXT +); + +CREATE TABLE IF NOT EXISTS companies ( + company_id TEXT PRIMARY KEY, + name TEXT, + target_sector TEXT, + crypto_native TEXT, + priority_tier INTEGER, + homepage_domain TEXT +); + +CREATE INDEX IF NOT EXISTS postings_company_id ON postings(company_id); +CREATE INDEX IF NOT EXISTS postings_location_class ON postings(location_class); +CREATE INDEX IF NOT EXISTS postings_seniority ON postings(seniority); + +CREATE VIRTUAL TABLE IF NOT EXISTS postings_fts USING fts5(posting_id UNINDEXED, title, company_name, location_text); diff --git a/apps/worker/scripts/bundle-profile.ts b/apps/worker/scripts/bundle-profile.ts new file mode 100755 index 0000000..73cd61c --- /dev/null +++ b/apps/worker/scripts/bundle-profile.ts @@ -0,0 +1,22 @@ +#!/usr/bin/env bun +// Copies the profile/rubric files the scorer bundles into src/profile/ (gitignored). +// Run before dev/deploy/typecheck (package.json scripts do). The repo files stay +// the single source of truth; this is a build artifact, never edited by hand. +import { mkdirSync, copyFileSync } from "node:fs"; +import { join, resolve } from "node:path"; + +const worker = resolve(import.meta.dir, ".."); +const repo = resolve(worker, "../.."); +const out = join(worker, "src/profile"); + +const files: Record = { + "candidate-profile.md": ".claude/skills/job-application-assistant/01-candidate-profile.md", + "job-evaluation.md": ".claude/skills/job-application-assistant/04-job-evaluation.md", + "search-queries.md": ".claude/skills/job-scraper/search-queries.md", +}; + +mkdirSync(out, { recursive: true }); +for (const [dest, src] of Object.entries(files)) { + copyFileSync(join(repo, src), join(out, dest)); +} +console.log(`bundled ${Object.keys(files).length} profile files into src/profile/`); diff --git a/apps/worker/scripts/import-postings.ts b/apps/worker/scripts/import-postings.ts new file mode 100755 index 0000000..635c30d --- /dev/null +++ b/apps/worker/scripts/import-postings.ts @@ -0,0 +1,165 @@ +#!/usr/bin/env bun +// Turns the local jobsdb snapshot (documents/postings/jobs.db) into D1 import +// chunks: .import/0000-schema.sql (a copy of schema.sql) followed by +// .import/NNNN.sql files of INSERT OR REPLACE statements for companies, +// postings and postings_fts. Apply them in name order with +// `wrangler d1 execute jobyssey-postings --remote --file=`; the script +// prints that loop, or runs it itself with --apply. +// +// bun run scripts/import-postings.ts [path/to/jobs.db] [--apply] +// JOBSDB_PATH=... bun run scripts/import-postings.ts +// +// Runs under Bun only (bun:sqlite); never imported by the Worker. +import { Database } from "bun:sqlite"; +import { existsSync, mkdirSync, readdirSync, rmSync, writeFileSync } from "node:fs"; +import { join, resolve } from "node:path"; + +const ROWS_PER_CHUNK = 2000; +// Multi-row VALUES keep the statement count (and wrangler's round trips) low; +// 50 rows stays well under D1's per-statement size limit for these ~350-byte rows. +const ROWS_PER_STATEMENT = 50; +const DATABASE_NAME = "jobyssey-postings"; + +const worker = resolve(import.meta.dir, ".."); +const repo = resolve(worker, "../.."); +const outDir = join(worker, ".import"); + +const args = process.argv.slice(2); +const apply = args.includes("--apply"); +const dbArg = args.find((a) => !a.startsWith("--")); +const dbPath = resolve(dbArg ?? process.env.JOBSDB_PATH ?? join(repo, "documents/postings/jobs.db")); +if (!existsSync(dbPath)) { + console.error(`database not found at ${dbPath} (pass a path or set JOBSDB_PATH)`); + process.exit(1); +} + +type Cell = string | number | bigint | null; + +function literal(v: Cell): string { + if (v === null) return "NULL"; + if (typeof v === "string") return `'${v.replace(/'/g, "''")}'`; + return String(v); +} + +/** One or more multi-row INSERT OR REPLACE statements for `rows`. */ +function insertStatements(table: string, columns: string[], rows: Cell[][]): string[] { + const head = `INSERT OR REPLACE INTO ${table} (${columns.join(", ")}) VALUES\n`; + const out: string[] = []; + for (let i = 0; i < rows.length; i += ROWS_PER_STATEMENT) { + const values = rows.slice(i, i + ROWS_PER_STATEMENT).map((r) => `(${r.map(literal).join(", ")})`); + out.push(`${head}${values.join(",\n")};\n`); + } + return out; +} + +const POSTING_COLUMNS = [ + "posting_id", + "company_id", + "company_name", + "company_domain", + "title", + "department", + "function", + "seniority", + "posted_date", + "location_text", + "remote_policy_stated", + "location_class", + "location_country", + "salary_min", + "salary_max", + "salary_currency", + "url", + "source_ats", + "source_tier", + "scrape_timestamp", + "classification_rationale", +]; +const COMPANY_COLUMNS = ["company_id", "name", "target_sector", "crypto_native", "priority_tier", "homepage_domain"]; +const FTS_COLUMNS = ["rowid", "posting_id", "title", "company_name", "location_text"]; + +type PostingRow = Record & { rid: number }; +type CompanyRow = Record; + +const db = new Database(dbPath, { readonly: true }); + +// The source `companies` table has no homepage column; postings carry +// company_domain, so a company's domain is its postings' most common one. +const companies = db + .query( + `SELECT c.company_id, c.company_name AS name, c.target_sector, c.crypto_native, c.priority_tier, + (SELECT p.company_domain FROM postings p + WHERE p.company_id = c.company_id AND p.company_domain IS NOT NULL AND p.company_domain != '' + GROUP BY p.company_domain ORDER BY count(*) DESC, p.company_domain LIMIT 1) AS homepage_domain + FROM companies c ORDER BY c.company_id`, + ) + .all(); +// The snapshot re-scrapes postings across days, so posting_id repeats (~87k rows, +// ~76k ids). D1 keys on posting_id: keep each id's latest scrape only, so the +// FTS rows (one per kept source rowid) line up 1:1 with postings. +const postings = db + .query( + `SELECT rid, ${POSTING_COLUMNS.join(", ")} FROM ( + SELECT rowid AS rid, *, row_number() OVER (PARTITION BY posting_id ORDER BY scrape_timestamp DESC, rowid DESC) AS rn + FROM postings + ) WHERE rn = 1 ORDER BY rid`, + ) + .all(); +db.close(); + +rmSync(outDir, { recursive: true, force: true }); +mkdirSync(outDir, { recursive: true }); +const files: string[] = []; +function emit(name: string, statements: string[]): void { + const file = join(outDir, name); + writeFileSync(file, statements.join("\n")); + files.push(file); +} +let chunk = 0; +const chunkName = () => `${String(++chunk).padStart(4, "0")}.sql`; + +emit("0000-schema.sql", [await Bun.file(join(worker, "schema.sql")).text()]); + +for (let i = 0; i < companies.length; i += ROWS_PER_CHUNK) { + const rows = companies.slice(i, i + ROWS_PER_CHUNK).map((c) => COMPANY_COLUMNS.map((col) => c[col] ?? null)); + emit(chunkName(), insertStatements("companies", COMPANY_COLUMNS, rows)); +} + +for (let i = 0; i < postings.length; i += ROWS_PER_CHUNK) { + const slice = postings.slice(i, i + ROWS_PER_CHUNK); + const rows = slice.map((p) => POSTING_COLUMNS.map((col) => p[col] ?? null)); + // The FTS row reuses the source rowid so a re-run replaces instead of duplicating. + const fts = slice.map((p) => [p.rid, p.posting_id ?? null, p.title ?? null, p.company_name ?? null, p.location_text ?? null]); + emit(chunkName(), [...insertStatements("postings", POSTING_COLUMNS, rows), ...insertStatements("postings_fts", FTS_COLUMNS, fts)]); +} + +console.log( + `wrote ${files.length} files to ${outDir}: schema + ${companies.length} companies + ${postings.length} postings (${ROWS_PER_CHUNK}/chunk)`, +); + +const wrangler = join(worker, "node_modules/.bin/wrangler"); +if (!apply) { + console.log(`\nApply in order (from ${worker}):\n`); + console.log(` for f in .import/*.sql; do\n ${wrangler} d1 execute ${DATABASE_NAME} --remote --yes --file="$f" || break\n done\n`); + console.log("or re-run with --apply to do that here."); + process.exit(0); +} + +for (const [n, file] of readdirSync(outDir) + .filter((f) => f.endsWith(".sql")) + .sort() + .map((f) => join(outDir, f)) + .entries()) { + console.log(`[${n + 1}/${files.length}] ${file}`); + const proc = Bun.spawn([wrangler, "d1", "execute", DATABASE_NAME, "--remote", "--yes", `--file=${file}`], { + cwd: worker, + stdout: "inherit", + stderr: "inherit", + }); + const code = await proc.exited; + if (code !== 0) { + console.error(`wrangler exited ${code} on ${file}; fix and re-run --apply (chunks are INSERT OR REPLACE, so replaying is safe)`); + process.exit(code); + } +} +console.log("import complete"); diff --git a/apps/worker/src/browser.ts b/apps/worker/src/browser.ts new file mode 100644 index 0000000..00b1e76 --- /dev/null +++ b/apps/worker/src/browser.ts @@ -0,0 +1,96 @@ +/** + * Browser Rendering helper for the portals that sit behind bot walls (LinkedIn, Wellfound). + * One browser session per portal, one page, static assets blocked, polite pauses between navigations. + */ +import puppeteer, { type HTTPRequest, type Page } from "@cloudflare/puppeteer"; +import type { WorkerEnv } from "./env"; + +const USER_AGENT = + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36"; +const BLOCKED_RESOURCES: Record = { image: true, stylesheet: true, font: true, media: true }; +const KEEP_ALIVE_MS = 60_000; +const DEFAULT_NAV_TIMEOUT_MS = 30_000; +const PAUSE_MIN_MS = 1500; +const PAUSE_MAX_MS = 3000; + +export interface FetchedPage { + /** HTTP status of the main document; 0 when Chrome produced no main response. */ + status: number; + /** URL after redirects. */ + finalUrl: string; + /** Serialised DOM (`page.content()`); fragments come back wrapped in html/body. */ + html: string; +} + +export interface FetchHtmlOptions { + waitUntil?: "domcontentloaded" | "load" | "networkidle0"; + timeoutMs?: number; +} + +export type BlockKind = "ok" | "rate_limited" | "not_found"; + +async function route(req: HTTPRequest): Promise { + try { + if (BLOCKED_RESOURCES[req.resourceType()]) await req.abort(); + else await req.continue(); + } catch (error) { + // Already handled (e.g. page navigated away); nothing to recover. + console.log(JSON.stringify({ event: "browser.route_error", url: req.url(), error: String(error) })); + } +} + +/** Launch a browser, open one prepared page, run `fn`, always close the browser. */ +export async function withBrowser(env: WorkerEnv, fn: (page: Page) => Promise): Promise { + const browser = await puppeteer.launch(env.BROWSER, { keep_alive: KEEP_ALIVE_MS }); + try { + const page = await browser.newPage(); + await page.setUserAgent(USER_AGENT); + await page.setExtraHTTPHeaders({ "Accept-Language": "en-US,en;q=0.9" }); + await page.setRequestInterception(true); + page.on("request", route); + return await fn(page); + } finally { + await browser.close(); + } +} + +/** Navigate and return the main document's status, final URL and serialised HTML. Throws on navigation errors/timeouts. */ +export async function fetchHtml(page: Page, url: string, opts: FetchHtmlOptions = {}): Promise { + const response = await page.goto(url, { + waitUntil: opts.waitUntil ?? "domcontentloaded", + timeout: opts.timeoutMs ?? DEFAULT_NAV_TIMEOUT_MS, + }); + const html = await page.content(); + return { status: response?.status() ?? 0, finalUrl: page.url(), html }; +} + +/** Distinguish a bot wall / rate limit from a genuine miss so callers never report a block as breakage. */ +export function classifyBlock({ status, finalUrl, html }: FetchedPage): BlockKind { + if (status === 429 || status === 999) return "rate_limited"; + if (finalUrl.includes("/authwall") || finalUrl.includes("/checkpoint/") || finalUrl.includes("login")) { + return "rate_limited"; + } + if ( + html.includes("captcha-delivery.com") || + html.includes("datadome") || + html.includes("Access Denied") || + html.includes("Just a moment") + ) { + return "rate_limited"; + } + if (status === 404) return "not_found"; + return "ok"; +} + +/** ES2022 lib (tsconfig) predates Promise.withResolvers; executor form is required here. */ +export function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** Jittered 1.5-3 s pause between navigations so a portal sees human-ish pacing. */ +export function politePause(): Promise { + const buf = new Uint32Array(1); + crypto.getRandomValues(buf); + const unit = (buf[0] ?? 0) / 0x1_0000_0000; + return sleep(PAUSE_MIN_MS + Math.floor(unit * (PAUSE_MAX_MS - PAUSE_MIN_MS))); +} diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts new file mode 100644 index 0000000..85dd037 --- /dev/null +++ b/apps/worker/src/env.ts @@ -0,0 +1,11 @@ +/** `Env` comes from `wrangler types`; secrets are added by `wrangler secret put` and never appear in the generated file. */ +export interface Secrets { + /** Twenty workspace API key (Settings -> API & Webhooks). */ + TWENTY_API_KEY: string; + /** Sent as `x-api-key` through the AI Gateway (no BYOK). */ + ANTHROPIC_API_KEY: string; + /** Bearer token for the manual `POST /run` endpoint. */ + RUN_TOKEN: string; +} + +export type WorkerEnv = Env & Secrets; diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts new file mode 100644 index 0000000..c5614b0 --- /dev/null +++ b/apps/worker/src/index.ts @@ -0,0 +1,80 @@ +import type { WorkerEnv } from "./env"; +import type { Portal } from "./types"; +import { WeeklyScrape, type WeeklyScrapeParams } from "./workflow"; + +export { WeeklyScrape }; + +const PORTALS: Record = { freehire: true, jobsdb: true, linkedin: true, wellfound: true }; + +function isPortal(value: unknown): value is Portal { + return typeof value === "string" && value in PORTALS; +} + +/** Validates the manual-trigger body; returns an error string when the shape is wrong. */ +function parseRunParams(value: unknown): WeeklyScrapeParams | string { + if (value === null || typeof value !== "object") return "body must be a JSON object"; + const params: WeeklyScrapeParams = {}; + if ("runDate" in value) { + if (typeof value.runDate !== "string" || !/^\d{4}-\d{2}-\d{2}$/.test(value.runDate)) return "runDate must be YYYY-MM-DD"; + params.runDate = value.runDate; + } + if ("portals" in value) { + if (!Array.isArray(value.portals) || value.portals.length === 0 || !value.portals.every(isPortal)) { + return `portals must be a non-empty subset of ${Object.keys(PORTALS).join(",")}`; + } + params.portals = value.portals; + } + if ("maxScored" in value) { + if (typeof value.maxScored !== "number" || !Number.isInteger(value.maxScored) || value.maxScored < 0) return "maxScored must be a non-negative integer"; + params.maxScored = value.maxScored; + } + return params; +} + +function timingSafeEqual(a: string, b: string): boolean { + const enc = new TextEncoder(); + const x = enc.encode(a); + const y = enc.encode(b); + if (x.byteLength !== y.byteLength) return false; + return crypto.subtle.timingSafeEqual(x, y); +} + +function authorized(req: Request, env: WorkerEnv): boolean { + const header = req.headers.get("authorization") ?? ""; + const token = header.startsWith("Bearer ") ? header.slice(7) : ""; + return token.length > 0 && timingSafeEqual(token, env.RUN_TOKEN); +} + +/** + * Manual control plane. The weekly run itself is the `schedules` entry on the + * WEEKLY_SCRAPE binding (wrangler.jsonc) - no `scheduled()` handler needed. + * + * POST /run {"portals": ["freehire"], "runDate": "2026-09-22"} -> {"id"} + * GET /runs/:id -> instance status (+ RunSummary in `output` when complete) + * GET /health -> ok + */ +export default { + async fetch(req: Request, env: WorkerEnv): Promise { + const url = new URL(req.url); + + if (req.method === "GET" && url.pathname === "/health") return new Response("ok"); + + if (!authorized(req, env)) return Response.json({ error: "unauthorized" }, { status: 401 }); + + if (req.method === "POST" && url.pathname === "/run") { + const raw: unknown = req.headers.get("content-type")?.includes("application/json") ? await req.json() : {}; + const params = parseRunParams(raw); + if (typeof params === "string") return Response.json({ error: params }, { status: 400 }); + const instance = await env.WEEKLY_SCRAPE.create({ id: `manual-${crypto.randomUUID()}`, params }); + return Response.json({ id: instance.id, status: await instance.status() }, { status: 202 }); + } + + const match = /^\/runs\/([A-Za-z0-9_-]+)$/.exec(url.pathname); + if (req.method === "GET" && match?.[1]) { + const instance = await env.WEEKLY_SCRAPE.get(match[1]); + return Response.json(await instance.status()); + } + + return Response.json({ error: "not found" }, { status: 404 }); + }, +} satisfies ExportedHandler; diff --git a/apps/worker/src/portals/freehire.ts b/apps/worker/src/portals/freehire.ts new file mode 100644 index 0000000..7622d6a --- /dev/null +++ b/apps/worker/src/portals/freehire.ts @@ -0,0 +1,248 @@ +/** + * freehire.me portal: a port of the freehire-search CLI's `search` command + * (.agents/skills/freehire-search/cli/src/commands/search.ts). Same endpoint, + * same query params, same retry policy. The CLI's helpers.ts is not importable + * here (it references `process`), so the pure pieces are copied below with + * their origin noted. + */ +import type { WorkerEnv } from "../env"; +import { JOBAGE_DAYS, type FreehireQuery, type QueryPlan } from "../queries"; +import type { PortalOutcome, Posting } from "../types"; + +const BASE_URL = "https://freehire.me"; +// Origin: SEARCH_PATH in .agents/skills/freehire-search/cli/src/commands/search.ts (not exported). +const SEARCH_PATH = "/api/v1/agent/jobs/search"; +// Origin: UA in .agents/skills/freehire-search/cli/src/helpers.ts (not exported). +const UA = "freehire-search-skill/1.0 (+https://freehire.me)"; +const LIMIT = 20; +const PAGE = 1; +const DAY_MS = 86_400_000; + +// Origin: Envelope in .agents/skills/freehire-search/cli/src/helpers.ts. +interface Envelope { + data: T; + meta?: { total?: number; limit?: number; offset?: number }; + error?: string; +} + +// Origin: FreehireJob in .agents/skills/freehire-search/cli/src/helpers.ts. +interface FreehireJob { + public_slug: string; + source: string; + external_id: string; + url: string; + title: string; + company: string; + company_slug: string; + location: string; + description: string; + skills: string[]; + work_mode?: string; + regions: string[]; + countries: string[]; + cities: string[]; + posted_at: string | null; + created_at: string | null; + // Always present in the wire shape (an unenriched job serializes it as `{}`). + enrichment: { + seniority?: string; + category?: string; + employment_type?: string; + salary_min?: number; + salary_max?: number; + salary_currency?: string; + }; +} + +// Origin: decodeHtmlEntities / numericEntity in .agents/skills/freehire-search/cli/src/helpers.ts. +function decodeHtmlEntities(text: string): string { + return text + .replace(/&/g, "&") + .replace(/</g, "<") + .replace(/>/g, ">") + .replace(/"/g, '"') + .replace(/'/g, "'") + .replace(/'/g, "'") + .replace(/&#(\d+);/g, (_, dec: string) => { + const cp = parseInt(dec, 10); + return cp >= 0 && cp <= 0x10ffff ? String.fromCodePoint(cp) : ""; + }) + .replace(/&#[xX]([0-9a-fA-F]+);/g, (_, hex: string) => { + const cp = parseInt(hex, 16); + return cp >= 0 && cp <= 0x10ffff ? String.fromCodePoint(cp) : ""; + }) + .replace(/ /g, " "); +} + +// Origin: cleanHtml in .agents/skills/freehire-search/cli/src/helpers.ts. +function cleanHtml(html: string | null | undefined): string | null { + if (!html) return null; + const withBreaks = html.replace(/<\s*br\s*\/?>/gi, "\n").replace(/<\/(p|li|ul|ol|div|h\d)>/gi, "\n"); + const text = decodeHtmlEntities(withBreaks.replace(/<[^>]+>/g, " ")) + .replace(/[ \t]+/g, " ") + .replace(/ *\n */g, "\n") + .replace(/\n{3,}/g, "\n\n") + .trim(); + return text || null; +} + +/** 429/5xx survived the CLI's retry policy: inconclusive, reported as `rate_limited`. */ +class RateLimited extends Error { + constructor(status: number, statusText: string) { + super(`freehire API request failed: ${status} ${statusText}`); + this.name = "RateLimited"; + } +} + +/** + * Port of `apiGet` from the CLI helpers with the base URL fixed (the CLI reads + * FREEHIRE_API_URL from process.env). Same policy: retry 429/5xx six times with + * 500ms doubling backoff capped at 8s plus jitter; 404 -> null; connection + * failure fails fast. + */ +async function apiGet(path: string): Promise | null> { + const url = `${BASE_URL}${path}`; + const maxRetries = 6; + let delay = 500; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + let response: Response; + try { + response = await fetch(url, { + headers: { "User-Agent": UA, Accept: "application/json" }, + redirect: "follow", + signal: AbortSignal.timeout(15000), + }); + } catch (e) { + throw new Error( + `could not reach the freehire API at ${BASE_URL} (${e instanceof Error ? e.message : String(e)})`, + ); + } + + if (response.status === 429 || response.status >= 500) { + if (attempt === maxRetries) throw new RateLimited(response.status, response.statusText); + // Executor form: tsconfig lib is ES2022, which has no Promise.withResolvers typing. + await new Promise((r) => setTimeout(r, delay + Math.floor(Math.random() * 500))); + delay = Math.min(delay * 2, 8000); + continue; + } + if (response.status === 404) return null; + + const body = (await response.json().catch(() => null)) as Envelope | null; + if (!response.ok) { + throw new Error(body?.error || `freehire API request failed: ${response.status} ${response.statusText}`); + } + if (!body) throw new Error("freehire API returned an unparseable response body"); + return body; + } + throw new Error("freehire API request failed after retries"); +} + +/** + * Port of `buildQuery` (commands/search.ts) for the weekly run's fixed options: + * limit 20, page 1, keyword search, hydrated text descriptions, posted within + * JOBAGE_DAYS, plus the query's facets as repeated params. + */ +function buildQuery(q: FreehireQuery): URLSearchParams { + const p = new URLSearchParams(); + if (q.q) p.set("q", q.q); + p.set("limit", String(LIMIT)); + p.set("offset", String((PAGE - 1) * LIMIT)); + p.set("semantic_ratio", "0"); // keyword search; the semantic index is opt-in + p.set("include_description", "true"); + p.set("description_format", "text"); + p.set("posted_within_days", String(JOBAGE_DAYS)); + if (q.workMode) p.set("work_mode", q.workMode); + const facets: Array<[string, string[] | undefined]> = [ + ["countries", q.countries], + ["seniority", q.seniority], + ["category", q.category], + ]; + for (const [param, values] of facets) { + for (const value of values ?? []) p.append(param, value); + } + return p; +} + +const REMOTE_BY_WORK_MODE: Record = { + remote: "remote", + hybrid: "hybrid", + onsite: "onsite", + "on-site": "onsite", +}; + +/** + * Field mapping follows `toResult` in the CLI helpers (title fallback, empty + * strings -> null); the search payload never carries seniority (detail only), + * and salary is taken only when enrichment states a USD (or currency-less) max. + */ +function toPosting(job: FreehireJob): Posting { + const e = job.enrichment; + const currency = e.salary_currency?.toUpperCase() ?? ""; + const statedUsdMax = + typeof e.salary_max === "number" && Number.isFinite(e.salary_max) && e.salary_max > 0 && (currency === "" || currency === "USD"); + return { + portal: "freehire", + id: job.public_slug, + title: job.title || "(untitled)", + company: job.company || null, + location: job.location || null, + date: job.posted_at ? job.posted_at.slice(0, 10) : null, + url: job.url, + description: cleanHtml(job.description), + salaryMax: statedUsdMax ? e.salary_max ?? null : null, + remote: job.work_mode ? REMOTE_BY_WORK_MODE[job.work_mode.toLowerCase()] ?? null : null, + seniority: null, + }; +} + +export async function search(_env: WorkerEnv, plan: QueryPlan): Promise { + const now = Date.now(); + const seen = new Set(); + const results: Posting[] = []; + const queries = { attempted: 0, nonEmpty: 0 }; + let succeeded = 0; + let lastError: string | undefined; + + const outcome = (status: PortalOutcome["status"], error?: string): PortalOutcome => ({ + portal: "freehire", + status, + queries, + results, + ...(error !== undefined ? { error } : {}), + }); + + for (const q of plan.freehire) { + queries.attempted++; + try { + const env = await apiGet(`${SEARCH_PATH}?${buildQuery(q).toString()}`); + // A 404 is a missing endpoint, not a missing job (see the CLI); treat it as a hard error. + if (!env) throw new Error(`${SEARCH_PATH} not found on ${BASE_URL}`); + const rows = (env.data ?? []).slice(0, LIMIT); + let fresh = 0; + for (const job of rows) { + const posting = toPosting(job); + // The server already filters by posted_within_days; this guards a dated row that slipped past it. + if (posting.date !== null) { + const t = Date.parse(posting.date); + if (!Number.isNaN(t) && now - t > JOBAGE_DAYS * DAY_MS) continue; + } + fresh++; + if (seen.has(posting.url)) continue; + seen.add(posting.url); + results.push(posting); + } + if (rows.length > 0) queries.nonEmpty++; + succeeded++; + console.log(JSON.stringify({ portal: "freehire", event: "query", q: q.q, hits: rows.length, fresh })); + } catch (e) { + const message = e instanceof Error ? e.message : String(e); + console.log(JSON.stringify({ portal: "freehire", event: "query_failed", q: q.q, error: message })); + if (e instanceof RateLimited) return outcome("rate_limited", message); + lastError = message; + } + } + + if (succeeded === 0 && lastError !== undefined) return outcome("failed", lastError); + return outcome("ok"); +} diff --git a/apps/worker/src/portals/jobsdb.ts b/apps/worker/src/portals/jobsdb.ts new file mode 100644 index 0000000..45c60e7 --- /dev/null +++ b/apps/worker/src/portals/jobsdb.ts @@ -0,0 +1,241 @@ +/** + * jobsdb portal: the local SQLite snapshot (documents/postings/jobs.db) imported + * into D1 by scripts/import-postings.ts. A port of the jobsdb-search CLI's + * `search` command (.agents/skills/jobsdb-search/cli/src/commands/search.ts); + * its helpers.ts is not importable in a Worker (bun:sqlite, node:fs), so the + * pure pieces are copied below with their origin noted. + * + * The snapshot is static: recency is measured against each row's own scrape + * date (JOBSDB_JOBAGE_DAYS), never against today's calendar. + */ +import type { WorkerEnv } from "../env"; +import { JOBSDB_JOBAGE_DAYS, type JobsdbQuery, type QueryPlan } from "../queries"; +import type { PortalOutcome, Posting } from "../types"; + +const PER_QUERY_CAP = 20; +const SQL_LIMIT = 200; +const DAY_MS = 86_400_000; + +// Origin: REMOTE_CLASS / CLASS_LABEL in .agents/skills/jobsdb-search/cli/src/helpers.ts. +const REMOTE_CLASS: Record = { remote: "R1", "remote-global": "R1", "remote-regional": "R2", hybrid: "H", onsite: "O" }; +const CLASS_LABEL: Record = { R1: "Remote (global)", R2: "Remote (regional)", H: "Hybrid", O: "Onsite" }; +/** location_class -> Posting.remote. */ +const REMOTE_BY_CLASS: Record = { R1: "remote", R2: "remote", H: "hybrid", O: "onsite" }; + +// Origin: Row in .agents/skills/jobsdb-search/cli/src/helpers.ts (the columns `p.*, c.*` the search reads). +interface Row { + posting_id: string; + company_id: string | null; + company_name: string | null; + title: string | null; + seniority: string | null; + posted_date: string | null; + location_text: string | null; + location_class: string | null; + location_country: string | null; + salary_min: string | null; + salary_max: string | null; + salary_currency: string | null; + url: string | null; + source_ats: string | null; + scrape_timestamp: string | null; + target_sector: string | null; + crypto_native: string | null; + priority_tier: number | null; +} + +// Origin: Job in .agents/skills/jobsdb-search/cli/src/helpers.ts. +interface Job { + id: string; + title: string | null; + company: string | null; + location: string | null; + date: string | null; + date_approximate: boolean; + url: string | null; + description: null; + remote: string | null; + seniority: string | null; + salary: string | null; + sector: string | null; + crypto_native: boolean; + priority_tier: number | null; + ats: string | null; + snapshot: string | null; +} + +// Origin: resolveDate in .agents/skills/jobsdb-search/cli/src/helpers.ts. +/** "Posted 8 Days Ago" + scrape_timestamp -> ISO date. "30+" is a floor, flagged approximate. */ +function resolveDate(posted: string | null, scraped: string | null): { date: string | null; approximate: boolean } { + if (!posted) return { date: null, approximate: false }; + // greenhouse/ashby/lever store ISO timestamps; workday stores relative text. + if (/^\d{4}-\d{2}-\d{2}/.test(posted)) return { date: posted.slice(0, 10), approximate: false }; + if (!scraped) return { date: null, approximate: false }; + const base = new Date(scraped); + if (isNaN(base.getTime())) return { date: null, approximate: false }; + const m = /(\d+)(\+?)\s*Days?/i.exec(posted); + let days: number; + let approximate = false; + if (/Today/i.test(posted)) days = 0; + else if (/Yesterday/i.test(posted)) days = 1; + else if (m) { + days = Number(m[1]); + approximate = m[2] === "+"; + } else return { date: null, approximate: false }; + base.setUTCDate(base.getUTCDate() - days); + return { date: base.toISOString().slice(0, 10), approximate }; +} + +// Origin: toJob in .agents/skills/jobsdb-search/cli/src/helpers.ts. +function toJob(r: Row): Job { + const { date, approximate } = resolveDate(r.posted_date, r.scrape_timestamp); + const salary = r.salary_min && r.salary_max ? `${r.salary_min}-${r.salary_max} ${r.salary_currency ?? ""}`.trim() : null; + return { + id: r.posting_id, + title: r.title || null, + company: r.company_name || null, + location: r.location_text || null, + date, + date_approximate: approximate, + url: r.url || null, + description: null, + remote: r.location_class ? CLASS_LABEL[r.location_class] ?? null : null, + seniority: r.seniority || null, + salary, + sector: r.target_sector || null, + crypto_native: r.crypto_native === "True", + priority_tier: r.priority_tier ?? null, + ats: r.source_ats || null, + snapshot: r.scrape_timestamp ? r.scrape_timestamp.slice(0, 10) : null, + }; +} + +// Origin: ftsMatch in .agents/skills/jobsdb-search/cli/src/commands/search.ts. +function ftsMatch(q: string): string { + // Quote every token so user text can never be parsed as FTS5 syntax. + return q + .split(/\s+/) + .filter(Boolean) + .map((t) => `"${t.replace(/"/g, '""')}"`) + .join(" "); +} + +/** + * The CLI's search SQL over the D1 schema: the FTS join is on posting_id (the + * D1 table is self-contained, not external-content), and the result set is + * capped in SQL since only the first PER_QUERY_CAP fresh rows are kept. + */ +function buildSql(q: JobsdbQuery): { sql: string; params: Array } { + const where: string[] = []; + const params: Array = []; + let from = "postings p LEFT JOIN companies c ON c.company_id = p.company_id"; + if (q.query) { + from = "postings_fts f JOIN postings p ON p.posting_id = f.posting_id LEFT JOIN companies c ON c.company_id = p.company_id"; + where.push("postings_fts MATCH ?"); + params.push(ftsMatch(q.query)); + } + if (q.company) { + where.push("p.company_name LIKE ?"); + params.push(`%${q.company}%`); + } + if (q.location) { + where.push("p.location_text LIKE ?"); + params.push(`%${q.location}%`); + } + if (q.remote?.length) { + const classes: string[] = []; + for (const r of q.remote) { + const code = REMOTE_CLASS[r.toLowerCase()]; + if (code === undefined) throw new Error(`remote accepts ${Object.keys(REMOTE_CLASS).join(", ")}, got ${r}`); + classes.push(code); + } + where.push(`p.location_class IN (${classes.map(() => "?").join(",")})`); + params.push(...classes); + } + if (q.seniority?.length) { + where.push(`p.seniority IN (${q.seniority.map(() => "?").join(",")})`); + params.push(...q.seniority); + } + if (q.sector?.length) { + where.push(`c.target_sector IN (${q.sector.map(() => "?").join(",")})`); + params.push(...q.sector); + } + if (q.cryptoNative) where.push("c.crypto_native = 'True'"); + if (q.tier !== undefined) { + where.push("c.priority_tier <= ?"); + params.push(q.tier); + } + const sql = `SELECT p.*, c.target_sector, c.crypto_native, c.priority_tier FROM ${from} + ${where.length ? "WHERE " + where.join(" AND ") : ""} + ORDER BY COALESCE(c.priority_tier, 9), p.scrape_timestamp DESC LIMIT ${SQL_LIMIT}`; + return { sql, params }; +} + +/** Numeric salary_max when the row states it in USD (or no currency); the source stores numbers as TEXT. */ +function statedUsdMax(r: Row): number | null { + const currency = (r.salary_currency ?? "").toUpperCase(); + if (currency !== "" && currency !== "USD") return null; + if (!r.salary_max) return null; + const n = Number(r.salary_max); + return Number.isFinite(n) && n > 0 ? n : null; +} + +/** Row + its resolved Job -> Posting; `url` is job.url already narrowed by the caller. */ +function toPosting(r: Row, job: Job, url: string): Posting { + return { + portal: "jobsdb", + id: job.id, + title: job.title ?? "(untitled)", + company: job.company, + location: job.location, + date: job.date, + url, + description: null, + salaryMax: statedUsdMax(r), + remote: r.location_class ? REMOTE_BY_CLASS[r.location_class] ?? null : null, + seniority: job.seniority, + }; +} + +export async function search(env: WorkerEnv, plan: QueryPlan): Promise { + const seen = new Set(); + const results: Posting[] = []; + const queries = { attempted: 0, nonEmpty: 0 }; + let succeeded = 0; + let lastError: string | undefined; + + for (const q of plan.jobsdb) { + queries.attempted++; + try { + const { sql, params } = buildSql(q); + const { results: rows } = await env.POSTINGS.prepare(sql).bind(...params).all(); + let kept = 0; + for (const r of rows) { + if (kept >= PER_QUERY_CAP) break; + const job = toJob(r); + // Recency relative to the row's own scrape snapshot, as the CLI does; undated rows are dropped. + if (!job.date || !job.snapshot) continue; + if ((Date.parse(job.snapshot) - Date.parse(job.date)) / DAY_MS > JOBSDB_JOBAGE_DAYS) continue; + // No url: nothing to dedup against Twenty or link to. + if (job.url === null) continue; + kept++; + if (seen.has(job.url)) continue; + seen.add(job.url); + results.push(toPosting(r, job, job.url)); + } + if (rows.length > 0) queries.nonEmpty++; + succeeded++; + console.log(JSON.stringify({ portal: "jobsdb", event: "query", query: q, rows: rows.length, kept })); + } catch (e) { + lastError = e instanceof Error ? e.message : String(e); + console.log(JSON.stringify({ portal: "jobsdb", event: "query_failed", query: q, error: lastError })); + } + } + + const outcome: PortalOutcome = { portal: "jobsdb", status: "ok", queries, results }; + if (succeeded === 0 && lastError !== undefined) { + outcome.status = "failed"; + outcome.error = lastError; + } + return outcome; +} diff --git a/apps/worker/src/portals/linkedin.ts b/apps/worker/src/portals/linkedin.ts new file mode 100644 index 0000000..c11381b --- /dev/null +++ b/apps/worker/src/portals/linkedin.ts @@ -0,0 +1,312 @@ +/** + * LinkedIn public "jobs-guest" search, fetched through Browser Rendering so the guest endpoint sees a + * real Chrome session. The card parser is copied from + * .agents/skills/linkedin-search/cli/src/helpers.ts (parseJobCards, decodeHtmlEntities, stripTags, clean, + * numericEntity, SEARCH_URL, jobageToTPR, workTypeFlag) and buildUrl from its commands/search.ts; the + * detail parser (DETAIL_URL, extractDivContent, parseJobDetail) from the same helpers.ts. The skill files + * reference `process` and fail noUncheckedIndexedAccess, so they cannot be imported here. + */ +import { classifyBlock, fetchHtml, politePause, withBrowser, type FetchedPage } from "../browser"; +import type { WorkerEnv } from "../env"; +import { JOBAGE_DAYS, type LinkedinQuery, type QueryPlan } from "../queries"; +import type { PortalOutcome, Posting } from "../types"; + +const SEARCH_URL = "https://www.linkedin.com/jobs-guest/jobs/api/seeMoreJobPostings/search"; +const DETAIL_URL = "https://www.linkedin.com/jobs-guest/jobs/api/jobPosting"; +const PER_QUERY_CAP = 20; +/** Detail fetches per hydrate() call: ~2 s pause each keeps a run well inside the Workflow step budget. */ +const HYDRATE_CAP = 60; + +interface JobCard { + id: string; + title: string; + company: string | null; + location: string | null; + date: string | null; + url: string; +} + +function numericEntity(cp: number): string { + return cp >= 0 && cp <= 0x10ffff ? String.fromCodePoint(cp) : ""; +} + +function decodeHtmlEntities(text: string): string { + return text + .replace(/&/g, "&") + .replace(/</g, "<") + .replace(/>/g, ">") + .replace(/"/g, '"') + .replace(/'/g, "'") + .replace(/'/g, "'") + .replace(/&#(\d+);/g, (_, dec: string) => numericEntity(parseInt(dec, 10))) + .replace(/&#[xX]([0-9a-fA-F]+);/g, (_, hex: string) => numericEntity(parseInt(hex, 16))) + .replace(/ /g, " "); +} + +function clean(html: string): string { + return decodeHtmlEntities(html.replace(/<[^>]+>/g, " ").replace(/\s+/g, " ").trim()); +} + +/** First capture group of `re` in `chunk`, or null. */ +function group(chunk: string, re: RegExp): string | null { + const m = chunk.match(re); + return m?.[1] ?? null; +} + +/** Split on the job-posting URN and parse each card independently so one malformed card cannot break the rest. */ +export function parseJobCards(html: string): JobCard[] { + const results: JobCard[] = []; + const chunks = html.split(/data-entity-urn="urn:li:jobPosting:/).slice(1); + for (const chunk of chunks) { + const id = group(chunk, /^(\d+)/); + if (!id) continue; + + const link = group(chunk, /class="base-card__full-link[^"]*"[^>]*href="([^"]+)"/i); + const url = link ? (decodeHtmlEntities(link).split("?")[0] ?? "") : ""; + + let title: string | null = null; + const h3 = group(chunk, /class="base-search-card__title"[^>]*>([\s\S]*?)<\/h3>/i); + if (h3) title = clean(h3); + if (!title) { + const sr = group(chunk, /class="sr-only"[^>]*>([\s\S]*?)<\/span>/i); + if (sr) title = clean(sr); + } + if (!title) continue; + + const sub = group(chunk, /class="base-search-card__subtitle"[^>]*>([\s\S]*?)<\/h4>/i); + const company = sub ? clean(sub) || null : null; + const loc = group(chunk, /class="job-search-card__location"[^>]*>([\s\S]*?)<\/span>/i); + const location = loc ? clean(loc) || null : null; + const date = group(chunk, /class="job-search-card__listdate[^"]*"[^>]*datetime="([^"]+)"/i); + + results.push({ id, title, company, location, date, url: url || `https://www.linkedin.com/jobs/view/${id}` }); + } + return results; +} + +/** Workplace-type flag: on-site=1, remote=2, hybrid=3. */ +const WORK_TYPE_FLAG: Record = { remote: "2", hybrid: "3", onsite: "1" }; + +/** Page 1 of the guest search; parameter order mirrors the CLI's buildUrl. */ +export function buildUrl(q: LinkedinQuery): string { + const params = new URLSearchParams(); + if (q.query) params.set("keywords", q.query); + if (q.location) params.set("location", q.location); + // jobageToTPR: `r${days*86400}` for 0 < days < 9999. + if (JOBAGE_DAYS > 0 && JOBAGE_DAYS < 9999) params.set("f_TPR", `r${JOBAGE_DAYS * 86400}`); + const wt = q.remote ? WORK_TYPE_FLAG[q.remote] : undefined; + if (wt) params.set("f_WT", wt); + params.set("start", "0"); + return `${SEARCH_URL}?${params.toString()}`; +} + +function toPosting(card: JobCard, remote: string | null): Posting { + return { + portal: "linkedin", + id: card.id, + title: card.title, + company: card.company, + location: card.location, + date: card.date, + url: card.url, + description: null, + salaryMax: null, + remote, + seniority: null, + }; +} + +interface JobDetail { + /** Plain text with paragraph breaks preserved; null when the page carried no description block. */ + description: string | null; + /** "Seniority level" criterion verbatim, or null. */ + seniority: string | null; +} + +/** Inner HTML of the first
carrying `className`, tracking nested
depth. */ +function extractDivContent(html: string, className: string): string | null { + const escaped = className.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); + const open = new RegExp(`]*class="[^"]*${escaped}[^"]*"[^>]*>`, "i").exec(html); + if (!open) return null; + const start = open.index + open[0].length; + let i = start; + let depth = 1; + while (depth > 0) { + const nextOpen = html.indexOf("", i); + if (nextClose === -1) return null; + if (nextOpen !== -1 && nextOpen < nextClose) { + depth++; + i = nextOpen + 4; + } else { + depth--; + i = nextClose + 6; + } + } + return html.slice(start, i - 6); +} + +/** + * Detail page -> description + seniority. Unlike the CLI copy, horizontal whitespace is collapsed but the + * newlines substituted for
/block closers survive, so the scorer sees paragraphs and bullets. + */ +export function parseJobDetail(html: string): JobDetail { + let description: string | null = null; + const descHtml = extractDivContent(html, "show-more-less-html__markup") ?? extractDivContent(html, "description__text"); + if (descHtml) { + const withBreaks = descHtml.replace(/<\s*br\s*\/?>/gi, "\n").replace(/<\/(p|li|ul|ol|div|h\d)>/gi, "\n"); + description = + decodeHtmlEntities(withBreaks.replace(/<[^>]+>/g, " ")) + .replace(/[^\S\n]+/g, " ") + .replace(/ ?\n ?/g, "\n") + .replace(/\n{3,}/g, "\n\n") + .trim() || null; + } + + // Job-criteria items: subheader label -> text value. + const criteria: Record = {}; + const itemRe = + /class="description__job-criteria-subheader"[^>]*>([\s\S]*?)<\/h3>[\s\S]*?class="description__job-criteria-text[^"]*"[^>]*>([\s\S]*?)<\/span>/gi; + let cm: RegExpExecArray | null; + while ((cm = itemRe.exec(html)) !== null) { + criteria[clean(cm[1] ?? "").toLowerCase()] = clean(cm[2] ?? ""); + } + + return { description, seniority: criteria["seniority level"] ?? null }; +} + +const REMOTE_PATTERNS: ReadonlyArray = [ + ["remote", /\bremote\b/i], + ["hybrid", /\bhybrid\b/i], + ["onsite", /\b(?:on-?site|in[- ]office)\b/i], +]; + +/** Workplace type only when the text names exactly one of remote / hybrid / on-site; otherwise unknown. */ +export function inferRemote(description: string): string | null { + let found: string | null = null; + for (const [kind, re] of REMOTE_PATTERNS) { + if (!re.test(description)) continue; + if (found !== null) return null; + found = kind; + } + return found; +} + +/** + * Fill description / seniority / remote for LinkedIn postings whose search card carried no body, so scoring + * is not blind. Returns a new array in the input order; everything else passes through untouched. Never + * throws: a bot wall stops further fetches, a single bad page is skipped, at most HYDRATE_CAP pages are read. + */ +export async function hydrate(env: WorkerEnv, postings: Posting[]): Promise { + const out = postings.slice(); + const targets: number[] = []; + for (let i = 0; i < out.length && targets.length < HYDRATE_CAP; i++) { + const p = out[i]; + if (p && p.portal === "linkedin" && p.description === null) targets.push(i); + } + if (targets.length === 0) return out; + + let fetched = 0; + let filled = 0; + try { + await withBrowser(env, async (page) => { + for (const index of targets) { + const posting = out[index]; + if (!posting) continue; + if (fetched > 0) await politePause(); + const url = `${DETAIL_URL}/${posting.id}`; + let pageResult: FetchedPage; + try { + pageResult = await fetchHtml(page, url, { waitUntil: "domcontentloaded" }); + } catch (error) { + console.log(JSON.stringify({ event: "linkedin.detail_failed", url, error: String(error) })); + continue; + } + fetched++; + const block = classifyBlock(pageResult); + if (block === "rate_limited") { + console.log( + JSON.stringify({ + event: "linkedin.detail_rate_limited", + url, + status: pageResult.status, + finalUrl: pageResult.finalUrl, + unhydrated: targets.length - fetched + 1, + }), + ); + return; + } + if (block === "not_found") continue; + + const detail = parseJobDetail(pageResult.html); + if (detail.description === null) continue; + filled++; + out[index] = { + ...posting, + description: detail.description, + seniority: detail.seniority ?? posting.seniority, + remote: posting.remote ?? inferRemote(detail.description), + }; + } + }); + } catch (error) { + // Browser launch/session failure: whatever was hydrated before it is kept. + console.log(JSON.stringify({ event: "linkedin.hydrate_failed", error: String(error) })); + } + console.log(JSON.stringify({ event: "linkedin.hydrate", targets: targets.length, fetched, filled })); + return out; +} + +export async function search(env: WorkerEnv, plan: QueryPlan): Promise { + const cutoff = new Date(Date.now() - JOBAGE_DAYS * 86_400_000).toISOString().slice(0, 10); + const outcome: PortalOutcome = { portal: "linkedin", status: "ok", queries: { attempted: 0, nonEmpty: 0 }, results: [] }; + const seen: Record = {}; + let ran = 0; + let lastError = ""; + + try { + await withBrowser(env, async (page) => { + for (const q of plan.linkedin) { + if (outcome.queries.attempted > 0) await politePause(); + outcome.queries.attempted++; + const url = buildUrl(q); + let fetched: FetchedPage; + try { + fetched = await fetchHtml(page, url, { waitUntil: "domcontentloaded" }); + } catch (error) { + lastError = String(error); + console.log(JSON.stringify({ event: "linkedin.query_failed", url, error: lastError })); + continue; + } + ran++; + const block = classifyBlock(fetched); + if (block === "rate_limited") { + console.log(JSON.stringify({ event: "linkedin.rate_limited", url, status: fetched.status, finalUrl: fetched.finalUrl })); + outcome.status = "rate_limited"; + return; + } + if (block === "not_found") continue; + + const cards = parseJobCards(fetched.html).slice(0, PER_QUERY_CAP); + if (cards.length > 0) outcome.queries.nonEmpty++; + for (const card of cards) { + if (card.date && card.date < cutoff) continue; + if (seen[card.url]) continue; + seen[card.url] = true; + outcome.results.push(toPosting(card, q.remote ?? null)); + } + console.log(JSON.stringify({ event: "linkedin.query", url, status: fetched.status, cards: cards.length })); + } + }); + } catch (error) { + // Browser launch/session failure: everything collected so far is still valid. + outcome.status = "failed"; + outcome.error = String(error); + console.log(JSON.stringify({ event: "linkedin.failed", error: outcome.error })); + } + if (outcome.status === "ok" && outcome.queries.attempted > 0 && ran === 0) { + outcome.status = "failed"; + outcome.error = `every query failed; last: ${lastError}`; + } + return outcome; +} diff --git a/apps/worker/src/portals/wellfound.ts b/apps/worker/src/portals/wellfound.ts new file mode 100644 index 0000000..a2269fc --- /dev/null +++ b/apps/worker/src/portals/wellfound.ts @@ -0,0 +1,237 @@ +/** + * Wellfound public SEO role pages (/role/r/, /role/l//), fetched through Browser + * Rendering because the site sits behind DataDome. Each page embeds a Next.js __NEXT_DATA__ Apollo cache + * with StartupResult -> highlightedJobListings -> JobListingSearchResult records. + * The parsers are copied from .agents/skills/wellfound-search/cli/src/helpers.ts (extractApollo, + * parseJobCards, decodeEntities, stripHtml, slugify, BASE) and the URL builder + title-token filter from + * its commands/search.ts: the skill files reference `process` and fail noUncheckedIndexedAccess. + */ +import { classifyBlock, fetchHtml, politePause, withBrowser, type FetchedPage } from "../browser"; +import type { WorkerEnv } from "../env"; +import { JOBAGE_DAYS, type QueryPlan, type WellfoundQuery } from "../queries"; +import type { PortalOutcome, Posting } from "../types"; + +const BASE = "https://wellfound.com"; +const PER_QUERY_CAP = 20; +/** Query words too generic to discriminate titles on a role page. */ +const STOP_WORDS: Record = { + senior: true, + sr: true, + staff: true, + lead: true, + principal: true, + junior: true, + engineer: true, + developer: true, + of: true, + and: true, + the: true, +}; + +interface JobCard { + id: string; + title: string; + company: string | null; + location: string | null; + date: string | null; + url: string; + description: string | null; + compensation: string | null; + remote: boolean | null; +} + +type ApolloRecord = Record; + +function slugify(s: string): string { + return s.toLowerCase().trim().replace(/[^a-z0-9]+/g, "-").replace(/^-+|-+$/g, ""); +} + +export function buildUrl(q: WellfoundQuery): string { + const role = slugify(q.query); + return q.remote || !q.location ? `${BASE}/role/r/${role}` : `${BASE}/role/l/${role}/${slugify(q.location)}`; +} + +/** Pull the Apollo cache out of a role page. Null when the page has no data blob (bot wall or layout change). */ +export function extractApollo(html: string): Record | null { + const m = /