linkme/index.ts

#!/usr/bin/env bun
import { searches } from "./config.ts";
import { fetchJobs, isViableContent, isViableUrl } from "./fetch.ts";
import { fetchDataCareer, fetchStellenBasel, fetchItJobsCh, fetchJobRoom, fetchBaselIntlOrgs, fetchJobBasel, fetchEnglishJobSearch, fetchBaselJobs } from "./fetch-other.ts";

const CSV = "./results/jobs.csv";
const OLD_HEADER = "date_found,posted_at,search,title,company,link";
const HEADER_NO_LOCATION = "date_found,posted_at,search,title,company,link,source";
const HEADER = "date_found,posted_at,search,title,company,link,source,location";

// Migrate CSV: add source column if the old header is present, then location
// (the posting's real location, as parsed off its own card — distinct from
// `search`, which is just the query that found it. See CLAUDE.md.)
const csvExists = await Bun.file(CSV).exists();
if (csvExists) {
  const raw = await Bun.file(CSV).text();
  if (raw.startsWith(OLD_HEADER + "\n") || raw === OLD_HEADER) {
    const lines = raw.split("\n");
    const migrated = [
      HEADER_NO_LOCATION,
      ...lines.slice(1).map((l) => (l.trim() ? l + ',"linkedin.com"' : "")),
    ]
      .join("\n")
      .replace(/\n+$/, "") + "\n";
    await Bun.write(CSV, migrated);
  }
}
{
  const raw = await Bun.file(CSV).text().catch(() => "");
  if (raw.startsWith(HEADER_NO_LOCATION + "\n") || raw === HEADER_NO_LOCATION) {
    const lines = raw.split("\n");
    const migrated = [
      HEADER,
      // Unknown for pre-existing rows — daily.ts falls back to `search` when empty.
      ...lines.slice(1).map((l) => (l.trim() ? l + ',""' : "")),
    ]
      .join("\n")
      .replace(/\n+$/, "") + "\n";
    await Bun.write(CSV, migrated);
  }
}

// Load existing links for deduplication
const existingLinks = new Set<string>();
try {
  const text = await Bun.file(CSV).text();
  for (const line of text.split("\n").slice(1)) {
    const url = line.match(/"(https?:\/\/[^"]+)"/)?.[1];
    if (url) existingLinks.add(url);
  }
} catch {
  // first run
}

const date = new Date().toISOString().slice(0, 10);
const newRows: string[] = [];
let totalLangDropped = 0; // jobs dropped for German/French content — never reach the CSV, so daily.ts can't see them on its own

function csvCell(val: string): string {
  return `"${val.replace(/"/g, '""')}"`;
}

// Every search/source below hits a different host (LinkedIn, job-room.ch,
// jobbasel.ch, itjobs.ch, unjobs.org, ...), so there's no reason they need to
// run one after another — that's what made a full run take 30-40+ minutes.
// Run them concurrently instead, bounded per group so we don't burst LinkedIn
// (unauthenticated scraping, sensitive to anti-bot heuristics — see the
// auth warning in CLAUDE.md) or open too many simultaneous connections to
// the smaller extra sources at once.
//
// `existingLinks` is claimed synchronously (`.add()` right after the `.has()`
// check, before any `await`) rather than after the content-check — since Bun
// is single-threaded and only yields at `await`, that ordering is what makes
// concurrent workers safe: two sources racing on the same link can't both
// slip past the check before either claims it.
//
// Trade-off: each source still logs its own page-by-page progress from
// inside fetch.ts/fetch-other.ts, so concurrent sources' output interleaves
// in the terminal instead of printing in clean sequential blocks. Cutting
// total runtime was the actual complaint; log readability is secondary and
// still traceable (grep the source label).
async function runWithConcurrency<T>(items: T[], limit: number, worker: (item: T) => Promise<void>): Promise<void> {
  const queue = [...items];
  async function runner() {
    while (queue.length) {
      const item = queue.shift()!;
      await worker(item);
    }
  }
  await Promise.all(Array.from({ length: Math.min(limit, items.length) }, runner));
}

