grant-outreach-engine/apps/outreach-worker/src/workflows/ingest-pnd-rss.ts

/**
 * Nightly Philanthropy News Digest RFP-feed ingestion workflow.
 *
 * Pulls the PND "RFPs" RSS feed, normalizes each item into a grant row, and
 * upserts them into the outreach schema. Follows the same
 * registration/deps-injection pattern as `ingest-grants.ts` (see that
 * module's doc comment for the full rationale): the scheduled function must
 * ALSO be registered as a plain workflow (both registrations referencing the
 * same function object), and deps are pulled from a module-scope registry
 * populated before `DBOS.launch()` — DBOS serializes workflow args, so
 * closures/functions can't cross that boundary.
 */
import { DBOS, SchedulerMode } from '@dbos-inc/dbos-sdk';
import type { schema } from '@novelpad/outreach-core';
import {
  serverInsertGrants,
  type NewGrantInput,
} from '@novelpad/outreach-core/server';
import type { NodePgDatabase } from 'drizzle-orm/node-postgres';

import { fetchPndRfpFeed, PND_RFP_FEED_URL } from '#~/sources/pnd-rss/client.js';
import { parsePndFeed } from '#~/sources/pnd-rss/normalize.js';

export type OutreachDb = NodePgDatabase<typeof schema>;

export interface IngestPndRssDeps {
  readonly db: OutreachDb;
}

let registeredDeps: IngestPndRssDeps | null = null;

export function setIngestPndRssDeps(deps: IngestPndRssDeps): void {
  registeredDeps = deps;
}

function getIngestPndRssDeps(): IngestPndRssDeps {
  if (registeredDeps == null) {
    throw new Error(
      'IngestPndRssDeps not registered. Call setIngestPndRssDeps() before DBOS.launch().',
    );
  }
  return registeredDeps;
}

/** Fetches the raw PND RFP feed XML, reading the override env var at call time. */
async function fetchPndFeed(): Promise<string> {
  const feedUrl = process.env.PND_RFP_FEED_URL ?? PND_RFP_FEED_URL;
  return fetchPndRfpFeed(feedUrl);
}
const fetchPndFeedStep = DBOS.registerStep(fetchPndFeed, {
  name: 'fetchPndFeed',
  retriesAllowed: true,
  maxAttempts: 3,
});

// Upsert in bounded batches so a large feed doesn't land as one oversized
// `INSERT ... ON CONFLICT` statement.
const UPSERT_BATCH_SIZE = 100;

async function upsertPndGrants(
  db: OutreachDb,
  grants: ReadonlyArray<NewGrantInput>,
): Promise<number> {
  if (grants.length === 0) return 0;
  for (let i = 0; i < grants.length; i += UPSERT_BATCH_SIZE) {
    const batch = grants.slice(i, i + UPSERT_BATCH_SIZE);
    await serverInsertGrants(db, batch);
  }
  return grants.length;
}
const upsertPndGrantsStep = DBOS.registerStep(upsertPndGrants, {
  name: 'upsertPndGrants',
  retriesAllowed: true,
  maxAttempts: 3,
});

async function runIngestPndRss(): Promise<void> {
  const { db } = getIngestPndRssDeps();

  const xml = await fetchPndFeedStep();
  const grants = parsePndFeed(xml);
  const upserted = await upsertPndGrantsStep(db, grants);

  console.log(
    `[ingest-pnd-rss] parsed ${grants.length} item(s), upserted ${upserted} grant(s)`,
  );
}

const g = globalThis as unknown as {
  __outreachIngestPndRssRegistered?: boolean;
};

if (!g.__outreachIngestPndRssRegistered) {
  g.__outreachIngestPndRssRegistered = true;

  const ingestPndRss = async (_scheduledTime: Date, _startedAt: Date) => {
    try {
      await runIngestPndRss();
    } catch (err) {
      console.error('[ingest-pnd-rss] pass failed:', err);
      throw err;
    }
  };

  // Must be registered as BOTH a workflow and a scheduled function,
  // referencing the same function object — see module doc comment.
  DBOS.registerWorkflow(ingestPndRss, { name: 'ingestPndRss' });
  DBOS.registerScheduled(ingestPndRss, {
    crontab: '30 3 * * *',
    name: 'ingestPndRss',
    mode: SchedulerMode.ExactlyOncePerInterval,
  });
}