`src/worker/index.ts`

67 lines
import { prisma } from "@/lib/db";
import { runResearch } from "@/pipeline/research";
import { ollamaReachable } from "@/lib/ai";

// Research worker. Two modes:
//   • REDIS_URL set   → BullMQ worker consuming a "research" queue.
//   • REDIS_URL unset → direct poll loop over PENDING listings (fine at this scale).

const QUEUE = "research";

async function processListing(listingId: string) {
  await prisma.listing.update({ where: { id: listingId }, data: { researchStatus: "QUEUED" } });
  try {
    const r = await runResearch(listingId, { useAI: true, writeIdentity: true });
    console.log(`✓ researched ${listingId} — net ${r.expectedNetProfit.toFixed(0)}, ROI ${(r.roi * 100).toFixed(0)}%`);
  } catch (e) {
    await prisma.listing.update({ where: { id: listingId }, data: { researchStatus: "FAILED" } });
    await prisma.listingEvent.create({
      data: { listingId, type: "RESEARCH_FAILED", message: (e as Error).message },
    });
    console.error(`✗ ${listingId}: ${(e as Error).message}`);
  }
}

async function pendingIds(): Promise<string[]> {
  const rows = await prisma.listing.findMany({
    where: { researchStatus: { in: ["PENDING", "QUEUED"] } },
    select: { id: true },
    take: 25,
  });
  return rows.map((r) => r.id);
}

async function main() {
  const local = await ollamaReachable();
  console.log(`GovArbitrage worker starting. Local AI (Ollama) reachable: ${local}`);

  if (process.env.REDIS_URL) {
    // BullMQ mode (optional dependency).
    const { Worker, Queue } = await import("bullmq");
    const connection = { url: process.env.REDIS_URL };
    const queue = new Queue(QUEUE, { connection });
    // Enqueue any currently-pending listings.
    for (const id of await pendingIds()) await queue.add("research", { listingId: id });
    const worker = new Worker(
      QUEUE,
      async (job) => processListing(job.data.listingId as string),
      { connection, concurrency: 2 },
    );
    worker.on("completed", (job) => console.log(`job ${job.id} done`));
    console.log("BullMQ worker online.");
  } else {
    // Direct poll loop.
    console.log("No REDIS_URL — running direct poll loop (Ctrl-C to stop).");
    for (;;) {
      const ids = await pendingIds();
      for (const id of ids) await processListing(id);
      await new Promise((r) => setTimeout(r, 5000));
    }
  }
}

main().catch((e) => {
  console.error(e);
  process.exit(1);
});