async function processLinkedInSearch(config: (typeof searches)[number]): Promise<void> {
  console.log(`\n[${config.label}]`);
  const jobs = await fetchJobs(config);

  if (jobs.length === 0) {
    console.log(`  [${config.label}] No jobs found.`);
    return;
  }

  let added = 0;
  let langDropped = 0;
  for (const job of jobs) {
    if (existingLinks.has(job.link)) continue;
    existingLinks.add(job.link); // claim before the await — see note above

    process.stdout.write(`  [${config.label}] checking content: ${job.title} … `);
    // One fetch, two checks: written in German/French, or written in English
    // but demanding German above B1.
    const viable = await isViableContent(job.id);
    if (!viable.ok) {
      console.log(`dropped (${viable.reason})`);
      langDropped++;
      totalLangDropped++;
      continue;
    }
    console.log("ok");
    newRows.push(
      [date, job.postedAt, config.label, job.title, job.company, job.link, "linkedin.com", job.location]
        .map(csvCell)
        .join(",")
    );
    added++;
  }
  console.log(`  [${config.label}] ${jobs.length} found, ${added} new, ${langDropped} non-English dropped`);
}

const extraSources: Array<{ label: string; fetch: () => ReturnType<typeof fetchDataCareer> }> = [
  { label: "datacareer.ch", fetch: fetchDataCareer },
  { label: "stellen-basel.ch", fetch: fetchStellenBasel },
  { label: "itjobs.ch", fetch: fetchItJobsCh },
  { label: "job-room.ch", fetch: fetchJobRoom },
  { label: "basel intl orgs", fetch: fetchBaselIntlOrgs },
  { label: "jobbasel.ch", fetch: fetchJobBasel },
  { label: "englishjobsearch.ch", fetch: fetchEnglishJobSearch },
  { label: "basel-jobs.ch", fetch: fetchBaselJobs },
];

async function processExtraSource({ label, fetch: fetchFn }: (typeof extraSources)[number]): Promise<void> {
  console.log(`\n[${label}]`);
  let jobs;
  try {
    jobs = await fetchFn();
  } catch (e) {
    console.error(`  [${label}] error: ${e}`);
    return;
  }

  if (jobs.length === 0) {
    console.log(`  [${label}] No jobs found.`);
    return;
  }

  let added = 0;
  let langDropped = 0;
  for (const job of jobs) {
    if (existingLinks.has(job.link)) continue;
    existingLinks.add(job.link); // claim before the await — see note above

    if (!job.contentChecked) {
      process.stdout.write(`  [${label}] checking content: ${job.title} … `);
      const viable = await isViableUrl(job.link);
      if (!viable.ok) {
        console.log(`dropped (${viable.reason})`);
        langDropped++;
        totalLangDropped++;
        continue;
      }
      console.log("ok");
    }
    newRows.push(
      [date, job.postedAt, job.search, job.title, job.company, job.link, job.source, job.location]
        .map(csvCell)
        .join(",")
    );
    added++;
  }
  console.log(`  [${label}] ${jobs.length} found, ${added} new, ${langDropped} non-English dropped`);
}

const LINKEDIN_CONCURRENCY = 3;
const EXTRA_SOURCE_CONCURRENCY = 3;

await Promise.all([
  runWithConcurrency(searches, LINKEDIN_CONCURRENCY, processLinkedInSearch),
  runWithConcurrency(extraSources, EXTRA_SOURCE_CONCURRENCY, processExtraSource),
]);

// Jobs dropped for German/French content never reach the CSV, so this is the
// only record of them — daily.ts reads it back to show a count in the digest
// without listing rejects that were never even parsed into a Row.
const LANG_DROPPED = "./results/.lang-dropped.json";
await Bun.write(LANG_DROPPED, JSON.stringify({ date, count: totalLangDropped }));

if (newRows.length === 0) {
  console.log("\nNo new jobs.");
  process.exit(0);
}

// Append to CSV
const existing = await Bun.file(CSV).exists();
const append = existing
  ? await Bun.file(CSV).text()
  : HEADER + "\n";

await Bun.write(CSV, append + newRows.join("\n") + "\n");

console.log(`\n${newRows.length} new jobs → ${CSV}`);