diff --git a/apps/outreach-worker/.env.example b/apps/outreach-worker/.env.example index 930da0d..8568e20 100644 --- a/apps/outreach-worker/.env.example +++ b/apps/outreach-worker/.env.example @@ -11,3 +11,12 @@ GCP_SERVICE_ACCOUNT_KEY_PATH=./secrets/gcp-service-account.json # Optional: cap the pg Pool size (defaults to 20 — see main.ts). # PG_POOL_MAX=20 + +# NHDOJ Charitable Trusts registry PDF — the URL changes whenever NHDOJ +# republishes (roughly monthly). When unset, the monthly ingestNhdojOrgs +# workflow logs a warning and no-ops instead of failing. +# NHDOJ_REGISTRY_PDF_URL=https://www.doj.nh.gov/.../charitable-trusts-registry.pdf + +# Optional override for the Philanthropy News Digest RFP feed (defaults to +# the public feed URL baked into src/sources/pnd-rss/client.ts). +# PND_RFP_FEED_URL=https://philanthropynewsdigest.org/feeds/rfps.rss diff --git a/apps/outreach-worker/package.json b/apps/outreach-worker/package.json index 28584ba..0afa021 100644 --- a/apps/outreach-worker/package.json +++ b/apps/outreach-worker/package.json @@ -8,7 +8,8 @@ "dev": "node --env-file=.env --env-file-if-exists=.env.local --import tsx/esm ./src/main.ts", "build": "esbuild src/main.ts --bundle --platform=node --format=esm --outfile=build/main.js --packages=external", "start": "node ./build/main.js", - "typecheck": "tsc --noEmit" + "typecheck": "tsc --noEmit", + "test": "vitest run" }, "imports": { "#~/*": "./src/*" @@ -18,6 +19,8 @@ "@dbos-inc/drizzle-datasource": "4.17.6", "@novelpad/outreach-core": "workspace:^", "drizzle-orm": "0.44.6", + "fast-xml-parser": "^4.5.0", + "pdfjs-dist": "^4.10.38", "pg": "8.20.0", "tsx": "^4.19.2" }, @@ -26,6 +29,7 @@ "@types/node": "^22", "@types/pg": "8.20.0", "esbuild": "^0.24.0", - "typescript": "^5.9.3" + "typescript": "^5.9.3", + "vitest": "^3.2.4" } } diff --git a/apps/outreach-worker/src/main.ts b/apps/outreach-worker/src/main.ts index 03d3627..5e68072 100644 --- a/apps/outreach-worker/src/main.ts +++ b/apps/outreach-worker/src/main.ts @@ -30,8 +30,11 @@ import pg from 'pg'; // scheduled-function args, so the `db` handle can't be passed through the // scheduler; it's threaded in via this module-scope registry instead (same // pattern as novelpad-desktop's `setStartDeps`). +import { setEnrichOrgsDeps } from './workflows/enrich-orgs.js'; import { setExpireGrantsDeps } from './workflows/expire-grants.js'; import { setIngestGrantsDeps } from './workflows/ingest-grants.js'; +import { setIngestNhdojOrgsDeps } from './workflows/ingest-nhdoj-orgs.js'; +import { setIngestPndRssDeps } from './workflows/ingest-pnd-rss.js'; if (process.env.DATABASE_URL == null) { throw new Error('outreach-worker: DATABASE_URL is required'); @@ -59,6 +62,9 @@ async function main() { // immediately on an `ExactlyOncePerInterval` catch-up) always has it. setIngestGrantsDeps({ db }); setExpireGrantsDeps({ db }); + setIngestPndRssDeps({ db }); + setIngestNhdojOrgsDeps({ db }); + setEnrichOrgsDeps({ db }); DBOS.setConfig({ name: 'helmdocs-outreach-worker', diff --git a/apps/outreach-worker/src/sources/grants-gov/client.test.ts b/apps/outreach-worker/src/sources/grants-gov/client.test.ts new file mode 100644 index 0000000..0895bc1 --- /dev/null +++ b/apps/outreach-worker/src/sources/grants-gov/client.test.ts @@ -0,0 +1,161 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + fetchOpportunityDetail, + fetchOpportunityDetails, + searchOpportunities, +} from '#~/sources/grants-gov/client.js'; + +function jsonResponse(body: unknown, init?: { readonly ok?: boolean; readonly status?: number }) { + const ok = init?.ok ?? true; + const status = init?.status ?? (ok ? 200 : 500); + return { + ok, + status, + statusText: ok ? 'OK' : 'Error', + json: () => Promise.resolve(body), + text: () => Promise.resolve(JSON.stringify(body)), + } as Response; +} + +describe('searchOpportunities', () => { + it('paginates until hitCount is exhausted', async () => { + const fetchImpl = vi.fn(); + fetchImpl + .mockResolvedValueOnce( + jsonResponse({ + errorcode: 0, + msg: 'success', + data: { + hitCount: 3, + startRecord: 0, + oppHits: [ + { id: '1', title: 'A' }, + { id: '2', title: 'B' }, + ], + }, + }), + ) + .mockResolvedValueOnce( + jsonResponse({ + errorcode: 0, + msg: 'success', + data: { + hitCount: 3, + startRecord: 2, + oppHits: [{ id: '3', title: 'C' }], + }, + }), + ); + + const hits = await searchOpportunities({ rows: 2, fetchImpl }); + + expect(hits.map((h) => h.id)).toEqual(['1', '2', '3']); + expect(fetchImpl).toHaveBeenCalledTimes(2); + + const secondCallBody = JSON.parse( + (fetchImpl.mock.calls[1]?.[1]?.body as string) ?? '{}', + ); + expect(secondCallBody).toMatchObject({ startRecordNum: 2, rows: 1 }); + }); + + it('stops once maxHits is reached without over-fetching', async () => { + const fetchImpl = vi.fn().mockResolvedValueOnce( + jsonResponse({ + errorcode: 0, + msg: 'success', + data: { + hitCount: 500, + startRecord: 0, + oppHits: Array.from({ length: 10 }, (_, i) => ({ + id: String(i), + title: `Opp ${i}`, + })), + }, + }), + ); + + const hits = await searchOpportunities({ rows: 10, maxHits: 10, fetchImpl }); + + expect(hits).toHaveLength(10); + expect(fetchImpl).toHaveBeenCalledTimes(1); + }); + + it('throws with a descriptive message on a non-2xx response', async () => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(jsonResponse({ errorcode: 0, msg: '' }, { ok: false, status: 503 })); + + await expect(searchOpportunities({ fetchImpl })).rejects.toThrow(/503/); + }); + + it('throws with a descriptive message on errorcode !== 0', async () => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(jsonResponse({ errorcode: 1, msg: 'bad request' })); + + await expect(searchOpportunities({ fetchImpl })).rejects.toThrow(/bad request/); + }); +}); + +describe('fetchOpportunityDetail', () => { + it('returns the data envelope on success', async () => { + const fetchImpl = vi.fn().mockResolvedValueOnce( + jsonResponse({ + errorcode: 0, + msg: 'success', + data: { id: '354821', opportunityNumber: null, opportunityTitle: 'X', synopsis: null }, + }), + ); + + const detail = await fetchOpportunityDetail('354821', { fetchImpl }); + expect(detail.id).toBe('354821'); + }); + + it('throws on errorcode !== 0, naming the opportunity id', async () => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(jsonResponse({ errorcode: 2, msg: 'not found' })); + + await expect( + fetchOpportunityDetail('999999', { fetchImpl }), + ).rejects.toThrow(/999999/); + }); +}); + +describe('fetchOpportunityDetails', () => { + it('fetches sequentially and waits delayMs between calls, but not after the last', async () => { + const fetchImpl = vi.fn().mockImplementation((_url: string, init: RequestInit) => { + const body = JSON.parse(init.body as string) as { opportunityId: string }; + return Promise.resolve( + jsonResponse({ + errorcode: 0, + msg: 'success', + data: { id: body.opportunityId, opportunityNumber: null, opportunityTitle: null, synopsis: null }, + }), + ); + }); + const sleepImpl = vi.fn().mockResolvedValue(undefined); + + const details = await fetchOpportunityDetails(['a', 'b', 'c'], { + fetchImpl, + sleepImpl, + delayMs: 250, + }); + + expect(details.map((d) => d.id)).toEqual(['a', 'b', 'c']); + expect(sleepImpl).toHaveBeenCalledTimes(2); + expect(sleepImpl).toHaveBeenCalledWith(250); + }); + + it('propagates a detail failure instead of swallowing it', async () => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(jsonResponse({ errorcode: 0, msg: 'success', data: { id: 'a', opportunityNumber: null, opportunityTitle: null, synopsis: null } })) + .mockResolvedValueOnce(jsonResponse({ errorcode: 5, msg: 'boom' })); + + await expect( + fetchOpportunityDetails(['a', 'b'], { fetchImpl, sleepImpl: vi.fn().mockResolvedValue(undefined) }), + ).rejects.toThrow(/boom/); + }); +}); diff --git a/apps/outreach-worker/src/sources/grants-gov/client.ts b/apps/outreach-worker/src/sources/grants-gov/client.ts new file mode 100644 index 0000000..c5ca279 --- /dev/null +++ b/apps/outreach-worker/src/sources/grants-gov/client.ts @@ -0,0 +1,279 @@ +/** + * Thin fetch client for the public Grants.gov Search2 API. + * + * No API key required. Two endpoints are used: + * + * - `POST /v1/api/search2` — keyword/filter search over posted + * opportunities, paginated via `rows`/`startRecordNum`. Returns summary + * hits only (title, agency, open/close dates) — no synopsis or + * eligibility detail. + * - `POST /v1/api/fetchOpportunity` — full detail for a single opportunity + * by id, including the `synopsis` block (description, applicant types, + * award amounts, response date). + * + * Both endpoints wrap their payload in an `{ errorcode, msg, data }` + * envelope; `errorcode !== 0` (or a non-2xx HTTP status) is treated as a + * hard failure — this client never silently swallows an upstream error, it + * throws with enough context to diagnose from a log line. + */ +import { nextSearchRequest } from '#~/sources/grants-gov/pagination.js'; + +const SEARCH_URL = 'https://api.grants.gov/v1/api/search2'; +const DETAIL_URL = 'https://api.grants.gov/v1/api/fetchOpportunity'; + +// Eligibility codes for nonprofit applicants: 12 = "Nonprofits having a +// 501(c)(3) status with the IRS, other than institutions of higher +// education", 13 = "Nonprofits that do not have a 501(c)(3) status with the +// IRS, other than institutions of higher education". +export const NONPROFIT_ELIGIBILITY_CODES = '12|13'; + +const DEFAULT_PAGE_SIZE = 100; +const DEFAULT_DETAIL_DELAY_MS = 250; + +/** A single opportunity summary as returned by `search2`'s `oppHits`. */ +export interface GrantsGovSearchHit { + readonly id: string; + readonly number: string | null; + readonly title: string; + readonly agencyCode: string | null; + readonly agency: string | null; + readonly openDate: string | null; + readonly closeDate: string | null; + readonly oppStatus: string | null; + readonly docType: string | null; +} + +interface GrantsGovSearchResponseEnvelope { + readonly errorcode: number; + readonly msg: string; + readonly data?: { + readonly hitCount: number; + readonly startRecord: number; + readonly oppHits: ReadonlyArray; + }; +} + +/** One entry of `synopsis.applicantTypes` in the detail response. */ +export interface GrantsGovApplicantType { + readonly id?: number | string | null; + readonly description: string | null; +} + +/** The `synopsis` block of a `fetchOpportunity` detail response. */ +export interface GrantsGovSynopsis { + readonly synopsisDesc: string | null; + readonly applicantTypes: ReadonlyArray | null; + readonly awardCeiling: string | number | null; + readonly awardFloor: string | number | null; + readonly estimatedFunding: string | number | null; + readonly expectedNumberOfAwards: string | number | null; + readonly responseDate: string | null; + readonly postingDate: string | null; + readonly agencyName: string | null; +} + +/** Full detail for a single opportunity, as returned by `fetchOpportunity`. */ +export interface GrantsGovOpportunityDetail { + readonly id: string; + readonly opportunityNumber: string | null; + readonly opportunityTitle: string | null; + readonly agencyDetails?: { + readonly agencyName?: string | null; + readonly agencyCode?: string | null; + } | null; + readonly synopsis: GrantsGovSynopsis | null; +} + +interface GrantsGovDetailResponseEnvelope { + readonly errorcode: number; + readonly msg: string; + readonly data?: GrantsGovOpportunityDetail; +} + +type FetchLike = typeof fetch; + +export interface SearchOpportunitiesOptions { + /** Page size per `search2` request. Defaults to 100. */ + readonly rows?: number; + /** Stop once this many total hits have been collected. Defaults to unbounded. */ + readonly maxHits?: number; + /** Eligibility filter codes, pipe-separated. Defaults to nonprofit codes. */ + readonly eligibilities?: string; + /** Injectable for tests; defaults to the global `fetch`. */ + readonly fetchImpl?: FetchLike; +} + +async function postJson( + url: string, + body: unknown, + fetchImpl: FetchLike, +): Promise { + const res = await fetchImpl(url, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + }); + if (!res.ok) { + const text = await res.text().catch(() => ''); + throw new Error( + `Grants.gov request to ${url} failed: ${res.status} ${res.statusText} — ${text}`, + ); + } + return (await res.json()) as T; +} + +/** Fetches a single `search2` page. Not paginated — see `searchOpportunities`. */ +async function fetchSearchPage(params: { + readonly startRecordNum: number; + readonly rows: number; + readonly eligibilities: string; + readonly fetchImpl: FetchLike; +}): Promise<{ + readonly hitCount: number; + readonly oppHits: ReadonlyArray; +}> { + const { startRecordNum, rows, eligibilities, fetchImpl } = params; + const envelope = await postJson( + SEARCH_URL, + { + rows, + startRecordNum, + oppStatuses: 'posted', + eligibilities, + }, + fetchImpl, + ); + + if (envelope.errorcode !== 0) { + throw new Error( + `Grants.gov search2 returned errorcode ${envelope.errorcode}: ${envelope.msg}`, + ); + } + if (envelope.data == null) { + throw new Error('Grants.gov search2 response is missing its data envelope'); + } + + return { + hitCount: envelope.data.hitCount, + oppHits: envelope.data.oppHits ?? [], + }; +} + +/** + * Searches Grants.gov for open ("posted") opportunities matching the given + * eligibility codes, paginating through `search2` until either every hit + * has been collected or `maxHits` is reached. Throws on any non-2xx + * response or `errorcode !== 0` — callers should not expect a partial + * result on failure. + */ +export async function searchOpportunities( + options: SearchOpportunitiesOptions = {}, +): Promise> { + const { + rows: pageSize = DEFAULT_PAGE_SIZE, + maxHits = Number.POSITIVE_INFINITY, + eligibilities = NONPROFIT_ELIGIBILITY_CODES, + fetchImpl = fetch, + } = options; + + const hits: GrantsGovSearchHit[] = []; + let startRecordNum = 0; + let hitCount: number | null = null; + + for (;;) { + const request = nextSearchRequest({ + startRecordNum, + collected: hits.length, + hitCount, + maxHits, + pageSize, + }); + if (request == null) break; + + const page = await fetchSearchPage({ + startRecordNum: request.startRecordNum, + rows: request.rows, + eligibilities, + fetchImpl, + }); + hitCount = page.hitCount; + + if (page.oppHits.length === 0) break; // defensive: avoid an infinite loop on a stuck cursor + hits.push(...page.oppHits); + startRecordNum += page.oppHits.length; + } + + return hits; +} + +export interface FetchOpportunityDetailOptions { + /** Injectable for tests; defaults to the global `fetch`. */ + readonly fetchImpl?: FetchLike; +} + +/** Fetches full detail (including `synopsis`) for a single opportunity id. */ +export async function fetchOpportunityDetail( + opportunityId: string, + options: FetchOpportunityDetailOptions = {}, +): Promise { + const { fetchImpl = fetch } = options; + + const envelope = await postJson( + DETAIL_URL, + { opportunityId }, + fetchImpl, + ); + + if (envelope.errorcode !== 0) { + throw new Error( + `Grants.gov fetchOpportunity returned errorcode ${envelope.errorcode} for opportunity ${opportunityId}: ${envelope.msg}`, + ); + } + if (envelope.data == null) { + throw new Error( + `Grants.gov fetchOpportunity response for opportunity ${opportunityId} is missing its data envelope`, + ); + } + + return envelope.data; +} + +export interface FetchOpportunityDetailsOptions + extends FetchOpportunityDetailOptions { + /** Delay between successive detail requests, in ms. Defaults to 250. */ + readonly delayMs?: number; + /** Injectable sleep for tests; defaults to a real `setTimeout`. */ + readonly sleepImpl?: (ms: number) => Promise; +} + +function defaultSleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * Fetches detail for a batch of opportunity ids sequentially, with a small + * politeness delay between requests (Grants.gov has no documented rate + * limit, but `fetchOpportunity` is one request per grant and we'd rather + * not hammer it). Order of the returned array matches `opportunityIds`. + */ +export async function fetchOpportunityDetails( + opportunityIds: ReadonlyArray, + options: FetchOpportunityDetailsOptions = {}, +): Promise> { + const { + delayMs = DEFAULT_DETAIL_DELAY_MS, + sleepImpl = defaultSleep, + ...rest + } = options; + + const details: GrantsGovOpportunityDetail[] = []; + for (let i = 0; i < opportunityIds.length; i += 1) { + const opportunityId = opportunityIds[i]; + if (opportunityId == null) continue; + details.push(await fetchOpportunityDetail(opportunityId, rest)); + if (delayMs > 0 && i < opportunityIds.length - 1) { + await sleepImpl(delayMs); + } + } + return details; +} diff --git a/apps/outreach-worker/src/sources/grants-gov/fixtures.ts b/apps/outreach-worker/src/sources/grants-gov/fixtures.ts new file mode 100644 index 0000000..64c8e79 --- /dev/null +++ b/apps/outreach-worker/src/sources/grants-gov/fixtures.ts @@ -0,0 +1,54 @@ +/** + * Realistic Grants.gov wire-shape fixtures for `normalize.test.ts`. Field + * values are representative of what `search2`/`fetchOpportunity` actually + * return, trimmed to the fields we consume. + */ +import type { + GrantsGovOpportunityDetail, + GrantsGovSearchHit, +} from '#~/sources/grants-gov/client.js'; + +export const searchHitFixture: GrantsGovSearchHit = { + id: '354821', + number: 'USDA-NIFA-OASA-011234', + title: 'Community Food Projects Competitive Grant Program', + agencyCode: 'USDA-NIFA', + agency: 'National Institute of Food and Agriculture', + openDate: '01/15/2026', + closeDate: '04/30/2026', + oppStatus: 'posted', + docType: 'synopsis', +}; + +export const opportunityDetailFixture: GrantsGovOpportunityDetail = { + id: '354821', + opportunityNumber: 'USDA-NIFA-OASA-011234', + opportunityTitle: 'Community Food Projects Competitive Grant Program', + agencyDetails: { + agencyName: 'National Institute of Food and Agriculture', + agencyCode: 'USDA-NIFA', + }, + synopsis: { + synopsisDesc: + '

