grant-outreach-engine/apps/outreach-worker/src/run-once.ts

/**
 * Supervised one-off workflow runner.
 *
 *   DATABASE_URL=... node --import tsx/esm src/run-once.ts <workflow> [...]
 *
 * where <workflow> is one or more of: ingestGrants, ingestPndRss,
 * ingestNhdojOrgs, enrichOrgs, expireGrants — or `all` for the standard
 * first-run order (grants → pnd → nhdoj → expire → enrich).
 *
 * Boots exactly like main.ts (same deps injection, same DBOS launch — see
 * that file's boot-order comment), runs the requested workflows through
 * their durable DBOS handles sequentially, then shuts down. Scheduled crons
 * ARE active while this process lives; runs are short enough that this
 * doesn't matter in practice.
 */
import { DBOS } from '@dbos-inc/dbos-sdk';
import { DrizzleDataSource } from '@dbos-inc/drizzle-datasource';
import { schema } from '@novelpad/outreach-core';
import { drizzle } from 'drizzle-orm/node-postgres';
import pg from 'pg';

import {
  runEmbedGrantsNow,
  setEmbedGrantsDeps,
} from './workflows/embed-grants.js';
import { runEnrichOrgsNow, setEnrichOrgsDeps } from './workflows/enrich-orgs.js';
import {
  runExpireGrantsNow,
  setExpireGrantsDeps,
} from './workflows/expire-grants.js';
import {
  runFederalPrecedentNow,
  setFederalPrecedentDeps,
} from './workflows/federal-precedent.js';
import {
  runIngestGrantsNow,
  setIngestGrantsDeps,
} from './workflows/ingest-grants.js';
import {
  runIngest990PfNow,
  setIngest990pfDeps,
} from './workflows/ingest-990pf.js';
import {
  runIngestNhdojOrgsNow,
  setIngestNhdojOrgsDeps,
} from './workflows/ingest-nhdoj-orgs.js';
import {
  runMatchGrantsNow,
  setMatchGrantsDeps,
} from './workflows/match-grants.js';
import {
  runIngestPndRssNow,
  setIngestPndRssDeps,
} from './workflows/ingest-pnd-rss.js';

const RUNNERS: Record<string, () => Promise<void>> = {
  ingestGrants: runIngestGrantsNow,
  ingestPndRss: runIngestPndRssNow,
  ingestNhdojOrgs: runIngestNhdojOrgsNow,
  enrichOrgs: runEnrichOrgsNow,
  expireGrants: runExpireGrantsNow,
  embedGrants: runEmbedGrantsNow,
  matchGrants: runMatchGrantsNow,
  ingest990pf: runIngest990PfNow,
  federalPrecedent: runFederalPrecedentNow,
};

const FIRST_RUN_ORDER = [
  'ingestGrants',
  'ingestPndRss',
  'ingestNhdojOrgs',
  'expireGrants',
  'enrichOrgs',
  'ingest990pf',
  'federalPrecedent',
  'embedGrants',
  'matchGrants',
];

if (process.env.DATABASE_URL == null) {
  throw new Error('run-once: DATABASE_URL is required');
}

const requested = process.argv.slice(2);
const names = requested.includes('all') ? FIRST_RUN_ORDER : requested;
if (names.length === 0 || names.some((n) => RUNNERS[n] == null)) {
  console.error(
    `Usage: run-once.ts <workflow...>\nKnown workflows: ${Object.keys(RUNNERS).join(', ')}, all`,
  );
  process.exit(2);
}

const { Pool } = pg;
const dbosClientConfig = {
  connectionString: process.env.DATABASE_URL,
  application_name: 'helmdocs-outreach-run-once',
} satisfies pg.ClientConfig;
const pool = new Pool({
  ...dbosClientConfig,
  max: Number(process.env.PG_POOL_MAX ?? 20),
});
const db = drizzle(pool, { schema });

async function main() {
  await DrizzleDataSource.initializeDBOSSchema(dbosClientConfig);

  setIngestGrantsDeps({ db });
  setExpireGrantsDeps({ db });
  setIngestPndRssDeps({ db });
  setIngestNhdojOrgsDeps({ db });
  setEnrichOrgsDeps({ db });
  setEmbedGrantsDeps({ db });
  setMatchGrantsDeps({ db });
  setIngest990pfDeps({ db });
  setFederalPrecedentDeps({ db });

  DBOS.setConfig({
    name: 'helmdocs-outreach-worker',
    systemDatabasePool: pool,
    runAdminServer: false,
  });
  await DBOS.launch();

  let failed = false;
  for (const name of names) {
    console.log(`\n=== run-once: ${name} ===`);
    const startedAt = Date.now();
    try {
      await RUNNERS[name]!();
      console.log(
        `=== run-once: ${name} OK in ${((Date.now() - startedAt) / 1000).toFixed(1)}s ===`,
      );
    } catch (err) {
      failed = true;
      console.error(`=== run-once: ${name} FAILED ===`, err);
    }
  }

  // DBOS.shutdown() ends the pool we handed it via systemDatabasePool —
  // a second pool.end() here throws "Called end on pool more than once".
  await DBOS.shutdown();
  process.exit(failed ? 1 : 0);
}

main().catch((err) => {
  console.error('[run-once] fatal:', err);
  process.exit(1);
});