The Community Food Projects (CFP) program supports community-based efforts to increase food security.

Applicants must demonstrate community partnership.

', + applicantTypes: [ + { + id: 25, + description: + 'Nonprofits having a 501(c)(3) status with the IRS, other than institutions of higher education', + }, + { + id: 13, + description: + 'Nonprofits that do not have a 501(c)(3) status with the IRS, other than institutions of higher education', + }, + ], + awardCeiling: '$500,000', + awardFloor: '$10,000', + estimatedFunding: '$9,000,000', + expectedNumberOfAwards: '18', + responseDate: '04/30/2026', + postingDate: '01/15/2026', + agencyName: 'National Institute of Food and Agriculture', + }, +}; diff --git a/apps/outreach-worker/src/sources/grants-gov/normalize.test.ts b/apps/outreach-worker/src/sources/grants-gov/normalize.test.ts new file mode 100644 index 0000000..9d3a1d9 --- /dev/null +++ b/apps/outreach-worker/src/sources/grants-gov/normalize.test.ts @@ -0,0 +1,234 @@ +import { describe, expect, it } from 'vitest'; + +import type { + GrantsGovOpportunityDetail, + GrantsGovSearchHit, +} from '#~/sources/grants-gov/client.js'; +import { + opportunityDetailFixture, + searchHitFixture, +} from '#~/sources/grants-gov/fixtures.js'; +import { + normalizeApplicantTypes, + normalizeGrantsGovOpportunity, + parseCount, + parseDollarAmount, + parseGrantsGovDate, + stripHtml, +} from '#~/sources/grants-gov/normalize.js'; + +describe('normalizeGrantsGovOpportunity', () => { + it('normalizes a happy-path hit + detail into an insertable grant', () => { + const grant = normalizeGrantsGovOpportunity( + searchHitFixture, + opportunityDetailFixture, + ); + + expect(grant).toMatchObject({ + sourceUrl: 'https://www.grants.gov/search-results-detail/354821', + source: 'grants_gov', + title: 'Community Food Projects Competitive Grant Program', + funder: 'National Institute of Food and Agriculture', + geographicScope: null, + programAreas: null, + awardFloor: 10_000, + awardCeiling: 500_000, + expectedAwardsCount: 18, + matchRequirement: false, + applicationEffortEstimate: 'unknown', + applicationFormSupported: false, + status: 'open', + }); + expect(grant.synopsis).toBe( + 'The Community Food Projects (CFP) program supports community-based efforts to increase food security.\nApplicants must demonstrate community partnership.', + ); + expect(grant.eligibilityEntityTypes).toEqual([ + 'Nonprofits having a 501(c)(3) status with the IRS, other than institutions of higher education', + 'Nonprofits that do not have a 501(c)(3) status with the IRS, other than institutions of higher education', + ]); + expect(grant.openDate).toEqual(new Date(Date.UTC(2026, 0, 15))); + expect(grant.closeDate).toEqual(new Date(Date.UTC(2026, 3, 30))); + expect(grant.lastVerifiedAt).toBeInstanceOf(Date); + }); + + it('falls back to the search hit title/dates when detail fields are missing', () => { + const hit: GrantsGovSearchHit = { + ...searchHitFixture, + openDate: '02/01/2026', + closeDate: '05/01/2026', + }; + const detail: GrantsGovOpportunityDetail = { + id: hit.id, + opportunityNumber: null, + opportunityTitle: null, + agencyDetails: null, + synopsis: null, + }; + + const grant = normalizeGrantsGovOpportunity(hit, detail); + + expect(grant.title).toBe(hit.title); + expect(grant.synopsis).toBeNull(); + expect(grant.eligibilityEntityTypes).toBeNull(); + expect(grant.awardFloor).toBeNull(); + expect(grant.awardCeiling).toBeNull(); + expect(grant.expectedAwardsCount).toBeNull(); + expect(grant.openDate).toEqual(new Date(Date.UTC(2026, 1, 1))); + expect(grant.closeDate).toEqual(new Date(Date.UTC(2026, 4, 1))); + }); + + describe('funder fallback chain', () => { + it('prefers agencyDetails.agencyName', () => { + const grant = normalizeGrantsGovOpportunity( + searchHitFixture, + opportunityDetailFixture, + ); + expect(grant.funder).toBe('National Institute of Food and Agriculture'); + }); + + it('falls back to synopsis.agencyName, then hit.agency', () => { + const detail: GrantsGovOpportunityDetail = { + ...opportunityDetailFixture, + agencyDetails: null, + }; + const grant = normalizeGrantsGovOpportunity(searchHitFixture, detail); + expect(grant.funder).toBe('National Institute of Food and Agriculture'); + }); + + it('falls back to an agency code when no name is present anywhere', () => { + const hit: GrantsGovSearchHit = { + ...searchHitFixture, + agency: null, + agencyCode: 'USDA-NIFA', + }; + const detail: GrantsGovOpportunityDetail = { + ...opportunityDetailFixture, + agencyDetails: null, + synopsis: { + ...opportunityDetailFixture.synopsis!, + agencyName: null, + }, + }; + const grant = normalizeGrantsGovOpportunity(hit, detail); + expect(grant.funder).toBe('USDA-NIFA'); + }); + + it('falls back to a literal placeholder when nothing is present', () => { + const hit: GrantsGovSearchHit = { + ...searchHitFixture, + agency: null, + agencyCode: null, + }; + const detail: GrantsGovOpportunityDetail = { + id: hit.id, + opportunityNumber: null, + opportunityTitle: null, + agencyDetails: null, + synopsis: null, + }; + const grant = normalizeGrantsGovOpportunity(hit, detail); + expect(grant.funder).toBe('Unknown federal agency'); + }); + }); +}); + +describe('parseDollarAmount', () => { + it('parses a formatted dollar string', () => { + expect(parseDollarAmount('$1,500,000')).toBe(1_500_000); + }); + + it('parses a plain numeric string', () => { + expect(parseDollarAmount('750000')).toBe(750_000); + }); + + it('rounds a fractional number', () => { + expect(parseDollarAmount(1234.6)).toBe(1235); + }); + + it('returns null for missing/empty/garbage input', () => { + expect(parseDollarAmount(null)).toBeNull(); + expect(parseDollarAmount(undefined)).toBeNull(); + expect(parseDollarAmount('')).toBeNull(); + expect(parseDollarAmount('TBD')).toBeNull(); + expect(parseDollarAmount(Number.NaN)).toBeNull(); + }); +}); + +describe('parseCount', () => { + it('parses a plain count string', () => { + expect(parseCount('18')).toBe(18); + }); + + it('returns null for missing input', () => { + expect(parseCount(null)).toBeNull(); + }); +}); + +describe('parseGrantsGovDate', () => { + it('parses MM/DD/YYYY', () => { + expect(parseGrantsGovDate('04/30/2026')).toEqual( + new Date(Date.UTC(2026, 3, 30)), + ); + }); + + it('returns null for missing/empty input', () => { + expect(parseGrantsGovDate(null)).toBeNull(); + expect(parseGrantsGovDate(undefined)).toBeNull(); + expect(parseGrantsGovDate('')).toBeNull(); + expect(parseGrantsGovDate(' ')).toBeNull(); + }); + + it('returns null for an out-of-range month/day', () => { + expect(parseGrantsGovDate('13/01/2026')).toBeNull(); + expect(parseGrantsGovDate('01/32/2026')).toBeNull(); + }); + + it('returns null for a rolled-over calendar date', () => { + expect(parseGrantsGovDate('02/30/2026')).toBeNull(); + }); + + it('returns null for unparseable garbage', () => { + expect(parseGrantsGovDate('not-a-date')).toBeNull(); + }); +}); + +describe('stripHtml', () => { + it('strips tags and decodes common entities', () => { + expect( + stripHtml('

Applicants & partners must be eligible.

'), + ).toBe('Applicants & partners must be eligible.'); + }); + + it('converts
and block boundaries to newlines and collapses runs', () => { + expect(stripHtml('

First.

Second.




Third.')).toBe( + 'First.\nSecond.\n\nThird.', + ); + }); + + it('returns null for missing or empty-after-stripping input', () => { + expect(stripHtml(null)).toBeNull(); + expect(stripHtml(undefined)).toBeNull(); + expect(stripHtml('

')).toBeNull(); + }); +}); + +describe('normalizeApplicantTypes', () => { + it('extracts and trims descriptions', () => { + expect( + normalizeApplicantTypes([ + { id: 1, description: ' Nonprofits ' }, + { id: 2, description: 'State governments' }, + ]), + ).toEqual(['Nonprofits', 'State governments']); + }); + + it('drops blank descriptions and returns null if none remain', () => { + expect(normalizeApplicantTypes([{ id: 1, description: ' ' }])).toBeNull(); + }); + + it('returns null for missing/empty input', () => { + expect(normalizeApplicantTypes(null)).toBeNull(); + expect(normalizeApplicantTypes(undefined)).toBeNull(); + expect(normalizeApplicantTypes([])).toBeNull(); + }); +}); diff --git a/apps/outreach-worker/src/sources/grants-gov/normalize.ts b/apps/outreach-worker/src/sources/grants-gov/normalize.ts new file mode 100644 index 0000000..1a2a213 --- /dev/null +++ b/apps/outreach-worker/src/sources/grants-gov/normalize.ts @@ -0,0 +1,181 @@ +/** + * Pure normalization from Grants.gov's wire shapes to our `grants` schema. + * No fetch, no I/O — everything here is a deterministic function of its + * inputs so it can be unit tested against fixture JSON without a network. + */ +import type { NewGrantInput } from '@novelpad/outreach-core/server'; + +import type { + GrantsGovApplicantType, + GrantsGovOpportunityDetail, + GrantsGovSearchHit, +} from '#~/sources/grants-gov/client.js'; + +const GRANTS_GOV_DATE_RE = /^(\d{2})\/(\d{2})\/(\d{4})$/; + +/** + * Parses a Grants.gov date. The typical wire format is `MM/DD/YYYY`; a + * handful of fields have historically arrived as ISO-ish strings, so that's + * accepted as a fallback. Anything that doesn't parse to a real calendar + * date — including out-of-range components and month/day rollover (e.g. + * "02/30/2024") — returns `null` rather than a garbage `Date`. + */ +export function parseGrantsGovDate( + value: string | null | undefined, +): Date | null { + if (value == null) return null; + const trimmed = value.trim(); + if (trimmed === '') return null; + + const match = GRANTS_GOV_DATE_RE.exec(trimmed); + if (match != null) { + const month = Number(match[1]); + const day = Number(match[2]); + const year = Number(match[3]); + if (month < 1 || month > 12 || day < 1 || day > 31) return null; + + const date = new Date(Date.UTC(year, month - 1, day)); + const rolledOver = + date.getUTCFullYear() !== year || + date.getUTCMonth() !== month - 1 || + date.getUTCDate() !== day; + return rolledOver ? null : date; + } + + const fallback = new Date(trimmed); + return Number.isNaN(fallback.getTime()) ? null : fallback; +} + +/** + * Parses an award amount that may arrive as a number or as a formatted + * string (`"$1,500,000"`, `"1500000"`). Schema stores whole dollars — award + * amounts are never sub-dollar, so the result is rounded. + */ +export function parseDollarAmount( + value: string | number | null | undefined, +): number | null { + if (value == null) return null; + if (typeof value === 'number') { + return Number.isFinite(value) ? Math.round(value) : null; + } + const cleaned = value.replace(/[$,]/g, '').trim(); + if (cleaned === '') return null; + const parsed = Number(cleaned); + return Number.isFinite(parsed) ? Math.round(parsed) : null; +} + +/** Same coercion rules as `parseDollarAmount`, for plain counts (not money). */ +export function parseCount( + value: string | number | null | undefined, +): number | null { + return parseDollarAmount(value); +} + +/** + * Strips HTML tags from a synopsis description, converting common block + * boundaries to newlines and decoding the handful of entities Grants.gov + * actually emits. Not a general-purpose HTML sanitizer — synopses are + * federal agency copy, not untrusted user content. + */ +export function stripHtml(html: string | null | undefined): string | null { + if (html == null) return null; + + const text = html + .replace(//gi, '\n') + .replace(/<\/(p|div|li)>/gi, '\n') + .replace(/<[^>]+>/g, '') + .replace(/ /gi, ' ') + .replace(/&/gi, '&') + .replace(/</gi, '<') + .replace(/>/gi, '>') + .replace(/'/g, "'") + .replace(/"/g, '"') + .replace(/\r\n/g, '\n') + .replace(/\n{3,}/g, '\n\n') + .trim(); + + return text === '' ? null : text; +} + +/** Extracts non-empty eligibility descriptions from `synopsis.applicantTypes`. */ +export function normalizeApplicantTypes( + applicantTypes: ReadonlyArray | null | undefined, +): string[] | null { + if (applicantTypes == null || applicantTypes.length === 0) return null; + + const descriptions = applicantTypes + .map((type) => type.description?.trim()) + .filter((description): description is string => Boolean(description)); + + return descriptions.length === 0 ? null : descriptions; +} + +function firstNonEmpty( + ...values: ReadonlyArray +): string | null { + for (const value of values) { + const trimmed = value?.trim(); + if (trimmed) return trimmed; + } + return null; +} + +/** + * Resolves the funder display name: prefer the detail response's agency + * name (most complete), fall back to the search hit's agency name, then + * either response's agency code, then a literal placeholder so we never + * write an empty `funder`. + */ +function resolveFunder( + hit: GrantsGovSearchHit, + detail: GrantsGovOpportunityDetail, +): string { + return ( + firstNonEmpty( + detail.agencyDetails?.agencyName, + detail.synopsis?.agencyName, + hit.agency, + ) ?? + firstNonEmpty(detail.agencyDetails?.agencyCode, hit.agencyCode) ?? + 'Unknown federal agency' + ); +} + +/** + * Combines a `search2` hit with its `fetchOpportunity` detail into an + * insertable grant row. `sourceUrl` is the schema's upsert key, so it's + * derived purely from the opportunity id and stable across re-crawls. + * + * `geographicScope` is deliberately left `null`: federal opportunities are + * national by default and Grants.gov doesn't give us a reliable field to + * derive a narrower scope from — better to leave it unset than guess. + */ +export function normalizeGrantsGovOpportunity( + hit: GrantsGovSearchHit, + detail: GrantsGovOpportunityDetail, +): NewGrantInput { + const synopsis = detail.synopsis; + + return { + sourceUrl: `https://www.grants.gov/search-results-detail/${hit.id}`, + source: 'grants_gov', + title: detail.opportunityTitle ?? hit.title, + funder: resolveFunder(hit, detail), + synopsis: stripHtml(synopsis?.synopsisDesc), + eligibilityEntityTypes: normalizeApplicantTypes( + synopsis?.applicantTypes, + ), + geographicScope: null, + programAreas: null, + awardFloor: parseDollarAmount(synopsis?.awardFloor), + awardCeiling: parseDollarAmount(synopsis?.awardCeiling), + expectedAwardsCount: parseCount(synopsis?.expectedNumberOfAwards), + openDate: parseGrantsGovDate(hit.openDate ?? synopsis?.postingDate), + closeDate: parseGrantsGovDate(hit.closeDate ?? synopsis?.responseDate), + matchRequirement: false, + applicationEffortEstimate: 'unknown', + applicationFormSupported: false, + status: 'open', + lastVerifiedAt: new Date(), + }; +} diff --git a/apps/outreach-worker/src/sources/grants-gov/pagination.test.ts b/apps/outreach-worker/src/sources/grants-gov/pagination.test.ts new file mode 100644 index 0000000..d9776f3 --- /dev/null +++ b/apps/outreach-worker/src/sources/grants-gov/pagination.test.ts @@ -0,0 +1,98 @@ +import { describe, expect, it } from 'vitest'; + +import { nextSearchRequest } from '#~/sources/grants-gov/pagination.js'; + +describe('nextSearchRequest', () => { + it('requests a full first page when hitCount is not yet known', () => { + expect( + nextSearchRequest({ + startRecordNum: 0, + collected: 0, + hitCount: null, + maxHits: 1000, + pageSize: 100, + }), + ).toEqual({ startRecordNum: 0, rows: 100 }); + }); + + it('requests the remaining hits once hitCount is known and smaller than pageSize', () => { + expect( + nextSearchRequest({ + startRecordNum: 100, + collected: 100, + hitCount: 130, + maxHits: 1000, + pageSize: 100, + }), + ).toEqual({ startRecordNum: 100, rows: 30 }); + }); + + it('returns null once every known hit has been collected', () => { + expect( + nextSearchRequest({ + startRecordNum: 130, + collected: 130, + hitCount: 130, + maxHits: 1000, + pageSize: 100, + }), + ).toBeNull(); + }); + + it('caps rows at the remaining maxHits budget even mid-page', () => { + expect( + nextSearchRequest({ + startRecordNum: 950, + collected: 950, + hitCount: 5000, + maxHits: 1000, + pageSize: 100, + }), + ).toEqual({ startRecordNum: 950, rows: 50 }); + }); + + it('returns null once maxHits has been reached', () => { + expect( + nextSearchRequest({ + startRecordNum: 1000, + collected: 1000, + hitCount: 5000, + maxHits: 1000, + pageSize: 100, + }), + ).toBeNull(); + }); + + it('returns null for a zero hitCount (no open opportunities)', () => { + expect( + nextSearchRequest({ + startRecordNum: 0, + collected: 0, + hitCount: 0, + maxHits: 1000, + pageSize: 100, + }), + ).toBeNull(); + }); + + it('returns null for a non-positive maxHits or pageSize', () => { + expect( + nextSearchRequest({ + startRecordNum: 0, + collected: 0, + hitCount: null, + maxHits: 0, + pageSize: 100, + }), + ).toBeNull(); + expect( + nextSearchRequest({ + startRecordNum: 0, + collected: 0, + hitCount: null, + maxHits: 1000, + pageSize: 0, + }), + ).toBeNull(); + }); +}); diff --git a/apps/outreach-worker/src/sources/grants-gov/pagination.ts b/apps/outreach-worker/src/sources/grants-gov/pagination.ts new file mode 100644 index 0000000..65589fe --- /dev/null +++ b/apps/outreach-worker/src/sources/grants-gov/pagination.ts @@ -0,0 +1,52 @@ +/** + * Pure pagination arithmetic for the Grants.gov Search2 API. + * + * `search2` doesn't accept a "give me everything" request — every call needs + * an explicit `rows`/`startRecordNum` pair, and the true total (`hitCount`) + * is only known after the first response comes back. This module isolates + * the "what should the next request look like, if anything" decision from + * the actual HTTP calls in `client.ts` so it can be unit tested without + * mocking fetch. + */ + +export interface NextSearchRequestState { + /** `startRecordNum` to use for the next request, if any. */ + readonly startRecordNum: number; + /** Hits collected so far across all prior pages. */ + readonly collected: number; + /** Total hits reported by the API, or `null` before the first response. */ + readonly hitCount: number | null; + /** Caller-supplied cap on total hits to collect this run. */ + readonly maxHits: number; + /** Page size (`rows`) to request per call. */ + readonly pageSize: number; +} + +export interface SearchPageRequest { + readonly startRecordNum: number; + readonly rows: number; +} + +/** + * Decides the next `{ startRecordNum, rows }` to request, or `null` if + * pagination is complete: either every known hit has been collected, or the + * caller-supplied `maxHits` cap has been reached. + */ +export function nextSearchRequest( + state: NextSearchRequestState, +): SearchPageRequest | null { + const { startRecordNum, collected, hitCount, maxHits, pageSize } = state; + + if (maxHits <= 0 || pageSize <= 0) return null; + if (collected >= maxHits) return null; + if (hitCount != null && collected >= hitCount) return null; + + const remainingByHitCount = + hitCount == null ? Number.POSITIVE_INFINITY : hitCount - collected; + const remainingByCap = maxHits - collected; + const rows = Math.min(pageSize, remainingByHitCount, remainingByCap); + + if (rows <= 0) return null; + + return { startRecordNum, rows }; +} diff --git a/apps/outreach-worker/src/sources/nhdoj/extract-pdf-text.ts b/apps/outreach-worker/src/sources/nhdoj/extract-pdf-text.ts new file mode 100644 index 0000000..b28ef01 --- /dev/null +++ b/apps/outreach-worker/src/sources/nhdoj/extract-pdf-text.ts @@ -0,0 +1,57 @@ +/** + * Thin impure layer: turns raw NHDOJ Charitable Trusts registry PDF bytes + * into positioned text items, one array per page. Uses pdfjs-dist's legacy + * Node build, which self-detects `isNodeJS` and disables the worker-thread + * path automatically (falls back to an in-process "fake worker" — see + * `PDFWorker#_initialize` in `pdfjs-dist/legacy/build/pdf.mjs`). No + * `GlobalWorkerOptions.workerSrc` or explicit disable flag is needed or + * exposed on `DocumentInitParameters`; do not add one. + * + * Deliberately dumb: no table/row reasoning here. That lives in + * `parse-registry.ts` as a pure function over the output of this module, so + * the row-reconstruction logic can be unit tested against synthetic + * fixtures without a real PDF. + */ +import { getDocument } from 'pdfjs-dist/legacy/build/pdf.mjs'; + +export interface PositionedTextItem { + readonly str: string; + readonly x: number; + readonly y: number; +} + +/** + * Extracts positioned text items for every page of a PDF. + * + * `transform` on a pdf.js text item is the 2D affine matrix + * `[a, b, c, d, e, f]` mapping glyph space to page space; `e`/`f` (indices + * 4/5) are the item's origin in page coordinates, which is all the row + * reconstruction in `parse-registry.ts` needs. + */ +export async function extractPositionedText( + pdfBytes: Uint8Array, +): Promise { + const loadingTask = getDocument({ data: pdfBytes, useSystemFonts: true }); + const doc = await loadingTask.promise; + try { + const pages: PositionedTextItem[][] = []; + for (let pageNum = 1; pageNum <= doc.numPages; pageNum++) { + const page = await doc.getPage(pageNum); + const textContent = await page.getTextContent(); + const items: PositionedTextItem[] = []; + for (const item of textContent.items) { + // `TextMarkedContent` entries (structure markers) have no + // `transform`/`str` pair usable here; skip anything that isn't a + // plain positioned glyph run. + if (!('str' in item) || !('transform' in item)) continue; + const [, , , , x, y] = item.transform; + items.push({ str: item.str, x, y }); + } + pages.push(items); + page.cleanup(); + } + return pages; + } finally { + await doc.destroy(); + } +} diff --git a/apps/outreach-worker/src/sources/nhdoj/parse-registry.test.ts b/apps/outreach-worker/src/sources/nhdoj/parse-registry.test.ts new file mode 100644 index 0000000..144a13a --- /dev/null +++ b/apps/outreach-worker/src/sources/nhdoj/parse-registry.test.ts @@ -0,0 +1,240 @@ +import { describe, expect, it } from 'vitest'; + +import type { PositionedTextItem } from './extract-pdf-text.js'; +import { + normalizeRegistrationStatus, + normalizeRegistryRows, + reconstructRegistryRows, + type RegistryRow, +} from './parse-registry.js'; + +/** Column x-starts used across fixtures: name @50, city @300, status @450. */ +const NAME_X = 50; +const CITY_X = 300; +const STATUS_X = 450; + +function item(str: string, x: number, y: number): PositionedTextItem { + return { str, x, y }; +} + +function headerRow(y: number): PositionedTextItem[] { + return [ + item('Organization', NAME_X, y), + item('City', CITY_X, y), + item('Status', STATUS_X, y), + ]; +} + +describe('reconstructRegistryRows', () => { + it('returns no rows for an empty page', () => { + expect(reconstructRegistryRows([[]])).toEqual([]); + expect(reconstructRegistryRows([])).toEqual([]); + }); + + it('infers column boundaries from the header row and buckets data by x', () => { + const page = [ + ...headerRow(900), + item('Acme Foundation', NAME_X, 880), + item('Concord', CITY_X, 880), + item('Good Standing', STATUS_X, 880), + ]; + + const rows = reconstructRegistryRows([page]); + + expect(rows).toEqual([ + { name: 'Acme Foundation', city: 'Concord', status: 'Good Standing' }, + ]); + }); + + it('infers boundaries correctly even when they differ from other fixtures', () => { + const page = [ + item('Organization', 40, 900), + item('City', 250, 900), + item('Status', 500, 900), + // Slightly right of each boundary — still buckets into the same column. + item('Bright Futures NH', 42, 870), + item('Nashua', 255, 870), + item('Active', 505, 870), + ]; + + const rows = reconstructRegistryRows([page]); + + expect(rows).toEqual([ + { name: 'Bright Futures NH', city: 'Nashua', status: 'Active' }, + ]); + }); + + it('folds a multi-line org name (continuation line has empty city/status) into the previous row', () => { + const page = [ + ...headerRow(900), + item('Very Long Nonprofit', NAME_X, 860), + item('Manchester', CITY_X, 860), + item('Lapsed', STATUS_X, 860), + // Wrapped second line of the same org name — no city/status text. + item('Name Incorporated', NAME_X, 840), + ]; + + const rows = reconstructRegistryRows([page]); + + expect(rows).toEqual([ + { + name: 'Very Long Nonprofit Name Incorporated', + city: 'Manchester', + status: 'Lapsed', + }, + ]); + }); + + it('supports multi-word continuation lines split across several text items', () => { + const page = [ + ...headerRow(900), + item('Friends of the', NAME_X, 860), + item('Merrimack', CITY_X, 860), + item('Current', STATUS_X, 860), + item('River', NAME_X, 840), + item('Watershed', NAME_X + 40, 840), + ]; + + const rows = reconstructRegistryRows([page]); + + expect(rows).toEqual([ + { + name: 'Friends of the River Watershed', + city: 'Merrimack', + status: 'Current', + }, + ]); + }); + + it('drops header, letterhead, and page-number footer lines', () => { + const page = [ + item('State of New Hampshire Department of Justice', NAME_X, 950), + ...headerRow(900), + item('Acme Foundation', NAME_X, 880), + item('Concord', CITY_X, 880), + item('Good Standing', STATUS_X, 880), + item('3', NAME_X, 50), + ]; + + const rows = reconstructRegistryRows([page]); + + expect(rows).toEqual([ + { name: 'Acme Foundation', city: 'Concord', status: 'Good Standing' }, + ]); + }); + + it('drops a "Page X of Y" footer line', () => { + const page = [ + ...headerRow(900), + item('Acme Foundation', NAME_X, 880), + item('Concord', CITY_X, 880), + item('Good Standing', STATUS_X, 880), + item('Page 1 of 12', NAME_X, 50), + ]; + + const rows = reconstructRegistryRows([page]); + + expect(rows).toEqual([ + { name: 'Acme Foundation', city: 'Concord', status: 'Good Standing' }, + ]); + }); + + it('accepts explicit columnBoundaries for a continuation page with no repeated header', () => { + const page = [ + item('Second Page Org', NAME_X, 900), + item('Keene', CITY_X, 900), + item('Unknown', STATUS_X, 900), + ]; + + const rows = reconstructRegistryRows([page], { + columnBoundaries: [NAME_X, CITY_X, STATUS_X], + }); + + expect(rows).toEqual([ + { name: 'Second Page Org', city: 'Keene', status: 'Unknown' }, + ]); + }); + + it('reuses the last inferred boundaries across pages whose header does not repeat', () => { + const page1 = [ + ...headerRow(900), + item('First Page Org', NAME_X, 880), + item('Concord', CITY_X, 880), + item('Good Standing', STATUS_X, 880), + ]; + const page2 = [ + item('Second Page Org', NAME_X, 900), + item('Keene', CITY_X, 900), + item('Lapsed', STATUS_X, 900), + ]; + + const rows = reconstructRegistryRows([page1, page2]); + + expect(rows).toEqual([ + { name: 'First Page Org', city: 'Concord', status: 'Good Standing' }, + { name: 'Second Page Org', city: 'Keene', status: 'Lapsed' }, + ]); + }); + + it('throws when a page has data but no boundaries can be determined', () => { + const page = [item('Mystery Org', NAME_X, 900)]; + + expect(() => reconstructRegistryRows([page])).toThrow( + /could not determine column boundaries/, + ); + }); +}); + +describe('normalizeRegistrationStatus', () => { + it.each([ + ['Good Standing', 'good_standing'], + ['GOOD STANDING', 'good_standing'], + ['Current', 'good_standing'], + ['Active', 'good_standing'], + ['Lapsed', 'lapsed'], + ['Delinquent', 'lapsed'], + ['Suspended', 'lapsed'], + ['Expired', 'lapsed'], + ['Revoked', 'lapsed'], + ['', 'unknown'], + ['Pending Review', 'unknown'], + ] as const)('maps %s -> %s', (input, expected) => { + expect(normalizeRegistrationStatus(input)).toBe(expected); + }); +}); + +describe('normalizeRegistryRows', () => { + it('maps status variants, trims/collapses whitespace, and preserves a null city', () => { + const rows: RegistryRow[] = [ + { name: ' Acme Foundation ', city: ' Concord ', status: 'Good Standing' }, + { name: 'Delinquent Org', city: null, status: 'Delinquent' }, + { name: 'Mystery Org', city: 'Keene', status: 'Something Else' }, + ]; + + expect(normalizeRegistryRows(rows)).toEqual([ + { name: 'Acme Foundation', city: 'Concord', status: 'good_standing' }, + { name: 'Delinquent Org', city: null, status: 'lapsed' }, + { name: 'Mystery Org', city: 'Keene', status: 'unknown' }, + ]); + }); + + it('drops rows with an empty name and header/footer artifact rows that slipped through', () => { + const rows: RegistryRow[] = [ + { name: '', city: 'Concord', status: 'Good Standing' }, + { name: 'Organization City Status', city: null, status: '' }, + { name: 'Real Org', city: 'Nashua', status: 'Active' }, + ]; + + expect(normalizeRegistryRows(rows)).toEqual([ + { name: 'Real Org', city: 'Nashua', status: 'good_standing' }, + ]); + }); + + it('collapses an empty-string city to null', () => { + const rows: RegistryRow[] = [{ name: 'Org', city: ' ', status: 'Active' }]; + + expect(normalizeRegistryRows(rows)).toEqual([ + { name: 'Org', city: null, status: 'good_standing' }, + ]); + }); +}); diff --git a/apps/outreach-worker/src/sources/nhdoj/parse-registry.ts b/apps/outreach-worker/src/sources/nhdoj/parse-registry.ts new file mode 100644 index 0000000..5b4a2a0 --- /dev/null +++ b/apps/outreach-worker/src/sources/nhdoj/parse-registry.ts @@ -0,0 +1,260 @@ +/** + * PURE row reconstruction over positioned text extracted from the NHDOJ + * Charitable Trusts registry PDF (see `extract-pdf-text.ts` for the impure + * layer that produces the input). No PDF or network dependency here — every + * function operates on plain `PositionedTextItem[][]` so tests can drive it + * with synthetic fixtures. + * + * The registry renders as a 3-column table (organization name, city, + * registration status). pdf.js gives us a flat bag of positioned glyph runs + * per page with no row/column structure, so reconstruction happens in two + * passes: + * + * 1. `reconstructRegistryRows` groups items into visual lines by y + * (tolerance ~2pt), buckets each line's items into columns by x + * (boundaries inferred from the header row's token x-positions, or + * supplied explicitly via `options.columnBoundaries` when a page's + * header doesn't repeat), and stitches multi-line org names back + * together — a continuation line has text only in the name column. + * 2. `normalizeRegistryRows` maps the free-text status column onto the + * tri-state the DB expects and drops anything that isn't really a data + * row (repeated header, page number, agency letterhead). + */ +import type { PositionedTextItem } from './extract-pdf-text.js'; + +export interface RegistryRow { + readonly name: string; + readonly city: string | null; + readonly status: string; +} + +export type NormalizedRegistrationStatus = 'good_standing' | 'lapsed' | 'unknown'; + +export interface NormalizedRegistryRow { + readonly name: string; + readonly city: string | null; + readonly status: NormalizedRegistrationStatus; +} + +export interface ParseRegistryOptions { + /** Max y-distance (pt) between items considered part of the same visual line. Default 2. */ + readonly yTolerance?: number; + /** + * Explicit x-position column starts `[nameStart, cityStart, statusStart]`, + * ascending. When omitted, boundaries are inferred per page from that + * page's header row (falling back to the most recently inferred/ provided + * boundaries for pages whose header doesn't repeat, e.g. continuation + * pages). + */ + readonly columnBoundaries?: readonly [number, number, number]; +} + +const DEFAULT_Y_TOLERANCE = 2; + +interface Line { + readonly y: number; + readonly items: PositionedTextItem[]; +} + +function collapseWhitespace(text: string): string { + return text.trim().replace(/\s+/g, ' '); +} + +/** + * Groups a page's positioned items into visual lines (top-to-bottom by y, + * items within a line ordered left-to-right by x). + */ +function groupLines( + items: readonly PositionedTextItem[], + yTolerance: number, +): Line[] { + // Page-space y increases upward; descending sort gives reading order. + const sorted = [...items].sort((a, b) => b.y - a.y); + const lines: Line[] = []; + + for (const item of sorted) { + const current = lines[lines.length - 1]; + if (current != null && Math.abs(current.y - item.y) <= yTolerance) { + current.items.push(item); + } else { + lines.push({ y: item.y, items: [item] }); + } + } + + for (const line of lines) { + line.items.sort((a, b) => a.x - b.x); + } + + return lines; +} + +/** Bucket an x-position into a column index given ascending column-start boundaries. */ +function columnIndexForX(x: number, boundaries: readonly number[]): number { + for (let i = boundaries.length - 1; i >= 0; i--) { + if (x >= boundaries[i]!) return i; + } + return 0; +} + +/** + * Looks for a line whose items spell out the three column headers + * ("Organization"/"Name", "City", "Status") and returns their x-positions, + * ascending — which is also left-to-right table order (name, city, status). + */ +function inferColumnBoundariesFromHeader( + lines: readonly Line[], +): [number, number, number] | null { + for (const line of lines) { + const nameItem = line.items.find((i) => + /organi[sz]ation|^name$/i.test(i.str.trim()), + ); + const cityItem = line.items.find((i) => /^city$/i.test(i.str.trim())); + const statusItem = line.items.find((i) => /status/i.test(i.str.trim())); + + if (nameItem != null && cityItem != null && statusItem != null) { + return [nameItem.x, cityItem.x, statusItem.x].sort( + (a, b) => a - b, + ) as [number, number, number]; + } + } + return null; +} + +function isColumnHeaderLine(lineText: string): boolean { + const upper = lineText.toUpperCase(); + return ( + (upper.includes('ORGANIZATION') || /\bNAME\b/.test(upper)) && + upper.includes('CITY') && + upper.includes('STATUS') + ); +} + +/** Page-number footers, repeated agency letterhead, and blank lines. */ +function isFooterOrArtifactLine(lineText: string): boolean { + const trimmed = lineText.trim(); + if (trimmed === '') return true; + if (/^\d+$/.test(trimmed)) return true; + if (/^page\s+\d+(\s+of\s+\d+)?$/i.test(trimmed)) return true; + if (/department of justice/i.test(trimmed)) return true; + return false; +} + +function isHeaderOrFooterLine(lineText: string): boolean { + return isColumnHeaderLine(lineText) || isFooterOrArtifactLine(lineText); +} + +/** + * Reconstructs `{ name, city, status }` rows from positioned text, one page + * array at a time. Multi-line org names (a continuation line with nothing + * in the city/status columns) are folded back into the previous row. + * + * Throws if a page has data but no column boundaries can be determined + * (no header row found on any page so far, and none supplied) — that means + * the registry's layout changed and silent misparsing is worse than a loud + * failure here. + */ +export function reconstructRegistryRows( + pages: readonly (readonly PositionedTextItem[])[], + options: ParseRegistryOptions = {}, +): RegistryRow[] { + const yTolerance = options.yTolerance ?? DEFAULT_Y_TOLERANCE; + let boundaries: readonly [number, number, number] | null = + options.columnBoundaries ?? null; + + const rows: RegistryRow[] = []; + + for (const pageItems of pages) { + if (pageItems.length === 0) continue; + + const lines = groupLines(pageItems, yTolerance); + + if (options.columnBoundaries == null) { + const inferred = inferColumnBoundariesFromHeader(lines); + if (inferred != null) boundaries = inferred; + } + + if (boundaries == null) { + throw new Error( + 'reconstructRegistryRows: could not determine column boundaries ' + + '(no header row found and none provided via options.columnBoundaries)', + ); + } + + for (const line of lines) { + const lineText = line.items.map((i) => i.str).join(' '); + if (isHeaderOrFooterLine(lineText)) continue; + + const columns: [string[], string[], string[]] = [[], [], []]; + for (const item of line.items) { + const idx = columnIndexForX(item.x, boundaries); + columns[idx as 0 | 1 | 2].push(item.str); + } + + const name = collapseWhitespace(columns[0].join(' ')); + const city = collapseWhitespace(columns[1].join(' ')); + const status = collapseWhitespace(columns[2].join(' ')); + + if (name === '' && city === '' && status === '') continue; + + // A line with text only in the name column continues the previous + // row's (wrapped) org name rather than starting a new row. + if (city === '' && status === '' && rows.length > 0) { + const prev = rows[rows.length - 1]!; + rows[rows.length - 1] = { + ...prev, + name: collapseWhitespace(`${prev.name} ${name}`), + }; + continue; + } + + rows.push({ name, city: city === '' ? null : city, status }); + } + } + + return rows; +} + +const GOOD_STANDING_PATTERNS = [/good\s*standing/i, /\bcurrent\b/i, /\bactive\b/i]; +const LAPSED_PATTERNS = [ + /\blapsed\b/i, + /\bdelinquent\b/i, + /\bsuspended\b/i, + /\bexpired\b/i, + /\brevoked\b/i, +]; + +export function normalizeRegistrationStatus( + status: string, +): NormalizedRegistrationStatus { + const trimmed = status.trim(); + if (trimmed === '') return 'unknown'; + if (GOOD_STANDING_PATTERNS.some((p) => p.test(trimmed))) return 'good_standing'; + if (LAPSED_PATTERNS.some((p) => p.test(trimmed))) return 'lapsed'; + return 'unknown'; +} + +/** + * Maps raw rows onto the tri-state status the DB expects, collapses + * whitespace, and drops anything that isn't really a registrant (a + * repeated header row or letterhead line that slipped through row + * reconstruction as a degenerate one-column row). + */ +export function normalizeRegistryRows( + rows: readonly RegistryRow[], +): NormalizedRegistryRow[] { + const normalized: NormalizedRegistryRow[] = []; + + for (const row of rows) { + const name = collapseWhitespace(row.name); + if (name === '' || isHeaderOrFooterLine(name)) continue; + + const city = row.city == null ? null : collapseWhitespace(row.city); + normalized.push({ + name, + city: city === '' ? null : city, + status: normalizeRegistrationStatus(row.status), + }); + } + + return normalized; +} diff --git a/apps/outreach-worker/src/sources/pnd-rss/client.ts b/apps/outreach-worker/src/sources/pnd-rss/client.ts new file mode 100644 index 0000000..94e04f0 --- /dev/null +++ b/apps/outreach-worker/src/sources/pnd-rss/client.ts @@ -0,0 +1,26 @@ +/** + * HTTP client for the Philanthropy News Digest "RFPs" RSS feed. + * + * Kept deliberately thin — a plain `fetch` returning the raw XML body. + * Parsing/normalization lives in `normalize.ts` so it can be unit tested + * against fixture strings without a network call. + */ + +/** Default PND RFP feed URL; overridable via `PND_RFP_FEED_URL` at the call site. */ +export const PND_RFP_FEED_URL = 'https://philanthropynewsdigest.org/feeds/rfps.rss'; + +/** + * Fetches the raw RSS XML body from the PND RFP feed. + * + * Throws on any non-2xx response so the caller's step-level retry policy + * (see `ingest-pnd-rss.ts`) can kick in. + */ +export async function fetchPndRfpFeed(feedUrl: string): Promise { + const response = await fetch(feedUrl); + if (!response.ok) { + throw new Error( + `fetchPndRfpFeed: ${feedUrl} responded with ${response.status} ${response.statusText}`, + ); + } + return response.text(); +} diff --git a/apps/outreach-worker/src/sources/pnd-rss/normalize.test.ts b/apps/outreach-worker/src/sources/pnd-rss/normalize.test.ts new file mode 100644 index 0000000..ebabbc4 --- /dev/null +++ b/apps/outreach-worker/src/sources/pnd-rss/normalize.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, it } from 'vitest'; + +import { parsePndFeed } from './normalize.js'; + +const MULTI_ITEM_FIXTURE = ` + + + Philanthropy News Digest: RFPs + https://philanthropynewsdigest.org/rfps + + The Ford Foundation Accepting Applications for Social Justice Grants + https://philanthropynewsdigest.org/rfps/the-ford-foundation-accepting-applications-for-social-justice-grants + <p>The <a href="https://example.org">Ford Foundation</a> is accepting applications for its Social Justice Grants program. Grants of up to \$250,000 will support <strong>nonprofit</strong> organizations working to advance social justice.</p> <p>Deadline: March 15, 2027.</p> + Mon, 01 Feb 2027 00:00:00 GMT + + + Community Health Innovators Grant Program Announced + https://philanthropynewsdigest.org/rfps/community-health-innovators-grant-program-announced + A regional funder has launched a new grant program for community health innovation. No named sponsor is given in this blurb, and no due date is specified either. + Tue, 02 Feb 2027 00:00:00 GMT + + + Arts Access Coalition Invites Proposals for Rural Arts Fund + https://philanthropynewsdigest.org/rfps/arts-access-coalition-invites-proposals-for-rural-arts-fund + The Arts Access Coalition invites proposals for its Rural Arts Fund, supporting arts programming in underserved rural communities. Applications are due 2027-04-30. + Wed, 03 Feb 2027 00:00:00 GMT + + + Untitled Listing With No Link + This item has no <link> element and must be skipped entirely. + Thu, 04 Feb 2027 00:00:00 GMT + + +`; + +const SINGLE_ITEM_FIXTURE = ` + + + Philanthropy News Digest: RFPs + + The Whitfield Family Foundation Offers Youth Mentorship Grants + https://philanthropynewsdigest.org/rfps/the-whitfield-family-foundation-offers-youth-mentorship-grants + The Whitfield Family Foundation offers grants supporting youth mentorship programs nationwide. Deadline: 2027-06-01. + + +`; + +describe('parsePndFeed', () => { + it('extracts the funder from a leading proper-noun phrase before the announcing verb', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + expect(grants[0]?.funder).toBe('The Ford Foundation'); + }); + + it('falls back to "See RFP" when no funder phrase can be extracted', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + expect(grants[1]?.funder).toBe('See RFP'); + }); + + it('extracts a "Deadline: " close date from the description', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + expect(grants[0]?.closeDate).toBeInstanceOf(Date); + expect(grants[0]?.closeDate?.toISOString().slice(0, 10)).toBe('2027-03-15'); + }); + + it('extracts a "due " close date from the description', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + expect(grants[2]?.closeDate).toBeInstanceOf(Date); + expect(grants[2]?.closeDate?.toISOString().slice(0, 10)).toBe('2027-04-30'); + }); + + it('returns null closeDate when no deadline phrase is present', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + expect(grants[1]?.closeDate).toBeNull(); + }); + + it('strips HTML tags and decodes entities from the description into synopsis', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + expect(grants[0]?.synopsis).not.toContain('<'); + expect(grants[0]?.synopsis).not.toContain('<'); + expect(grants[0]?.synopsis).not.toContain('&'); + expect(grants[0]?.synopsis).toContain('Ford Foundation'); + expect(grants[0]?.synopsis).toContain('nonprofit'); + }); + + it('skips items with no link element', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + expect(grants).toHaveLength(3); + expect( + grants.some((g) => g.title === 'Untitled Listing With No Link'), + ).toBe(false); + }); + + it('sets source, applicationEffortEstimate, and lastVerifiedAt on every item', () => { + const grants = parsePndFeed(MULTI_ITEM_FIXTURE); + for (const grant of grants) { + expect(grant.source).toBe('pnd_rss'); + expect(grant.applicationEffortEstimate).toBe('unknown'); + expect(grant.lastVerifiedAt).toBeInstanceOf(Date); + expect(grant.openDate).toBeNull(); + } + }); + + it('coerces a single-item feed (fast-xml-parser collapses one-element arrays)', () => { + const grants = parsePndFeed(SINGLE_ITEM_FIXTURE); + expect(grants).toHaveLength(1); + expect(grants[0]?.sourceUrl).toBe( + 'https://philanthropynewsdigest.org/rfps/the-whitfield-family-foundation-offers-youth-mentorship-grants', + ); + expect(grants[0]?.funder).toBe('The Whitfield Family Foundation'); + expect(grants[0]?.closeDate?.toISOString().slice(0, 10)).toBe('2027-06-01'); + }); + + it('returns an empty array for a feed with no items', () => { + const empty = `Empty`; + expect(parsePndFeed(empty)).toEqual([]); + }); +}); diff --git a/apps/outreach-worker/src/sources/pnd-rss/normalize.ts b/apps/outreach-worker/src/sources/pnd-rss/normalize.ts new file mode 100644 index 0000000..9e0d3da --- /dev/null +++ b/apps/outreach-worker/src/sources/pnd-rss/normalize.ts @@ -0,0 +1,141 @@ +/** + * Pure normalization of a Philanthropy News Digest "RFPs" RSS feed body into + * `NewGrantInput` rows. + * + * PND's RFP feed doesn't carry structured funder/deadline fields the way + * Grants.gov's API does — everything upstream is prose inside `` and + * `<description>`. Funder and close-date extraction here are therefore + * best-effort regex heuristics over that prose, not authoritative parses; + * items that don't match fall back to a neutral placeholder (`'See RFP'`) + * or `null` rather than guessing. + */ +import { XMLParser } from 'fast-xml-parser'; + +import type { NewGrantInput } from '@novelpad/outreach-core/server'; + +interface PndRssItem { + readonly title?: string; + readonly link?: string; + readonly description?: string; +} + +interface PndRssParsed { + readonly rss?: { + readonly channel?: { + readonly item?: PndRssItem[]; + }; + }; +} + +// `isArray` forces `<item>` to always parse as an array, even when a feed +// has exactly one `<item>` — fast-xml-parser otherwise collapses a +// single-element repeated tag down to a bare object. +const parser = new XMLParser({ + ignoreAttributes: true, + trimValues: true, + isArray: (_name, jpath) => jpath === 'rss.channel.item', +}); + +/** Coerces a possibly-singular parsed `<item>` node into an array. Belt-and-suspenders alongside `isArray` above. */ +function coerceArray( + value: PndRssItem | PndRssItem[] | undefined, +): readonly PndRssItem[] { + if (value == null) return []; + return Array.isArray(value) ? value : [value]; +} + +const HTML_TAG_PATTERN = /<[^>]*>/g; +const NAMED_ENTITY_DECODERS: ReadonlyArray<readonly [RegExp, string]> = [ + [/ /gi, ' '], + [/&/gi, '&'], + [/</gi, '<'], + [/>/gi, '>'], + [/"/gi, '"'], + [/&(?:#39|apos);/gi, "'"], +]; + +/** Strips HTML markup and decodes common entities from RSS `<description>` bodies. */ +function stripHtml(html: string): string { + let text = html.replace(HTML_TAG_PATTERN, ' '); + for (const [pattern, replacement] of NAMED_ENTITY_DECODERS) { + text = text.replace(pattern, replacement); + } + // Numeric entities, e.g. ’ (right single quote). + text = text.replace(/&#(\d+);/g, (_match, code: string) => + String.fromCharCode(Number(code)), + ); + return text.replace(/\s+/g, ' ').trim(); +} + +// Matches the leading phrase of a PND blurb up to the verb announcing the +// call for applications, e.g. "The Ford Foundation is accepting +// applications for..." -> "The Ford Foundation". +const FUNDER_PATTERN = + /^(.+?)\s+(?:is\s+|are\s+|has\s+|have\s+)?(?:now\s+)?(?:accepting|invites?|seeks?|announces?|offers?)\b/i; + +function matchFunder(text: string): string | null { + const match = text.match(FUNDER_PATTERN); + const candidate = match?.[1]?.trim(); + return candidate != null && candidate.length > 0 ? candidate : null; +} + +/** Best-effort funder extraction: try the title first, then the (stripped) description, else fall back. */ +function extractFunder(title: string, synopsis: string): string { + return matchFunder(title) ?? matchFunder(synopsis) ?? 'See RFP'; +} + +// Best-effort deadline extraction over free-text prose. Ordered +// most-specific-first; the first pattern to match and parse to a valid +// Date wins. +const DEADLINE_PATTERNS: ReadonlyArray<RegExp> = [ + /deadline:?\s*(?:is\s*)?([A-Za-z]+\.?\s+\d{1,2},?\s+\d{4})/i, + /deadline:?\s*(?:is\s*)?(\d{4}-\d{2}-\d{2})/i, + /due\s+(?:date\s+(?:is\s+)?)?(?:on\s+)?([A-Za-z]+\.?\s+\d{1,2},?\s+\d{4})/i, + /due\s+(?:date\s+(?:is\s+)?)?(?:on\s+)?(\d{4}-\d{2}-\d{2})/i, +]; + +function extractCloseDate(text: string): Date | null { + for (const pattern of DEADLINE_PATTERNS) { + const match = text.match(pattern); + const dateText = match?.[1]; + if (dateText == null) continue; + const parsed = new Date(dateText); + if (!Number.isNaN(parsed.getTime())) return parsed; + } + return null; +} + +/** + * Parses a raw PND RFP RSS feed body into upsert-ready grant rows. + * + * Items with no `<link>` are skipped — `sourceUrl` is the upsert key in + * `serverInsertGrants`, so a linkless item has nothing stable to key on. + */ +export function parsePndFeed(xml: string): NewGrantInput[] { + const parsed = parser.parse(xml) as PndRssParsed; + const items = coerceArray(parsed.rss?.channel?.item); + + const grants: NewGrantInput[] = []; + for (const item of items) { + const link = typeof item.link === 'string' ? item.link.trim() : ''; + if (link.length === 0) continue; + + const title = typeof item.title === 'string' ? item.title.trim() : ''; + const rawDescription = + typeof item.description === 'string' ? item.description : ''; + const synopsis = stripHtml(rawDescription); + + grants.push({ + funder: extractFunder(title, synopsis), + title: title.length > 0 ? title : 'Untitled RFP', + sourceUrl: link, + synopsis: synopsis.length > 0 ? synopsis : null, + closeDate: extractCloseDate(`${title} ${synopsis}`), + openDate: null, + applicationEffortEstimate: 'unknown', + source: 'pnd_rss', + lastVerifiedAt: new Date(), + }); + } + return grants; +} diff --git a/apps/outreach-worker/src/sources/propublica/client.ts b/apps/outreach-worker/src/sources/propublica/client.ts new file mode 100644 index 0000000..301d6ff --- /dev/null +++ b/apps/outreach-worker/src/sources/propublica/client.ts @@ -0,0 +1,98 @@ +/** + * ProPublica Nonprofit Explorer API v2 client. + * + * Free, unauthenticated, but rate-limited enough that we self-throttle with a + * politeness delay between calls (injectable so tests don't pay the wait). + * See https://projects.propublica.org/nonprofits/api + */ + +const BASE_URL = 'https://projects.propublica.org/nonprofits/api/v2'; +const DEFAULT_POLITENESS_DELAY_MS = 300; + +export interface ProPublicaOrganizationSummary { + readonly ein: number; + readonly name: string; + readonly city: string | null; + readonly state: string | null; + readonly ntee_code: string | null; +} + +export interface ProPublicaSearchResponse { + readonly organizations: ReadonlyArray<ProPublicaOrganizationSummary>; +} + +export interface ProPublicaOrganizationDetail { + readonly ein: number; + readonly name: string; + readonly ntee_code: string | null; +} + +export interface ProPublicaFiling { + readonly totrevenue: number | null; + readonly tax_prd_yr: number; + readonly tax_prd: number; +} + +export interface ProPublicaOrganizationResponse { + readonly organization: ProPublicaOrganizationDetail; + readonly filings_with_data: ReadonlyArray<ProPublicaFiling>; +} + +async function wait(ms: number): Promise<void> { + if (ms <= 0) return; + await new Promise((resolve) => setTimeout(resolve, ms)); +} + +export interface ProPublicaClientOptions { + /** Delay awaited between successive network calls. Defaults to ~300ms. */ + readonly politenessDelayMs?: number; +} + +/** + * Search Nonprofit Explorer for organizations by name, optionally scoped to + * a state. Throws on any non-2xx response. + */ +export async function searchOrganizations( + name: string, + state = 'NH', + options: ProPublicaClientOptions = {}, +): Promise<ProPublicaSearchResponse> { + const url = new URL(`${BASE_URL}/search.json`); + url.searchParams.set('q', name); + url.searchParams.set('state[id]', state); + + const res = await fetch(url); + if (!res.ok) { + throw new Error( + `ProPublica search failed for "${name}" (${state}): ${res.status} ${res.statusText}`, + ); + } + const body = (await res.json()) as ProPublicaSearchResponse; + + await wait(options.politenessDelayMs ?? DEFAULT_POLITENESS_DELAY_MS); + + return body; +} + +/** + * Fetch full detail (including filing history) for a single organization by + * EIN. Throws on any non-2xx response. + */ +export async function getOrganization( + ein: number, + options: ProPublicaClientOptions = {}, +): Promise<ProPublicaOrganizationResponse> { + const url = new URL(`${BASE_URL}/organizations/${ein}.json`); + + const res = await fetch(url); + if (!res.ok) { + throw new Error( + `ProPublica organization lookup failed for EIN ${ein}: ${res.status} ${res.statusText}`, + ); + } + const body = (await res.json()) as ProPublicaOrganizationResponse; + + await wait(options.politenessDelayMs ?? DEFAULT_POLITENESS_DELAY_MS); + + return body; +} diff --git a/apps/outreach-worker/src/sources/propublica/extract.test.ts b/apps/outreach-worker/src/sources/propublica/extract.test.ts new file mode 100644 index 0000000..020d0f8 --- /dev/null +++ b/apps/outreach-worker/src/sources/propublica/extract.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it } from 'vitest'; + +import { extractEnrichment } from './extract.js'; +import type { ProPublicaOrganizationResponse } from './client.js'; + +function orgDetail( + overrides: Partial<ProPublicaOrganizationResponse>, +): ProPublicaOrganizationResponse { + return { + organization: { ein: 12345678, name: 'Test Org', ntee_code: 'P20' }, + filings_with_data: [], + ...overrides, + }; +} + +describe('extractEnrichment', () => { + it('zero-pads the EIN to a 9-digit string', () => { + const detail = orgDetail({ + organization: { ein: 12345678, name: 'Test Org', ntee_code: null }, + }); + expect(extractEnrichment(detail).ein).toBe('012345678'); + }); + + it('does not pad an already-9-digit EIN', () => { + const detail = orgDetail({ + organization: { ein: 123456789, name: 'Test Org', ntee_code: null }, + }); + expect(extractEnrichment(detail).ein).toBe('123456789'); + }); + + it('passes through the ntee code, null when absent', () => { + const withCode = orgDetail({ + organization: { ein: 12345678, name: 'Test Org', ntee_code: 'P20' }, + }); + expect(extractEnrichment(withCode).nteeCode).toBe('P20'); + + const withoutCode = orgDetail({ + organization: { ein: 12345678, name: 'Test Org', ntee_code: null }, + }); + expect(extractEnrichment(withoutCode).nteeCode).toBeNull(); + }); + + it('returns null revenue and fiscal year end when filings_with_data is empty', () => { + const detail = orgDetail({ filings_with_data: [] }); + const result = extractEnrichment(detail); + expect(result.totalRevenue).toBeNull(); + expect(result.fiscalYearEnd).toBeNull(); + }); + + it('selects the most recent filing by tax_prd_yr for revenue and fiscal year end', () => { + const detail = orgDetail({ + filings_with_data: [ + { totrevenue: 100_000, tax_prd_yr: 2021, tax_prd: 202112 }, + { totrevenue: 250_000, tax_prd_yr: 2023, tax_prd: 202306 }, + { totrevenue: 175_000, tax_prd_yr: 2022, tax_prd: 202209 }, + ], + }); + const result = extractEnrichment(detail); + expect(result.totalRevenue).toBe(250_000); + expect(result.fiscalYearEnd).toBe('06'); + }); + + it('derives the fiscal year end month from tax_prd (YYYYMM)', () => { + const detail = orgDetail({ + filings_with_data: [{ totrevenue: 500_000, tax_prd_yr: 2023, tax_prd: 202306 }], + }); + expect(extractEnrichment(detail).fiscalYearEnd).toBe('06'); + }); + + it('handles a null totrevenue on the most recent filing', () => { + const detail = orgDetail({ + filings_with_data: [{ totrevenue: null, tax_prd_yr: 2023, tax_prd: 202312 }], + }); + expect(extractEnrichment(detail).totalRevenue).toBeNull(); + }); +}); diff --git a/apps/outreach-worker/src/sources/propublica/extract.ts b/apps/outreach-worker/src/sources/propublica/extract.ts new file mode 100644 index 0000000..2c0518d --- /dev/null +++ b/apps/outreach-worker/src/sources/propublica/extract.ts @@ -0,0 +1,42 @@ +/** + * Pure extraction of enrichment fields from a ProPublica Nonprofit Explorer + * organization detail response. + */ +import type { ProPublicaFiling, ProPublicaOrganizationResponse } from './client.js'; + +export interface OrgEnrichmentFields { + readonly ein: string; + readonly nteeCode: string | null; + readonly totalRevenue: number | null; + readonly fiscalYearEnd: string | null; +} + +function mostRecentFiling( + filings: ReadonlyArray<ProPublicaFiling>, +): ProPublicaFiling | null { + if (filings.length === 0) return null; + return filings.reduce((latest, filing) => + filing.tax_prd_yr > latest.tax_prd_yr ? filing : latest, + ); +} + +/** Derive the 'MM' fiscal year-end month from a `tax_prd` in YYYYMM form. */ +function fiscalMonthFromTaxPrd(taxPrd: number): string | null { + const month = taxPrd % 100; + if (month < 1 || month > 12) return null; + return String(month).padStart(2, '0'); +} + +export function extractEnrichment( + orgDetail: ProPublicaOrganizationResponse, +): OrgEnrichmentFields { + const ein = String(orgDetail.organization.ein).padStart(9, '0'); + const nteeCode = orgDetail.organization.ntee_code ?? null; + + const latestFiling = mostRecentFiling(orgDetail.filings_with_data); + const totalRevenue = latestFiling?.totrevenue ?? null; + const fiscalYearEnd = + latestFiling == null ? null : fiscalMonthFromTaxPrd(latestFiling.tax_prd); + + return { ein, nteeCode, totalRevenue, fiscalYearEnd }; +} diff --git a/apps/outreach-worker/src/sources/propublica/match.test.ts b/apps/outreach-worker/src/sources/propublica/match.test.ts new file mode 100644 index 0000000..330c362 --- /dev/null +++ b/apps/outreach-worker/src/sources/propublica/match.test.ts @@ -0,0 +1,110 @@ +import { describe, expect, it } from 'vitest'; + +import { normalizeOrgName, pickBestMatch } from './match.js'; + +describe('normalizeOrgName', () => { + it('lowercases and strips punctuation', () => { + expect(normalizeOrgName("St. Mary's Food Pantry, Inc.")).toBe('st marys food pantry'); + }); + + it('drops noise tokens: legal suffixes and articles', () => { + expect(normalizeOrgName('The Nashua Corp of Charity Incorporated')).toBe( + 'nashua charity', + ); + }); + + it('drops the "nh" token and the "new hampshire" phrase', () => { + expect(normalizeOrgName('NH Food Bank')).toBe('food bank'); + expect(normalizeOrgName('New Hampshire Food Bank')).toBe('food bank'); + }); + + it('does not drop legitimate tokens that merely resemble noise phrases', () => { + // "new" is only stripped as part of the "new hampshire" phrase, not on + // its own - a name like "New Beginnings" must keep it. + expect(normalizeOrgName('New Beginnings Shelter')).toBe('new beginnings shelter'); + }); + + it('collapses repeated whitespace', () => { + expect(normalizeOrgName(' Manchester Youth Services ')).toBe( + 'manchester youth services', + ); + }); +}); + +describe('pickBestMatch', () => { + const target = { name: 'Manchester Youth Services Inc', city: 'Manchester' }; + + it('returns null when there are no candidates', () => { + expect(pickBestMatch(target, [])).toBeNull(); + }); + + it('picks an exact normalized-name match', () => { + const candidates = [ + { name: 'Manchester Youth Services', city: 'Manchester' }, + { name: 'Something Entirely Different', city: 'Nashua' }, + ]; + expect(pickBestMatch(target, candidates)).toBe(candidates[0]); + }); + + it('rejects a candidate below the 0.8 Jaccard threshold', () => { + // target tokens {manchester, youth, services}; candidate tokens + // {manchester, youth, services, program} => 3/4 = 0.75, below threshold. + const candidates = [{ name: 'Manchester Youth Services Program', city: 'Manchester' }]; + expect(pickBestMatch(target, candidates)).toBeNull(); + }); + + it('accepts a candidate at/above the 0.8 Jaccard threshold', () => { + // Same token set, different word order: string equality fails (so this + // exercises the Jaccard path, not the exact-match branch), but the + // token sets are identical => Jaccard 1.0. + const reorderedTarget = { name: 'Youth Services Manchester', city: 'Manchester' }; + const candidates = [{ name: 'Manchester Services Youth', city: 'Manchester' }]; + expect(pickBestMatch(reorderedTarget, candidates)).toBe(candidates[0]); + }); + + it('disqualifies a candidate on city mismatch even with strong name similarity', () => { + const candidates = [{ name: 'Manchester Youth Services', city: 'Nashua' }]; + expect(pickBestMatch(target, candidates)).toBeNull(); + }); + + it('does not disqualify on city when the target has no city on file', () => { + const noCityTarget = { name: 'Manchester Youth Services Inc', city: null }; + const candidates = [{ name: 'Manchester Youth Services', city: 'Nashua' }]; + expect(pickBestMatch(noCityTarget, candidates)).toBe(candidates[0]); + }); + + it('does not disqualify when the candidate has no city on file', () => { + const candidates = [{ name: 'Manchester Youth Services', city: null }]; + expect(pickBestMatch(target, candidates)).toBe(candidates[0]); + }); + + it('returns null (never a guess) when all candidates are weak matches', () => { + const weakTarget = { name: 'Community Support Alliance', city: null }; + const candidates = [ + { name: 'Regional Health Network', city: null }, + { name: 'Downtown Arts Collective', city: null }, + ]; + expect(pickBestMatch(weakTarget, candidates)).toBeNull(); + }); + + it('breaks ties between multiple qualifying candidates by city match', () => { + const tieTarget = { name: 'Riverside Community Center', city: 'Concord' }; + const candidates = [ + { name: 'Riverside Community Center', city: 'Nashua' }, + { name: 'Riverside Community Center', city: 'Concord' }, + ]; + expect(pickBestMatch(tieTarget, candidates)).toBe(candidates[1]); + }); + + it('breaks ties between same-city, non-exact candidates by shortest name distance', () => { + // Both candidates clear the Jaccard threshold (neither is an exact + // normalized-name match to the target), so this exercises the distance + // tie-break rather than the exact-match short-circuit. + const tieTarget = { name: 'Alpha Beta Gamma Delta', city: 'Concord' }; + const candidates = [ + { name: 'Alpha Beta Gamma Delta Epsilon', city: 'Concord' }, + { name: 'Delta Gamma Beta Alpha', city: 'Concord' }, + ]; + expect(pickBestMatch(tieTarget, candidates)).toBe(candidates[0]); + }); +}); diff --git a/apps/outreach-worker/src/sources/propublica/match.ts b/apps/outreach-worker/src/sources/propublica/match.ts new file mode 100644 index 0000000..55fb2f9 --- /dev/null +++ b/apps/outreach-worker/src/sources/propublica/match.ts @@ -0,0 +1,175 @@ +/** + * Pure name+city fuzzy matching between an org in our schema and a set of + * ProPublica Nonprofit Explorer search candidates. + * + * Deliberately conservative: a wrong EIN poisons revenue/ICP-band scoring + * downstream for that org (and any outreach built on it), and there's no + * cheap way to detect the mistake later since the enrichment queue treats + * "resolved" as done. A missed match just stays in the enrichment backlog + * to retry later — recoverable. A false-positive match silently corrupts + * data — not recoverable without a manual audit. So every ambiguous case + * resolves to `null` rather than a best-effort guess. + */ + +const NOISE_TOKENS = new Set(['inc', 'incorporated', 'corp', 'corporation', 'the', 'of', 'nh']); +// Multi-word noise phrase — stripped before single-token filtering so we +// don't accidentally drop legitimate tokens like "new" in "New Hope Center". +const NOISE_PHRASES = ['new hampshire']; + +/** + * Lowercase, strip punctuation, drop noise tokens (legal suffixes, articles, + * and state-name boilerplate that's redundant given we already filter by + * state), collapse whitespace. + */ +export function normalizeOrgName(name: string): string { + // Drop apostrophes without inserting a space ("Mary's" -> "marys"); every + // other punctuation character becomes a space-separator. + let lowered = name.toLowerCase().replace(/['’]/g, ''); + lowered = lowered.replace(/[^\w\s]/g, ' '); + for (const phrase of NOISE_PHRASES) { + lowered = lowered.split(phrase).join(' '); + } + const stripped = lowered + .split(/\s+/) + .filter((token) => token.length > 0 && !NOISE_TOKENS.has(token)); + return stripped.join(' ').trim(); +} + +function tokenSet(name: string): Set<string> { + return new Set(normalizeOrgName(name).split(' ').filter(Boolean)); +} + +function jaccardSimilarity(a: Set<string>, b: Set<string>): number { + if (a.size === 0 && b.size === 0) return 1; + if (a.size === 0 || b.size === 0) return 0; + let intersectionSize = 0; + for (const token of a) { + if (b.has(token)) intersectionSize += 1; + } + const unionSize = a.size + b.size - intersectionSize; + return unionSize === 0 ? 0 : intersectionSize / unionSize; +} + +function normalizedCity(city: string | null | undefined): string | null { + if (city == null) return null; + const trimmed = city.trim().toLowerCase(); + return trimmed.length > 0 ? trimmed : null; +} + +/** Simple Levenshtein distance, used only as a tie-breaker. */ +function levenshtein(a: string, b: string): number { + const m = a.length; + const n = b.length; + if (m === 0) return n; + if (n === 0) return m; + + let prevRow: number[] = Array.from({ length: n + 1 }, (_, j) => j); + for (let i = 1; i <= m; i += 1) { + const currRow: number[] = [i]; + for (let j = 1; j <= n; j += 1) { + // Indices are always in-bounds by the loop bounds above (0..n on + // prevRow, 0..j on currRow-so-far); non-null assertions just work + // around `noUncheckedIndexedAccess` not doing that range analysis. + const cost = a[i - 1] === b[j - 1] ? 0 : 1; + currRow.push( + Math.min( + prevRow[j]! + 1, // deletion + currRow[j - 1]! + 1, // insertion + prevRow[j - 1]! + cost, // substitution + ), + ); + } + prevRow = currRow; + } + return prevRow[n]!; +} + +const JACCARD_THRESHOLD = 0.8; + +export interface MatchTarget { + readonly name: string; + readonly city: string | null; +} + +export interface MatchCandidate { + readonly name: string; + readonly city: string | null; +} + +/** + * Pick the best candidate for `target`, or `null` when nothing is a + * confident enough match. See module doc comment for why `null` is the + * safe default. + */ +export function pickBestMatch<T extends MatchCandidate>( + target: MatchTarget, + candidates: ReadonlyArray<T>, +): T | null { + if (candidates.length === 0) return null; + + const targetNormalizedName = normalizeOrgName(target.name); + const targetCity = normalizedCity(target.city); + const targetTokens = tokenSet(target.name); + + // When the target has a known city, any candidate with a known-but- + // different city is disqualified outright (a strong "not this org" signal + // that outweighs name similarity) — this applies to BOTH the exact-name + // and Jaccard paths below. Two orgs can legitimately share an exact legal + // name across different towns (e.g. chapters of the same national org), + // and a wrong EIN there is exactly the poisoned-data case this function + // is designed to avoid. + const cityCompatible = (candidate: MatchCandidate): boolean => { + const candidateCity = normalizedCity(candidate.city); + return !(targetCity != null && candidateCity != null && candidateCity !== targetCity); + }; + + // Exact normalized-name match wins, among city-compatible candidates. + const exactMatches = candidates.filter( + (candidate) => + normalizeOrgName(candidate.name) === targetNormalizedName && cityCompatible(candidate), + ); + if (exactMatches.length === 1) return exactMatches[0]!; + if (exactMatches.length > 1) { + return pickByCityThenDistance(target, targetCity, exactMatches); + } + + // Otherwise, token-set Jaccard similarity must clear the threshold. + const eligible = candidates.filter( + (candidate) => + cityCompatible(candidate) && + jaccardSimilarity(targetTokens, tokenSet(candidate.name)) >= JACCARD_THRESHOLD, + ); + + if (eligible.length === 0) return null; + if (eligible.length === 1) return eligible[0]!; + + return pickByCityThenDistance(target, targetCity, eligible); +} + +function pickByCityThenDistance<T extends MatchCandidate>( + target: MatchTarget, + targetCity: string | null, + candidates: ReadonlyArray<T>, +): T { + const targetNormalizedName = normalizeOrgName(target.name); + + // Prefer candidates whose city matches the target's, when known. + const cityMatches = + targetCity != null + ? candidates.filter((c) => normalizedCity(c.city) === targetCity) + : []; + const pool = cityMatches.length > 0 ? cityMatches : candidates; + // `candidates` is guaranteed non-empty by both call sites (exactMatches + // and eligible are only routed here once their length is > 1), so `pool` + // is always non-empty too. + const [first, ...rest] = pool; + + return rest.reduce((best, candidate) => { + const bestDistance = levenshtein(normalizeOrgName(best.name), targetNormalizedName); + const candidateDistance = levenshtein( + normalizeOrgName(candidate.name), + targetNormalizedName, + ); + return candidateDistance < bestDistance ? candidate : best; + }, first!); +} diff --git a/apps/outreach-worker/src/workflows/enrich-orgs.ts b/apps/outreach-worker/src/workflows/enrich-orgs.ts new file mode 100644 index 0000000..4bd80ce --- /dev/null +++ b/apps/outreach-worker/src/workflows/enrich-orgs.ts @@ -0,0 +1,206 @@ +/** + * Daily IRS enrichment sweep. + * + * Pulls a bounded batch of orgs missing EIN/revenue data, resolves each + * against ProPublica's Nonprofit Explorer (search by name/state, fuzzy-match + * on name+city, fetch filing detail for the matched EIN), and writes the + * result back via `serverEnrichOrg` — including an explicit all-null write + * for orgs that fail to resolve, so the queue doesn't retry them every run. + * + * See `ingest-grants.ts` for the registration pattern this mirrors (deps + * registry + dual workflow/scheduled registration, both referencing the same + * function object). + * + * Step granularity: each ProPublica network call (search, org detail) and + * each `serverEnrichOrg` write is its own DBOS step, called in a loop from + * the workflow body rather than wrapping the whole batch in one step. DBOS + * checkpoints after every step completes, so if the process dies mid-batch + * (or a single org's calls exhaust their retries), a workflow replay resumes + * from the last completed step instead of re-querying ProPublica for orgs + * already resolved earlier in the run. A single big-batch step would trade + * that away: any failure deep in a 200-org batch would force the whole batch + * to redo its network calls from scratch on retry. The calls are also each + * individually idempotent/retryable (GETs; `serverEnrichOrg` is a keyed + * update), which is what step-level `retriesAllowed` assumes. + */ +import { DBOS, SchedulerMode } from '@dbos-inc/dbos-sdk'; +import type { schema } from '@novelpad/outreach-core'; +import { + serverEnrichOrg, + serverListOrgsNeedingEnrichment, + type OrgEnrichmentInput, + type OrgNeedingEnrichment, +} from '@novelpad/outreach-core/server'; +import type { NodePgDatabase } from 'drizzle-orm/node-postgres'; + +import { + getOrganization, + searchOrganizations, + type ProPublicaOrganizationSummary, +} from '../sources/propublica/client.js'; +import { extractEnrichment } from '../sources/propublica/extract.js'; +import { pickBestMatch } from '../sources/propublica/match.js'; + +export type OutreachDb = NodePgDatabase<typeof schema>; + +export interface EnrichOrgsDeps { + readonly db: OutreachDb; +} + +let registeredDeps: EnrichOrgsDeps | null = null; + +export function setEnrichOrgsDeps(deps: EnrichOrgsDeps): void { + registeredDeps = deps; +} + +function getEnrichOrgsDeps(): EnrichOrgsDeps { + if (registeredDeps == null) { + throw new Error( + 'EnrichOrgsDeps not registered. Call setEnrichOrgsDeps() before DBOS.launch().', + ); + } + return registeredDeps; +} + +const BATCH_LIMIT = 200; +// Above this fraction of per-org failures, treat the batch as a systemic +// problem (e.g. ProPublica is down) rather than a handful of bad orgs, and +// fail the workflow run so DBOS surfaces it for retry/visibility. +const FAILURE_THRESHOLD_RATIO = 0.2; + +const ALL_NULL_ENRICHMENT: OrgEnrichmentInput = { + ein: null, + nteeCode: null, + totalRevenue: null, + fiscalYearEnd: null, +}; + +async function listOrgs(db: OutreachDb): Promise<OrgNeedingEnrichment[]> { + return serverListOrgsNeedingEnrichment(db, { limit: BATCH_LIMIT }); +} +const listOrgsStep = DBOS.registerStep(listOrgs, { + name: 'listOrgsNeedingEnrichment', + retriesAllowed: true, + maxAttempts: 3, +}); + +async function searchOrg( + name: string, + state: string, +): Promise<ReadonlyArray<ProPublicaOrganizationSummary>> { + const { organizations } = await searchOrganizations(name, state); + return organizations; +} +const searchOrgStep = DBOS.registerStep(searchOrg, { + name: 'searchProPublicaOrg', + retriesAllowed: true, + maxAttempts: 3, +}); + +async function fetchOrgDetail(ein: number) { + return getOrganization(ein); +} +const fetchOrgDetailStep = DBOS.registerStep(fetchOrgDetail, { + name: 'fetchProPublicaOrgDetail', + retriesAllowed: true, + maxAttempts: 3, +}); + +async function writeEnrichment( + db: OutreachDb, + orgId: string, + enrichment: OrgEnrichmentInput, +): Promise<void> { + await serverEnrichOrg(db, orgId, enrichment); +} +const writeEnrichmentStep = DBOS.registerStep(writeEnrichment, { + name: 'writeOrgEnrichment', + retriesAllowed: true, + maxAttempts: 3, +}); + +interface BatchSummary { + attempted: number; + resolved: number; + unresolved: number; + failed: number; +} + +async function enrichOneOrg( + db: OutreachDb, + org: OrgNeedingEnrichment, +): Promise<'resolved' | 'unresolved'> { + const candidates = await searchOrgStep(org.name, org.state); + const match = pickBestMatch({ name: org.name, city: org.city }, candidates); + + if (match == null) { + await writeEnrichmentStep(db, org.id, ALL_NULL_ENRICHMENT); + return 'unresolved'; + } + + const detail = await fetchOrgDetailStep(match.ein); + const enrichment = extractEnrichment(detail); + await writeEnrichmentStep(db, org.id, enrichment); + return 'resolved'; +} + +async function runEnrichOrgs(): Promise<void> { + const { db } = getEnrichOrgsDeps(); + const orgs = await listOrgsStep(db); + + const summary: BatchSummary = { attempted: 0, resolved: 0, unresolved: 0, failed: 0 }; + + // Sequential for-of, NOT forEach+async: each org's calls must complete + // (including the politeness delay baked into the client) before the next + // org starts, both to respect ProPublica's rate limits and so the + // try/catch below actually isolates one org's failure from the next. + for (const org of orgs) { + summary.attempted += 1; + try { + const outcome = await enrichOneOrg(db, org); + if (outcome === 'resolved') summary.resolved += 1; + else summary.unresolved += 1; + } catch (err) { + // One bad org must not kill the batch — log and continue. Systemic + // failure is caught by the ratio check below instead. + console.error(`[enrich-orgs] failed to enrich org "${org.name}" (${org.id}):`, err); + summary.failed += 1; + } + } + + console.log( + `[enrich-orgs] batch summary: attempted=${summary.attempted} resolved=${summary.resolved} unresolved=${summary.unresolved} failed=${summary.failed}`, + ); + + if (summary.attempted > 0 && summary.failed / summary.attempted > FAILURE_THRESHOLD_RATIO) { + throw new Error( + `[enrich-orgs] systemic failure: ${summary.failed}/${summary.attempted} orgs failed enrichment (>${FAILURE_THRESHOLD_RATIO * 100}%)`, + ); + } +} + +const g = globalThis as unknown as { + __outreachEnrichOrgsRegistered?: boolean; +}; + +if (!g.__outreachEnrichOrgsRegistered) { + g.__outreachEnrichOrgsRegistered = true; + + const enrichOrgs = async (_scheduledTime: Date, _startedAt: Date) => { + try { + await runEnrichOrgs(); + } catch (err) { + console.error('[enrich-orgs] 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(enrichOrgs, { name: 'enrichOrgs' }); + DBOS.registerScheduled(enrichOrgs, { + crontab: '0 5 * * *', + name: 'enrichOrgs', + mode: SchedulerMode.ExactlyOncePerInterval, + }); +} diff --git a/apps/outreach-worker/src/workflows/ingest-grants.ts b/apps/outreach-worker/src/workflows/ingest-grants.ts index c7cdd25..f613576 100644 --- a/apps/outreach-worker/src/workflows/ingest-grants.ts +++ b/apps/outreach-worker/src/workflows/ingest-grants.ts @@ -20,19 +20,32 @@ import { } from '@novelpad/outreach-core/server'; import type { NodePgDatabase } from 'drizzle-orm/node-postgres'; +import { + fetchOpportunityDetails, + searchOpportunities, + type GrantsGovOpportunityDetail, + type GrantsGovSearchHit, +} from '#~/sources/grants-gov/client.js'; +import { normalizeGrantsGovOpportunity } from '#~/sources/grants-gov/normalize.js'; + export type OutreachDb = NodePgDatabase<typeof schema>; -/** Shape of a single opportunity as returned by the Grants.gov Search2 API. */ -export interface RawGrantsGovOpportunity { - readonly opportunityId: string; - readonly opportunityNumber: string; - readonly opportunityTitle: string; - readonly agencyCode: string | null; - readonly openDate: string | null; - readonly closeDate: string | null; - readonly awardCeiling: number | null; - readonly awardFloor: number | null; -} +/** + * Cap on how many opportunities we'll enumerate from Grants.gov's `search2` + * endpoint per run. Search hits are cheap (one paginated request per ~100), + * so this is generous headroom above the actual nightly posting volume. + */ +const SEARCH_HIT_CAP = 1000; +/** + * Cap on how many opportunities get a `fetchOpportunity` detail call per + * run. Detail fetches are one request per grant (plus a politeness delay), + * so this bounds both the run time and the load we put on Grants.gov. + * Opportunities beyond the cap are simply left for the next nightly run — + * see the console.warn below when it binds. + */ +const DETAIL_FETCH_CAP = 200; +/** Batch size for `serverInsertGrants` upserts. */ +const UPSERT_BATCH_SIZE = 100; export interface IngestGrantsDeps { readonly db: OutreachDb; @@ -53,48 +66,76 @@ function getIngestGrantsDeps(): IngestGrantsDeps { return registeredDeps; } -/** - * Fetch open opportunities from Grants.gov. - * - * Real implementation calls the Grants.gov Search2 API - * (`POST https://api.grants.gov/v1/api/search2`, JSON body with - * `oppStatuses: "posted"` and pagination via `rows`/`startRecordNum`). Stubbed - * to an empty array until that integration lands. - */ -async function fetchGrantsGov(): Promise<ReadonlyArray<RawGrantsGovOpportunity>> { - // TODO(outreach): call Grants.gov Search2 API and paginate through results. - return []; +/** Step 1: enumerate open ("posted") nonprofit-eligible opportunities. */ +async function searchGrantsGov(): Promise<ReadonlyArray<GrantsGovSearchHit>> { + const hits = await searchOpportunities({ maxHits: SEARCH_HIT_CAP }); + console.log( + `[ingest-grants] search2 returned ${hits.length} open opportunities (cap ${SEARCH_HIT_CAP})`, + ); + return hits; } -const fetchGrantsGovStep = DBOS.registerStep(fetchGrantsGov, { - name: 'fetchGrantsGov', +const searchGrantsGovStep = DBOS.registerStep(searchGrantsGov, { + name: 'searchGrantsGov', retriesAllowed: true, maxAttempts: 3, }); -function normalize( - raw: ReadonlyArray<RawGrantsGovOpportunity>, -): ReadonlyArray<NewGrantInput> { - return raw.map((opp) => ({ - // Stable identity across re-crawls; also the schema's upsert key. - sourceUrl: `https://www.grants.gov/search-results-detail/${opp.opportunityId}`, - source: 'grants_gov' as const, - title: opp.opportunityTitle, - funder: opp.agencyCode ?? 'Unknown federal agency', - openDate: opp.openDate == null ? null : new Date(opp.openDate), - closeDate: opp.closeDate == null ? null : new Date(opp.closeDate), - // Schema stores whole dollars (award amounts are never sub-dollar). - awardFloor: opp.awardFloor == null ? null : Math.round(opp.awardFloor), - awardCeiling: opp.awardCeiling == null ? null : Math.round(opp.awardCeiling), - lastVerifiedAt: new Date(), - })); +interface HitWithDetail { + readonly hit: GrantsGovSearchHit; + readonly detail: GrantsGovOpportunityDetail; } +/** + * Step 2: fetch full detail (synopsis, eligibility, award amounts) for up + * to `DETAIL_FETCH_CAP` of the search hits. Truncation is never silent — + * when the cap binds we log exactly how many opportunities were dropped + * this run. + */ +async function fetchGrantDetails( + hits: ReadonlyArray<GrantsGovSearchHit>, +): Promise<ReadonlyArray<HitWithDetail>> { + const toFetch = hits.slice(0, DETAIL_FETCH_CAP); + if (hits.length > toFetch.length) { + console.warn( + `[ingest-grants] detail-fetch cap ${DETAIL_FETCH_CAP} reached: dropping ${hits.length - toFetch.length} of ${hits.length} opportunities this run`, + ); + } + + const details = await fetchOpportunityDetails(toFetch.map((hit) => hit.id)); + return toFetch.map((hit, i) => { + const detail = details[i]; + if (detail == null) { + throw new Error( + `[ingest-grants] missing detail response for opportunity ${hit.id} at index ${i}`, + ); + } + return { hit, detail }; + }); +} +const fetchGrantDetailsStep = DBOS.registerStep(fetchGrantDetails, { + name: 'fetchGrantDetails', + retriesAllowed: true, + maxAttempts: 3, +}); + +/** Step 3: pure normalization from wire shapes to insertable grant rows. */ +function normalize( + pairs: ReadonlyArray<HitWithDetail>, +): ReadonlyArray<NewGrantInput> { + return pairs.map(({ hit, detail }) => + normalizeGrantsGovOpportunity(hit, detail), + ); +} + +/** Step 4: upsert normalized grants in batches of `UPSERT_BATCH_SIZE`. */ async function upsertGrants( db: OutreachDb, grants: ReadonlyArray<NewGrantInput>, ): Promise<void> { - if (grants.length === 0) return; - await serverInsertGrants(db, grants); + for (let i = 0; i < grants.length; i += UPSERT_BATCH_SIZE) { + const batch = grants.slice(i, i + UPSERT_BATCH_SIZE); + await serverInsertGrants(db, batch); + } } const upsertGrantsStep = DBOS.registerStep(upsertGrants, { name: 'upsertGrants', @@ -105,8 +146,9 @@ const upsertGrantsStep = DBOS.registerStep(upsertGrants, { async function runIngestGrants(): Promise<void> { const { db } = getIngestGrantsDeps(); - const raw = await fetchGrantsGovStep(); - const normalized = normalize(raw); + const hits = await searchGrantsGovStep(); + const pairs = await fetchGrantDetailsStep(hits); + const normalized = normalize(pairs); await upsertGrantsStep(db, normalized); } diff --git a/apps/outreach-worker/src/workflows/ingest-nhdoj-orgs.ts b/apps/outreach-worker/src/workflows/ingest-nhdoj-orgs.ts new file mode 100644 index 0000000..bef0224 --- /dev/null +++ b/apps/outreach-worker/src/workflows/ingest-nhdoj-orgs.ts @@ -0,0 +1,141 @@ +/** + * Monthly NHDOJ Charitable Trusts registry ingestion workflow. + * + * The NH Department of Justice Charitable Trusts Unit publishes a PDF + * roster of registered charitable organizations. Unlike Grants.gov/PND/ + * ProPublica this isn't a grants feed — it's an org registry re-scan, so a + * fresh registrant (`inserted: true` from `serverUpsertOrgFromRegistry`) is + * itself the signal: it's a segment often actively seeking first-time + * funding. + * + * Registration follows `ingest-grants.ts` exactly: 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 { serverUpsertOrgFromRegistry } from '@novelpad/outreach-core/server'; +import type { NodePgDatabase } from 'drizzle-orm/node-postgres'; + +import { extractPositionedText } from '#~/sources/nhdoj/extract-pdf-text.js'; +import { + normalizeRegistryRows, + reconstructRegistryRows, + type NormalizedRegistryRow, +} from '#~/sources/nhdoj/parse-registry.js'; + +export type OutreachDb = NodePgDatabase<typeof schema>; + +const SOURCE_REGISTRY = 'nhdoj_charitable_trusts'; + +export interface IngestNhdojOrgsDeps { + readonly db: OutreachDb; +} + +let registeredDeps: IngestNhdojOrgsDeps | null = null; + +export function setIngestNhdojOrgsDeps(deps: IngestNhdojOrgsDeps): void { + registeredDeps = deps; +} + +function getIngestNhdojOrgsDeps(): IngestNhdojOrgsDeps { + if (registeredDeps == null) { + throw new Error( + 'IngestNhdojOrgsDeps not registered. Call setIngestNhdojOrgsDeps() before DBOS.launch().', + ); + } + return registeredDeps; +} + +async function fetchNhdojRegistryPdf(url: string): Promise<Uint8Array> { + const res = await fetch(url); + if (!res.ok) { + throw new Error( + `fetchNhdojRegistryPdf: ${url} responded ${res.status} ${res.statusText}`, + ); + } + const buf = await res.arrayBuffer(); + return new Uint8Array(buf); +} +const fetchNhdojRegistryPdfStep = DBOS.registerStep(fetchNhdojRegistryPdf, { + name: 'fetchNhdojRegistryPdf', + retriesAllowed: true, + maxAttempts: 3, +}); + +async function upsertRegistryOrg( + db: OutreachDb, + row: NormalizedRegistryRow, +): Promise<{ id: string; inserted: boolean }> { + return serverUpsertOrgFromRegistry(db, { + name: row.name, + city: row.city, + registrationStatus: row.status, + sourceRegistry: SOURCE_REGISTRY, + }); +} +const upsertRegistryOrgStep = DBOS.registerStep(upsertRegistryOrg, { + name: 'upsertNhdojRegistryOrg', + retriesAllowed: true, + maxAttempts: 3, +}); + +async function runIngestNhdojOrgs(): Promise<void> { + const { db } = getIngestNhdojOrgsDeps(); + + const pdfUrl = process.env.NHDOJ_REGISTRY_PDF_URL; + if (pdfUrl == null || pdfUrl.trim() === '') { + // The registry PDF's URL changes whenever NHDOJ republishes it (no + // stable endpoint) — a hard failure here would page on-call for a + // config gap rather than a real outage, so skip quietly instead. + console.warn( + '[ingest-nhdoj-orgs] NHDOJ_REGISTRY_PDF_URL is not set; skipping this pass.', + ); + return; + } + + const pdfBytes = await fetchNhdojRegistryPdfStep(pdfUrl); + const pages = await extractPositionedText(pdfBytes); + const rawRows = reconstructRegistryRows(pages); + const normalizedRows = normalizeRegistryRows(rawRows); + + let upserted = 0; + let newlyInserted = 0; + for (const row of normalizedRows) { + const result = await upsertRegistryOrgStep(db, row); + upserted++; + if (result.inserted) newlyInserted++; + } + + console.log( + `[ingest-nhdoj-orgs] rowsParsed=${normalizedRows.length} upserted=${upserted} newlyInserted=${newlyInserted}`, + ); +} + +const g = globalThis as unknown as { + __outreachIngestNhdojOrgsRegistered?: boolean; +}; + +if (!g.__outreachIngestNhdojOrgsRegistered) { + g.__outreachIngestNhdojOrgsRegistered = true; + + const ingestNhdojOrgs = async (_scheduledTime: Date, _startedAt: Date) => { + try { + await runIngestNhdojOrgs(); + } catch (err) { + console.error('[ingest-nhdoj-orgs] 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(ingestNhdojOrgs, { name: 'ingestNhdojOrgs' }); + DBOS.registerScheduled(ingestNhdojOrgs, { + crontab: '0 4 1 * *', + name: 'ingestNhdojOrgs', + mode: SchedulerMode.ExactlyOncePerInterval, + }); +} diff --git a/apps/outreach-worker/src/workflows/ingest-pnd-rss.ts b/apps/outreach-worker/src/workflows/ingest-pnd-rss.ts new file mode 100644 index 0000000..56bd3c0 --- /dev/null +++ b/apps/outreach-worker/src/workflows/ingest-pnd-rss.ts @@ -0,0 +1,113 @@ +/** + * 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, + }); +} diff --git a/docs/features/ingestion.md b/docs/features/ingestion.md new file mode 100644 index 0000000..3e38039 --- /dev/null +++ b/docs/features/ingestion.md @@ -0,0 +1,96 @@ +# Data-spine ingestion (Phase 1) + +How grant opportunities and NH nonprofit orgs enter the outreach database. All ingestion runs as DBOS scheduled workflows in `apps/outreach-worker`; every source has a thin impure client, a pure normalize layer (unit-tested on fixtures), and a workflow that upserts through `@novelpad/outreach-core/server` actions — never inline DB calls. + +Grant upserts key on `grants.source_url`; org registry upserts key on case-insensitive `(name, city, state)`; IRS enrichment applies to org rows by id and re-bands ICP via `icpBandForRevenue` ($100K–$5M = primary). + +## Schedules + +| Workflow | Cron (UTC) | Source | +|---|---|---| +| `ingestGrants` | `0 3 * * *` | Grants.gov Search2 | +| `ingestPndRss` | `30 3 * * *` | Philanthropy News Digest RFP feed | +| `expireGrants` | `0 * * * *` | (sweep: `status='expired'` past `close_date`) | +| `enrichOrgs` | `0 5 * * *` | ProPublica Nonprofit Explorer | +| `ingestNhdojOrgs` | `0 4 1 * *` | NHDOJ Charitable Trusts registry PDF | + +## Sources + +<!-- Per-source sections assembled below --> + +### Grants.gov + +**Source**: `apps/outreach-worker/src/sources/grants-gov/` — `client.ts` (fetch), `normalize.ts` (pure transform), `pagination.ts` (pure pagination arithmetic). Wired into `apps/outreach-worker/src/workflows/ingest-grants.ts`. + +**Cadence**: Nightly, `0 3 * * *` (03:00 UTC), registered as a DBOS scheduled workflow (`ingestGrants`, `ExactlyOncePerInterval`). + +**API**: Public Grants.gov Search2 API, no key required. +- `POST https://api.grants.gov/v1/api/search2` — enumerates opportunities, filtered to `oppStatuses: 'posted'` and nonprofit eligibility codes `12|13` (501(c)(3) and non-501(c)(3) nonprofits). Paginated via `rows`/`startRecordNum`. +- `POST https://api.grants.gov/v1/api/fetchOpportunity` — full detail (`synopsis`: description, applicant types, award amounts, response date) for a single opportunity id. +- Both endpoints wrap responses in an `{ errorcode, msg, data }` envelope. Any non-2xx HTTP status or `errorcode !== 0` throws immediately with a descriptive message — the client never silently drops or swallows an upstream failure. + +**Caps** (per nightly run): +- Search enumeration: up to **1000** hits (`SEARCH_HIT_CAP`). Count logged via `console.log`. +- Detail fetch: up to **200** opportunities (`DETAIL_FETCH_CAP`), each a separate `fetchOpportunity` call with a **~250ms** politeness delay between requests. When the search result set exceeds the cap, the excess is dropped for that run (picked up on a later run) and a `console.warn` states exactly how many opportunities were skipped — never a silent truncation. +- Upsert: batched in groups of **100** via `serverInsertGrants`. + +**Normalization rules** (`normalizeGrantsGovOpportunity`, pure — no I/O): +- `sourceUrl` (the upsert key): `https://www.grants.gov/search-results-detail/{opportunityId}` — stable across re-crawls. +- `funder`: `agencyDetails.agencyName` → `synopsis.agencyName` → search hit's `agency` → an agency code fallback → literal `'Unknown federal agency'` if nothing is present. +- `synopsis`: HTML-stripped from `synopsis.synopsisDesc` (block tags become newlines, common entities decoded); `null` if empty after stripping. +- `awardFloor` / `awardCeiling` / `expectedAwardsCount`: parsed from string-or-number wire values (handles `"$1,500,000"`-style formatting), rounded to whole dollars/counts; `null` on unparseable input. +- `openDate` / `closeDate`: parsed from the search hit's `MM/DD/YYYY` fields (falling back to the detail's `postingDate`/`responseDate`); any invalid, out-of-range, or calendar-rollover date (e.g. `02/30/2026`) resolves to `null` rather than a garbage `Date`. +- `eligibilityEntityTypes`: trimmed, non-empty descriptions from `synopsis.applicantTypes`; `null` if none. +- `geographicScope`: always `null` — federal grants are national by default and Grants.gov has no reliable field to derive a narrower scope from; left unset rather than guessed. +- `programAreas`: `null` (not yet derived from this source). +- `applicationEffortEstimate`: `'unknown'`; `matchRequirement` / `applicationFormSupported`: `false`; `status`: `'open'`; `source`: `'grants_gov'`; `lastVerifiedAt`: set to the normalization timestamp. + +### Philanthropy News Digest RSS + +Nightly ingestion of the Philanthropy News Digest "RFPs" RSS feed (`https://philanthropynewsdigest.org/feeds/rfps.rss`, overridable via `PND_RFP_FEED_URL`). + +- **Client** — `src/sources/pnd-rss/client.ts`: `fetchPndRfpFeed(feedUrl)` does a plain `fetch`, throwing on any non-2xx response. +- **Normalize** — `src/sources/pnd-rss/normalize.ts`: `parsePndFeed(xml)` is a pure function (`fast-xml-parser`'s `XMLParser`, with `isArray` forcing `<item>` to always parse as an array so single-item feeds don't collapse to a bare object) that maps each RSS `<item>` to a `NewGrantInput`: + - `title` → `title`; `link` → `sourceUrl` (items with no link are skipped — there's nothing stable to upsert-key on) + - `description` → `synopsis`, after stripping HTML tags and decoding named + numeric entities + - `funder` — best-effort regex over the title, then the description: captures the leading proper-noun phrase before an announcing verb (`accepting` / `invites` / `seeks` / `announces` / `offers`); falls back to `'See RFP'` + - `closeDate` — best-effort regex for `Deadline: <date>` / `due <date>` phrasing (both month-name and ISO-ish date forms); invalid or absent → `null` + - `applicationEffortEstimate: 'unknown'`, `source: 'pnd_rss'`, `openDate: null`, `lastVerifiedAt: new Date()` +- **Workflow** — `src/workflows/ingest-pnd-rss.ts`: scheduled nightly at `30 3 * * *` (`ingestPndRss`), following the same registration pattern as `ingest-grants.ts` (module-scope deps registry populated via `setIngestPndRssDeps` before `DBOS.launch()`, dual workflow+scheduled registration, `globalThis` double-registration guard). Steps: fetch (3 retries) → parse (pure) → upsert via `serverInsertGrants` in batches of 100. + +### ProPublica Nonprofit Explorer enrichment + +Daily scheduled workflow (`enrichOrgs`, cron `0 5 * * *`) that fills in IRS-derived fields — EIN, NTEE code, most-recent-filing total revenue, fiscal year-end month — for orgs discovered by the other ingestion sources but never resolved against the IRS. + +**Flow:** +1. `serverListOrgsNeedingEnrichment(db, { limit: 200 })` — orgs with no EIN and no revenue on file, oldest-`updatedAt` first. +2. For each org, sequentially (never `forEach`+async — the client's politeness delay depends on awaiting each call before starting the next): + - `searchOrganizations(org.name, org.state)` against Nonprofit Explorer's `search.json`. + - `pickBestMatch({ name, city }, candidates)` — pure, conservative fuzzy match (see below). Returns `null` rather than guess when ambiguous. + - On no match: `serverEnrichOrg(db, org.id, { ein: null, nteeCode: null, totalRevenue: null, fiscalYearEnd: null })` — stamps `updatedAt` so the org drops out of the next run's `serverListOrgsNeedingEnrichment` result without pretending to have real data. `icpBandForRevenue` still runs inside `serverEnrichOrg` and correctly returns `'unknown'` for a null revenue. + - On match: `getOrganization(ein)` for filing history, `extractEnrichment(detail)`, then `serverEnrichOrg(db, org.id, enrichment)`. +3. Per-org failures are caught, logged, and counted — one bad org doesn't kill the batch. If failures exceed 20% of the attempted batch, the workflow rethrows (systemic-failure signal for DBOS retry/alerting) after logging attempted/resolved/unresolved/failed counts. + +**Matching (`src/sources/propublica/match.ts`):** `normalizeOrgName` lowercases, strips punctuation (apostrophes drop silently, other punctuation becomes a separator), strips legal-suffix/article noise tokens (`inc`, `corp`, `the`, `of`, `nh`, the phrase `new hampshire`), and collapses whitespace. `pickBestMatch` prefers an exact normalized-name match, falls back to token-set Jaccard similarity ≥ 0.8, and disqualifies any candidate whose known city differs from the target's known city — on *both* paths, since same-legal-name-different-town is exactly the ambiguous case worth refusing rather than guessing. Ties break by city match, then shortest Levenshtein distance on the normalized name. A wrong EIN silently poisons downstream revenue/ICP-band data with no cheap way to detect it later, so every ambiguous case resolves to `null` (org stays in next run's backlog) instead of a best-effort guess. + +**Extraction (`src/sources/propublica/extract.ts`):** `extractEnrichment` zero-pads the numeric EIN to 9 digits, passes through `ntee_code` (null-safe), and — from `filings_with_data` — picks the filing with the highest `tax_prd_yr` for `totalRevenue` and derives `fiscalYearEnd` as the zero-padded `MM` from that filing's `tax_prd` (`YYYYMM`, e.g. `202306` → `'06'`). Empty filing history yields `totalRevenue: null, fiscalYearEnd: null`. + +**Wiring:** `apps/outreach-worker/src/main.ts` imports `setEnrichOrgsDeps` from `./workflows/enrich-orgs.js` (which also registers the workflow/scheduled function as an import side effect) and calls `setEnrichOrgsDeps({ db })` before `DBOS.launch()`, alongside the existing `setIngestGrantsDeps`/`setExpireGrantsDeps` calls. + +### NHDOJ Charitable Trusts registry + +Monthly re-scan of the NH Department of Justice Charitable Trusts Unit's registry PDF — a roster of registered charitable organizations, not a grants feed. This is the org side of the pipeline: it upserts into `orgs` (keyed on case-insensitive `(name, city, state)`, same as the ProPublica enrichment source), not `grants`. A brand-new registrant (`inserted: true`) is itself a signal worth tracking downstream — it's a segment often actively seeking first-time funding. + +- **Extract** — `src/sources/nhdoj/extract-pdf-text.ts`: `extractPositionedText(pdfBytes)` is the only impure layer. Uses `pdfjs-dist`'s legacy Node build (`pdfjs-dist/legacy/build/pdf.mjs`) with `getDocument({ data, useSystemFonts: true })`, returning one `PositionedTextItem[]` (`{ str, x, y }`, taken from each text item's transform matrix `[4]`/`[5]`) per page. + - **Node-runtime caveat**: no `disableWorker` option exists on `DocumentInitParameters`, and none is needed — the legacy build self-detects `isNodeJS` at module load and unconditionally sets `PDFWorker.#isWorkerDisabled = true`, falling back to an in-process "fake worker" automatically (see `PDFWorker#_initialize` in the bundled `pdf.mjs`). No `GlobalWorkerOptions.workerSrc` setup is required for text extraction. +- **Parse** — `src/sources/nhdoj/parse-registry.ts`: pure row reconstruction over `PositionedTextItem[][]`, entirely unit-testable on synthetic fixtures (no real PDF needed). + - `reconstructRegistryRows(pages, options?)` groups items into visual lines by y (±2pt tolerance, configurable), buckets each line's items into 3 columns (name/city/status) by x-position, and stitches multi-line org names back together — a continuation line has text only in the name column. Column boundaries are inferred per page from that page's header row token x-positions (`Organization`/`Name`, `City`, `Status`); pages whose header doesn't repeat (e.g. page 2+) reuse the last-inferred boundaries, or accept an explicit `columnBoundaries: [nameX, cityX, statusX]` override. Throws if a page has data but no boundaries can be determined at all — a layout change should fail loudly, not misparse silently. + - `normalizeRegistryRows(rows)` maps free-text status onto `'good_standing'` (`good standing` / `current` / `active`) | `'lapsed'` (`lapsed` / `delinquent` / `suspended` / `expired` / `revoked`) | `'unknown'` (anything else, including blank), collapses whitespace in names/cities, and drops non-registrant artifact rows (repeated header, page-number footers, `Page X of Y`, NHDOJ letterhead). +- **Workflow** — `src/workflows/ingest-nhdoj-orgs.ts`: scheduled monthly at `0 4 1 * *` (`ingestNhdojOrgs`), same registration pattern as `ingest-grants.ts` (module-scope deps registry via `setIngestNhdojOrgsDeps`, dual workflow+scheduled registration, `globalThis` guard). Steps: fetch PDF bytes from `NHDOJ_REGISTRY_PDF_URL` (3 retries) → extract + parse (pure, not a DBOS step) → `serverUpsertOrgFromRegistry` per row in a plain `for...of` loop (`sourceRegistry: 'nhdoj_charitable_trusts'`) → `console.log` of rows parsed / upserted / newly-inserted counts. + - **Config gap ≠ outage**: if `NHDOJ_REGISTRY_PDF_URL` is unset, the workflow logs a `console.warn` and returns without throwing. NHDOJ has no stable URL for the registry PDF — it changes whenever they republish — so a hard failure here would page on-call for a config gap rather than a real problem. + - **Not yet wired into `apps/outreach-worker/src/main.ts`** — the module registers itself as a side effect of being imported (per the pattern above), but `main.ts` needs an explicit import (for registration-before-launch ordering) plus a `setIngestNhdojOrgsDeps({ db })` call before `DBOS.launch()`: + ```ts + import { setIngestNhdojOrgsDeps } from './workflows/ingest-nhdoj-orgs.js'; + // ... + setIngestNhdojOrgsDeps({ db }); + ``` diff --git a/packages/outreach-core/src/index.ts b/packages/outreach-core/src/index.ts index 90bf15b..9aa80ce 100644 --- a/packages/outreach-core/src/index.ts +++ b/packages/outreach-core/src/index.ts @@ -7,3 +7,4 @@ // server-only action/query. export * from './db/index.js'; export * from './matches/hard-gates.js'; +export * from './orgs/icp-band.js'; diff --git a/packages/outreach-core/src/orgs/actions/enrich-org.server.ts b/packages/outreach-core/src/orgs/actions/enrich-org.server.ts new file mode 100644 index 0000000..625c40b --- /dev/null +++ b/packages/outreach-core/src/orgs/actions/enrich-org.server.ts @@ -0,0 +1,37 @@ +import { eq, sql } from 'drizzle-orm'; + +import type { NpOutreachDatabase, NpOutreachTransaction } from '#~/db/db.js'; +import { schema } from '#~/db/db.js'; +import { icpBandForRevenue } from '../icp-band.js'; + +export interface OrgEnrichmentInput { + readonly ein: string | null; + readonly nteeCode: string | null; + readonly totalRevenue: number | null; + readonly fiscalYearEnd: string | null; +} + +/** + * Applies IRS-derived enrichment (ProPublica Nonprofit Explorer) to an org + * and re-bands it. Orgs that failed to resolve should be recorded too + * (all-null input) so `updatedAt` moves and the enrichment queue doesn't + * retry them every run — the icp band stays/returns `unknown`, which + * down-prioritizes rather than drops them. + */ +export async function serverEnrichOrg( + db: NpOutreachDatabase | NpOutreachTransaction, + orgId: string, + enrichment: OrgEnrichmentInput, +): Promise<void> { + await db + .update(schema.orgs) + .set({ + ein: enrichment.ein, + nteeCode: enrichment.nteeCode, + totalRevenue: enrichment.totalRevenue, + fiscalYearEnd: enrichment.fiscalYearEnd, + icpBand: icpBandForRevenue(enrichment.totalRevenue), + updatedAt: sql`now()`, + }) + .where(eq(schema.orgs.id, orgId)); +} diff --git a/packages/outreach-core/src/orgs/actions/index.server.ts b/packages/outreach-core/src/orgs/actions/index.server.ts index 67c9161..bda5dfb 100644 --- a/packages/outreach-core/src/orgs/actions/index.server.ts +++ b/packages/outreach-core/src/orgs/actions/index.server.ts @@ -1 +1,3 @@ export * from './insert-org.server.js'; +export * from './upsert-org-from-registry.server.js'; +export * from './enrich-org.server.js'; diff --git a/packages/outreach-core/src/orgs/actions/upsert-org-from-registry.server.ts b/packages/outreach-core/src/orgs/actions/upsert-org-from-registry.server.ts new file mode 100644 index 0000000..0386817 --- /dev/null +++ b/packages/outreach-core/src/orgs/actions/upsert-org-from-registry.server.ts @@ -0,0 +1,79 @@ +import { and, eq, sql } from 'drizzle-orm'; + +import type { NpOutreachDatabase, NpOutreachTransaction } from '#~/db/db.js'; +import { schema } from '#~/db/db.js'; + +export interface RegistryOrgInput { + readonly name: string; + readonly city: string | null; + readonly state?: string; + readonly registrationStatus: 'good_standing' | 'lapsed' | 'unknown'; + readonly sourceRegistry: string; +} + +export interface RegistryUpsertResult { + readonly id: string; + readonly inserted: boolean; +} + +/** + * Upserts an org discovered in a charity registry re-scan (e.g. the NHDOJ + * Charitable Trusts list). Registries carry no EIN, so identity is the + * case-insensitive (name, city, state) triple — good enough for a + * single-state registry where the same legal name in the same city is the + * same org. Existing rows get their registration status refreshed (orgs + * lapse and re-register); enrichment fields (EIN, revenue, NTEE) are never + * touched here. + * + * Returns `inserted: true` for brand-new registrants — a segment the + * pipeline treats specially (often actively seeking first-time funding). + */ +export async function serverUpsertOrgFromRegistry( + db: NpOutreachDatabase | NpOutreachTransaction, + org: RegistryOrgInput, +): Promise<RegistryUpsertResult> { + const state = org.state ?? 'NH'; + + const existing = await db + .select({ id: schema.orgs.id }) + .from(schema.orgs) + .where( + and( + sql`lower(${schema.orgs.name}) = lower(${org.name})`, + org.city == null + ? sql`${schema.orgs.city} IS NULL` + : sql`lower(${schema.orgs.city}) = lower(${org.city})`, + eq(schema.orgs.state, state), + ), + ) + .limit(1); + + const found = existing[0]; + if (found != null) { + await db + .update(schema.orgs) + .set({ + registrationStatus: org.registrationStatus, + sourceRegistry: org.sourceRegistry, + updatedAt: sql`now()`, + }) + .where(eq(schema.orgs.id, found.id)); + return { id: found.id, inserted: false }; + } + + const [row] = await db + .insert(schema.orgs) + .values({ + name: org.name, + city: org.city, + state, + registrationStatus: org.registrationStatus, + sourceRegistry: org.sourceRegistry, + }) + .returning({ id: schema.orgs.id }); + + if (row == null) { + throw new Error('serverUpsertOrgFromRegistry: insert returned no row'); + } + return { id: row.id, inserted: true }; +} diff --git a/packages/outreach-core/src/orgs/icp-band.test.ts b/packages/outreach-core/src/orgs/icp-band.test.ts new file mode 100644 index 0000000..be3cdc4 --- /dev/null +++ b/packages/outreach-core/src/orgs/icp-band.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from 'vitest'; + +import { + ICP_PRIMARY_MAX_REVENUE, + ICP_PRIMARY_MIN_REVENUE, + icpBandForRevenue, +} from './icp-band.js'; + +describe('icpBandForRevenue', () => { + it('returns unknown for missing revenue', () => { + expect(icpBandForRevenue(null)).toBe('unknown'); + expect(icpBandForRevenue(undefined)).toBe('unknown'); + expect(icpBandForRevenue(Number.NaN)).toBe('unknown'); + }); + + it('bands below the primary floor as below', () => { + expect(icpBandForRevenue(0)).toBe('below'); + expect(icpBandForRevenue(ICP_PRIMARY_MIN_REVENUE - 1)).toBe('below'); + }); + + it('bands the inclusive primary range as primary', () => { + expect(icpBandForRevenue(ICP_PRIMARY_MIN_REVENUE)).toBe('primary'); + expect(icpBandForRevenue(1_250_000)).toBe('primary'); + expect(icpBandForRevenue(ICP_PRIMARY_MAX_REVENUE)).toBe('primary'); + }); + + it('bands above the primary ceiling as above', () => { + expect(icpBandForRevenue(ICP_PRIMARY_MAX_REVENUE + 1)).toBe('above'); + }); +}); diff --git a/packages/outreach-core/src/orgs/icp-band.ts b/packages/outreach-core/src/orgs/icp-band.ts new file mode 100644 index 0000000..a30d90c --- /dev/null +++ b/packages/outreach-core/src/orgs/icp-band.ts @@ -0,0 +1,21 @@ +/** + * ICP banding from annual revenue (per docs/plan.md, Stage 3): + * $100K–$5M is the primary ICP — below that, orgs rarely have anyone in a + * grant-writing seat; above it, professional development staff need a + * different message. Kept as a pure function so ingestion and re-banding + * jobs share one definition. + */ + +export type IcpBand = 'below' | 'primary' | 'above' | 'unknown'; + +export const ICP_PRIMARY_MIN_REVENUE = 100_000; +export const ICP_PRIMARY_MAX_REVENUE = 5_000_000; + +export function icpBandForRevenue( + totalRevenue: number | null | undefined, +): IcpBand { + if (totalRevenue == null || Number.isNaN(totalRevenue)) return 'unknown'; + if (totalRevenue < ICP_PRIMARY_MIN_REVENUE) return 'below'; + if (totalRevenue <= ICP_PRIMARY_MAX_REVENUE) return 'primary'; + return 'above'; +} diff --git a/packages/outreach-core/src/orgs/queries/index.server.ts b/packages/outreach-core/src/orgs/queries/index.server.ts index 60e5214..efee2e1 100644 --- a/packages/outreach-core/src/orgs/queries/index.server.ts +++ b/packages/outreach-core/src/orgs/queries/index.server.ts @@ -1 +1,2 @@ export * from './list-orgs-in-icp-band.server.js'; +export * from './list-orgs-needing-enrichment.server.js'; diff --git a/packages/outreach-core/src/orgs/queries/list-orgs-needing-enrichment.server.ts b/packages/outreach-core/src/orgs/queries/list-orgs-needing-enrichment.server.ts new file mode 100644 index 0000000..32460ee --- /dev/null +++ b/packages/outreach-core/src/orgs/queries/list-orgs-needing-enrichment.server.ts @@ -0,0 +1,36 @@ +import { and, asc, isNull } from 'drizzle-orm'; + +import type { NpOutreachDatabase, NpOutreachTransaction } from '#~/db/db.js'; +import { schema } from '#~/db/db.js'; + +export interface OrgNeedingEnrichment { + id: string; + name: string; + city: string | null; + state: string; +} + +/** + * Orgs that have never been through IRS enrichment: no EIN resolved and + * no revenue on file. Ordered oldest-touched first so a bounded batch job + * works through the backlog fairly. Orgs already attempted (recorded via + * serverEnrichOrg with all-null input) still match this filter only until + * a design for retry windows lands — callers should bound `limit` to keep + * a nightly run cheap. + */ +export async function serverListOrgsNeedingEnrichment( + db: NpOutreachDatabase | NpOutreachTransaction, + { limit }: { limit: number }, +): Promise<OrgNeedingEnrichment[]> { + return db + .select({ + id: schema.orgs.id, + name: schema.orgs.name, + city: schema.orgs.city, + state: schema.orgs.state, + }) + .from(schema.orgs) + .where(and(isNull(schema.orgs.ein), isNull(schema.orgs.totalRevenue))) + .orderBy(asc(schema.orgs.updatedAt)) + .limit(limit); +} diff --git a/yarn.lock b/yarn.lock index 5075f43..7ae0dce 100644 --- a/yarn.lock +++ b/yarn.lock @@ -1370,6 +1370,129 @@ __metadata: languageName: node linkType: hard +"@napi-rs/canvas-android-arm64@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-android-arm64@npm:0.1.100" + conditions: os=android & cpu=arm64 + languageName: node + linkType: hard + +"@napi-rs/canvas-darwin-arm64@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-darwin-arm64@npm:0.1.100" + conditions: os=darwin & cpu=arm64 + languageName: node + linkType: hard + +"@napi-rs/canvas-darwin-x64@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-darwin-x64@npm:0.1.100" + conditions: os=darwin & cpu=x64 + languageName: node + linkType: hard + +"@napi-rs/canvas-linux-arm-gnueabihf@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-linux-arm-gnueabihf@npm:0.1.100" + conditions: os=linux & cpu=arm + languageName: node + linkType: hard + +"@napi-rs/canvas-linux-arm64-gnu@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-linux-arm64-gnu@npm:0.1.100" + conditions: os=linux & cpu=arm64 & libc=glibc + languageName: node + linkType: hard + +"@napi-rs/canvas-linux-arm64-musl@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-linux-arm64-musl@npm:0.1.100" + conditions: os=linux & cpu=arm64 & libc=musl + languageName: node + linkType: hard + +"@napi-rs/canvas-linux-riscv64-gnu@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-linux-riscv64-gnu@npm:0.1.100" + conditions: os=linux & cpu=riscv64 & libc=glibc + languageName: node + linkType: hard + +"@napi-rs/canvas-linux-x64-gnu@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-linux-x64-gnu@npm:0.1.100" + conditions: os=linux & cpu=x64 & libc=glibc + languageName: node + linkType: hard + +"@napi-rs/canvas-linux-x64-musl@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-linux-x64-musl@npm:0.1.100" + conditions: os=linux & cpu=x64 & libc=musl + languageName: node + linkType: hard + +"@napi-rs/canvas-win32-arm64-msvc@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-win32-arm64-msvc@npm:0.1.100" + conditions: os=win32 & cpu=arm64 + languageName: node + linkType: hard + +"@napi-rs/canvas-win32-x64-msvc@npm:0.1.100": + version: 0.1.100 + resolution: "@napi-rs/canvas-win32-x64-msvc@npm:0.1.100" + conditions: os=win32 & cpu=x64 + languageName: node + linkType: hard + +"@napi-rs/canvas@npm:^0.1.65": + version: 0.1.100 + resolution: "@napi-rs/canvas@npm:0.1.100" + dependencies: + "@napi-rs/canvas-android-arm64": "npm:0.1.100" + "@napi-rs/canvas-darwin-arm64": "npm:0.1.100" + "@napi-rs/canvas-darwin-x64": "npm:0.1.100" + "@napi-rs/canvas-linux-arm-gnueabihf": "npm:0.1.100" + "@napi-rs/canvas-linux-arm64-gnu": "npm:0.1.100" + "@napi-rs/canvas-linux-arm64-musl": "npm:0.1.100" + "@napi-rs/canvas-linux-riscv64-gnu": "npm:0.1.100" + "@napi-rs/canvas-linux-x64-gnu": "npm:0.1.100" + "@napi-rs/canvas-linux-x64-musl": "npm:0.1.100" + "@napi-rs/canvas-win32-arm64-msvc": "npm:0.1.100" + "@napi-rs/canvas-win32-x64-msvc": "npm:0.1.100" + dependenciesMeta: + "@napi-rs/canvas-android-arm64": + optional: true + "@napi-rs/canvas-darwin-arm64": + optional: true + "@napi-rs/canvas-darwin-x64": + optional: true + "@napi-rs/canvas-linux-arm-gnueabihf": + optional: true + "@napi-rs/canvas-linux-arm64-gnu": + optional: true + "@napi-rs/canvas-linux-arm64-musl": + optional: true + "@napi-rs/canvas-linux-riscv64-gnu": + optional: true + "@napi-rs/canvas-linux-x64-gnu": + optional: true + "@napi-rs/canvas-linux-x64-musl": + optional: true + "@napi-rs/canvas-win32-arm64-msvc": + optional: true + "@napi-rs/canvas-win32-x64-msvc": + optional: true + canvas: + built: true + skia-canvas: + built: true + checksum: 10c0/d3dfca5620a41c34addd344bd4c448a5334d3f32f2562ec8390507b3897747ed1de9f08fcb9169cff6dd27111afb5b3eaf7e42cc1494c8fa4128e7d4235f6bea + languageName: node + linkType: hard + "@nodelib/fs.scandir@npm:2.1.5": version: 2.1.5 resolution: "@nodelib/fs.scandir@npm:2.1.5" @@ -1487,9 +1610,12 @@ __metadata: "@types/pg": "npm:8.20.0" drizzle-orm: "npm:0.44.6" esbuild: "npm:^0.24.0" + fast-xml-parser: "npm:^4.5.0" + pdfjs-dist: "npm:^4.10.38" pg: "npm:8.20.0" tsx: "npm:^4.19.2" typescript: "npm:^5.9.3" + vitest: "npm:^3.2.4" languageName: unknown linkType: soft @@ -3580,6 +3706,17 @@ __metadata: languageName: node linkType: hard +"fast-xml-parser@npm:^4.5.0": + version: 4.5.7 + resolution: "fast-xml-parser@npm:4.5.7" + dependencies: + strnum: "npm:^1.0.5" + bin: + fxparser: src/cli/cli.js + checksum: 10c0/5fccf3f53d6b2b83143d73089f04ef5db5343422891193cf10d92d3eb856007b7337818494109e46f6fa47b31b9716be15c4b275ff87a0aeb6f7315aa2edc181 + languageName: node + linkType: hard + "fastq@npm:^1.6.0": version: 1.20.1 resolution: "fastq@npm:1.20.1" @@ -4899,6 +5036,18 @@ __metadata: languageName: node linkType: hard +"pdfjs-dist@npm:^4.10.38": + version: 4.10.38 + resolution: "pdfjs-dist@npm:4.10.38" + dependencies: + "@napi-rs/canvas": "npm:^0.1.65" + dependenciesMeta: + "@napi-rs/canvas": + optional: true + checksum: 10c0/77b022109be7aac00372750a53decea3979409e6ef1cf93bf554351569cd4d1fafc70afae4a9a3e4b4de3facf59d3acd54d324b0fcff781374bcb00493d449ce + languageName: node + linkType: hard + "pg-cloudflare@npm:^1.3.0": version: 1.4.0 resolution: "pg-cloudflare@npm:1.4.0" @@ -5948,6 +6097,13 @@ __metadata: languageName: node linkType: hard +"strnum@npm:^1.0.5": + version: 1.1.2 + resolution: "strnum@npm:1.1.2" + checksum: 10c0/a0fce2498fa3c64ce64a40dada41beb91cabe3caefa910e467dc0518ef2ebd7e4d10f8c2202a6104f1410254cae245066c0e94e2521fb4061a5cb41831952392 + languageName: node + linkType: hard + "sucrase@npm:^3.35.0": version: 3.35.1 resolution: "sucrase@npm:3.35.1"