christian/grant-outreach-engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(scoring): 990-PF funder-precedent index — the 25-point subscore goes live

Browse Source 
New funders/funder_grants schema + ingest990pf monthly workflow: IRS BMF
state file discovers NH private foundations (747), e-file index CSVs
select their latest 990-PF filings, batch ZIPs stream through fflate
(4/run cap, most-hits-first, deferred logged), grants-paid rows land in
funder_grants, and funders with >=2 NH grants synthesize rolling grant
rows (source irs_990pf, funder_ein linked) that flow through the
existing embed+match pipeline.

Scoring v2: funderPrecedentSubscore tiers repeated in-state giving
(1/3/5/10 -> 8/15/20/25); easy win = >=65 total AND >=12 precedent
(plan's precedent floor); scale is the full 0-100. Rolling deadlines
pass the runway gate. Retrieval computes per-funder in-state counts and
exposes funder_ein.

Lead-quality gates from the first precedent run's failures: candidate
orgs exclude NTEE T* grantmakers; self-matches gated by EIN + normalized
name (NHDOJ registers foundations as charities, several without resolved
EINs — the first run's top 'leads' were foundations matched to
themselves).

Live: ~6.5GB of IRS batches processed, 2,766 grants-paid rows, 123
synthesized foundation grants, 89 easy wins across 27 orgs, credible
top-10 (AIDS Response-Seacoast -> Foundation for Seacoast Health, 25/25
precedent). 153 tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Croissant Le Doux
2026-07-16 18:17:28 -04:00
parent 0ee478ec3d
commit 63b58e514d
34 changed files with 3634 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/outreach-worker/src/main.ts
View File

@@ -34,6 +34,7 @@ import { setEmbedGrantsDeps } from './workflows/embed-grants.js';
import { setEnrichOrgsDeps } from './workflows/enrich-orgs.js';
import { setExpireGrantsDeps } from './workflows/expire-grants.js';
import { setIngestGrantsDeps } from './workflows/ingest-grants.js';
import { setIngest990pfDeps } from './workflows/ingest-990pf.js';
import { setIngestNhdojOrgsDeps } from './workflows/ingest-nhdoj-orgs.js';
import { setMatchGrantsDeps } from './workflows/match-grants.js';
import { setIngestPndRssDeps } from './workflows/ingest-pnd-rss.js';
@@ -69,6 +70,7 @@ async function main() {
setEnrichOrgsDeps({ db });
setEmbedGrantsDeps({ db });
setMatchGrantsDeps({ db });
setIngest990pfDeps({ db });
DBOS.setConfig({
name: 'helmdocs-outreach-worker',

7
apps/outreach-worker/src/run-once.ts
View File

@@ -32,6 +32,10 @@ import {
runIngestGrantsNow,
setIngestGrantsDeps,
} from './workflows/ingest-grants.js';
import {
runIngest990PfNow,
setIngest990pfDeps,
} from './workflows/ingest-990pf.js';
import {
runIngestNhdojOrgsNow,
setIngestNhdojOrgsDeps,
@@ -53,6 +57,7 @@ const RUNNERS: Record<string, () => Promise<void>> = {
expireGrants: runExpireGrantsNow,
embedGrants: runEmbedGrantsNow,
matchGrants: runMatchGrantsNow,
ingest990pf: runIngest990PfNow,
};
const FIRST_RUN_ORDER = [
@@ -61,6 +66,7 @@ const FIRST_RUN_ORDER = [
'ingestNhdojOrgs',
'expireGrants',
'enrichOrgs',
'ingest990pf',
'embedGrants',
'matchGrants',
];
@@ -99,6 +105,7 @@ async function main() {
setEnrichOrgsDeps({ db });
setEmbedGrantsDeps({ db });
setMatchGrantsDeps({ db });
setIngest990pfDeps({ db });
DBOS.setConfig({
name: 'helmdocs-outreach-worker',

135
apps/outreach-worker/src/sources/irs-990pf/bmf.test.ts Normal file
View File

@@ -0,0 +1,135 @@
import { describe, expect, it } from 'vitest';
import { parseBmfFoundations, parseCsvLine } from './bmf.js';
const HEADER =
'EIN,NAME,ICO,STREET,CITY,STATE,ZIP,GROUP,SUBSECTION,AFFILIATION,CLASSIFICATION,RULING,DEDUCTIBILITY,FOUNDATION,ACTIVITY,ORGANIZATION,STATUS,TAX_PERIOD,ASSET_CD,INCOME_CD,FILING_REQ_CD,PF_FILING_REQ_CD,ACCT_PD,ASSET_AMT,INCOME_AMT,REVENUE_AMT,NTEE_CD,SORT_NAME';
function row(fields: Partial<Record<string, string>>): string {
const cols = HEADER.split(',');
return cols.map((c) => fields[c] ?? '').join(',');
}
describe('parseCsvLine', () => {
it('splits an unquoted line on commas', () => {
expect(parseCsvLine('a,b,c')).toEqual(['a', 'b', 'c']);
});
it('handles a quoted field containing a comma', () => {
expect(parseCsvLine('a,"b, still b",c')).toEqual(['a', 'b, still b', 'c']);
});
it('unescapes doubled quotes inside a quoted field', () => {
expect(parseCsvLine('a,"he said ""hi""",c')).toEqual([
'a',
'he said "hi"',
'c',
]);
});
it('handles an empty trailing field', () => {
expect(parseCsvLine('a,b,')).toEqual(['a', 'b', '']);
});
});
describe('parseBmfFoundations', () => {
it('keeps only rows with PF_FILING_REQ_CD === "1"', () => {
const csv = [
HEADER,
row({
EIN: '020123456',
NAME: 'Granite State Foundation',
CITY: 'Manchester',
STATE: 'NH',
NTEE_CD: 'T20',
PF_FILING_REQ_CD: '1',
ASSET_AMT: '5000000',
}),
row({
EIN: '020654321',
NAME: 'Not A Private Foundation Inc',
CITY: 'Concord',
STATE: 'NH',
PF_FILING_REQ_CD: '0',
ASSET_AMT: '1000000',
}),
].join('\n');
const foundations = parseBmfFoundations(csv);
expect(foundations).toHaveLength(1);
expect(foundations[0]?.name).toBe('Granite State Foundation');
});
it('zero-pads a short EIN to 9 digits', () => {
const csv = [
HEADER,
row({
EIN: '123456', // 6 digits, as BMF sometimes drops leading zeros
NAME: 'Small EIN Foundation',
PF_FILING_REQ_CD: '1',
}),
].join('\n');
const foundations = parseBmfFoundations(csv);
expect(foundations[0]?.ein).toBe('000123456');
});
it('parses a quoted NAME field containing a comma', () => {
const csv = [
HEADER,
row({
EIN: '020123456',
NAME: '"Smith, Jones & Family Foundation"',
CITY: 'Nashua',
STATE: 'NH',
PF_FILING_REQ_CD: '1',
ASSET_AMT: '2500000',
NTEE_CD: 'T30',
}),
].join('\n');
const foundations = parseBmfFoundations(csv);
expect(foundations[0]?.name).toBe('Smith, Jones & Family Foundation');
expect(foundations[0]?.city).toBe('Nashua');
expect(foundations[0]?.totalAssets).toBe(2_500_000);
});
it('maps blank optional fields to null', () => {
const csv = [
HEADER,
row({
EIN: '020999999',
NAME: 'No Extras Foundation',
PF_FILING_REQ_CD: '1',
}),
].join('\n');
const foundations = parseBmfFoundations(csv);
expect(foundations[0]).toMatchObject({
city: null,
state: null,
nteeCode: null,
totalAssets: null,
});
});
it('skips rows with a blank EIN or blank NAME', () => {
const csv = [
HEADER,
row({ EIN: '', NAME: 'No EIN Foundation', PF_FILING_REQ_CD: '1' }),
row({ EIN: '020111111', NAME: '', PF_FILING_REQ_CD: '1' }),
].join('\n');
expect(parseBmfFoundations(csv)).toEqual([]);
});
it('returns an empty array for a header-only CSV', () => {
expect(parseBmfFoundations(HEADER)).toEqual([]);
});
it('throws when required columns are missing from the header', () => {
expect(() => parseBmfFoundations('FOO,BAR\n1,2')).toThrow(
/missing required column/i,
);
});
});

144
apps/outreach-worker/src/sources/irs-990pf/bmf.ts Normal file
View File

@@ -0,0 +1,144 @@
/**
* IRS Business Master File (BMF) state extract — the discovery layer for
* the 990-PF funder-precedent index. The BMF lists every EO-recognized
* organization in a state, one row per EIN, including whether it's a
* private foundation required to file a 990-PF (`PF_FILING_REQ_CD === '1'`).
*
* `fetchBmfCsv` is the only impure piece; `parseBmfFoundations` is a pure
* function of CSV text so it's unit-testable on fixture strings.
*/
export interface BmfFoundation {
/** Zero-padded to 9 digits, e.g. "020123456". */
readonly ein: string;
readonly name: string;
readonly city: string | null;
readonly state: string | null;
readonly nteeCode: string | null;
readonly totalAssets: number | null;
}
/**
* Tolerant CSV line splitter: handles double-quoted fields (with embedded
* commas and escaped `""`) defensively, even though the IRS BMF export's
* fields are unquoted in practice.
*/
export function parseCsvLine(line: string): string[] {
const fields: string[] = [];
let current = '';
let inQuotes = false;
for (let i = 0; i < line.length; i++) {
const char = line[i];
if (inQuotes) {
if (char === '"') {
if (line[i + 1] === '"') {
current += '"';
i++;
} else {
inQuotes = false;
}
} else {
current += char;
}
} else if (char === '"') {
inQuotes = true;
} else if (char === ',') {
fields.push(current);
current = '';
} else {
current += char;
}
}
fields.push(current);
return fields;
}
const REQUIRED_COLUMNS = ['EIN', 'NAME', 'PF_FILING_REQ_CD'] as const;
/**
* Parses a BMF state CSV into the private-foundation subset —
* `PF_FILING_REQ_CD === '1'` — that this vertical cares about. Column
* lookup is by header name (not position), since the BMF export carries
* many more columns than we use.
*
* Malformed/short rows (fewer fields than the header, blank EIN or name)
* are skipped rather than thrown on — a single-row anomaly in a 1.6MB
* government CSV shouldn't fail the whole discovery pass.
*/
export function parseBmfFoundations(csv: string): BmfFoundation[] {
const lines = csv.split(/\r?\n/).filter((line) => line.trim().length > 0);
if (lines.length === 0) return [];
const header = parseCsvLine(lines[0]!).map((h) => h.trim().toUpperCase());
const einIdx = header.indexOf('EIN');
const nameIdx = header.indexOf('NAME');
const cityIdx = header.indexOf('CITY');
const stateIdx = header.indexOf('STATE');
const nteeIdx = header.indexOf('NTEE_CD');
const pfReqIdx = header.indexOf('PF_FILING_REQ_CD');
const assetIdx = header.indexOf('ASSET_AMT');
const missing = REQUIRED_COLUMNS.filter((col) => header.indexOf(col) === -1);
if (missing.length > 0) {
throw new Error(
`parseBmfFoundations: BMF CSV header is missing required column(s): ${missing.join(', ')}`,
);
}
const out: BmfFoundation[] = [];
for (let i = 1; i < lines.length; i++) {
const fields = parseCsvLine(lines[i]!);
const pfCode = fields[pfReqIdx]?.trim();
if (pfCode !== '1') continue; // not a 990-PF filer
const rawEin = fields[einIdx]?.trim() ?? '';
const digits = rawEin.replace(/\D/g, '');
if (digits === '') continue;
const ein = digits.padStart(9, '0');
const name = fields[nameIdx]?.trim() ?? '';
if (name === '') continue;
const city = cityIdx === -1 ? null : fields[cityIdx]?.trim() || null;
const state = stateIdx === -1 ? null : fields[stateIdx]?.trim() || null;
const nteeCode = nteeIdx === -1 ? null : fields[nteeIdx]?.trim() || null;
let totalAssets: number | null = null;
if (assetIdx !== -1) {
const rawAsset = fields[assetIdx]?.trim();
if (rawAsset != null && rawAsset !== '') {
const parsed = Number(rawAsset);
totalAssets = Number.isFinite(parsed) ? Math.round(parsed) : null;
}
}
out.push({ ein, name, city, state, nteeCode, totalAssets });
}
return out;
}
/**
* Fetches a state's IRS BMF extract. irs.gov sits behind Akamai bot
* detection — bare/default clients get a 403; Node's fetch with
* browser-like headers passes (same approach as
* `ingest-nhdoj-orgs.ts`'s `fetchNhdojRegistryPdf`, verified 2026-07-16).
*/
export async function fetchBmfCsv(state = 'nh'): Promise<string> {
const url = `https://www.irs.gov/pub/irs-soi/eo_${state}.csv`;
const res = await fetch(url, {
headers: {
'User-Agent':
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
Accept: 'text/csv,*/*',
},
});
if (!res.ok) {
throw new Error(
`fetchBmfCsv: ${url} responded ${res.status} ${res.statusText}`,
);
}
return res.text();
}

141
apps/outreach-worker/src/sources/irs-990pf/index-csv.test.ts Normal file
View File

@@ -0,0 +1,141 @@
import { describe, expect, it } from 'vitest';
import { selectFilings } from './index-csv.js';
const HEADER =
'RETURN_ID,FILING_TYPE,EIN,TAX_PERIOD,SUB_DATE,TAXPAYER_NAME,RETURN_TYPE,DLN,OBJECT_ID,XML_BATCH_ID';
function row(fields: {
returnId?: string;
filingType?: string;
ein: string;
taxPeriod: string;
subDate?: string;
taxpayerName?: string;
returnType: string;
dln?: string;
objectId: string;
batchId: string;
}): string {
return [
fields.returnId ?? '1',
fields.filingType ?? 'EFILE',
fields.ein,
fields.taxPeriod,
fields.subDate ?? '20270301',
fields.taxpayerName ?? 'SOME FOUNDATION',
fields.returnType,
fields.dln ?? '93000000000000',
fields.objectId,
fields.batchId,
].join(',');
}
describe('selectFilings', () => {
it('keeps only RETURN_TYPE === "990PF" rows', () => {
const csv = [
HEADER,
row({
ein: '020123456',
taxPeriod: '202612',
returnType: '990PF',
objectId: '111',
batchId: 'BATCH1',
}),
row({
ein: '020654321',
taxPeriod: '202612',
returnType: '990',
objectId: '222',
batchId: 'BATCH1',
}),
].join('\n');
const targetEins = new Set(['020123456', '020654321']);
const filings = selectFilings(csv, targetEins);
expect(filings).toHaveLength(1);
expect(filings[0]?.ein).toBe('020123456');
});
it('filters to the target EIN set', () => {
const csv = [
HEADER,
row({
ein: '020999999',
taxPeriod: '202612',
returnType: '990PF',
objectId: '333',
batchId: 'BATCH1',
}),
].join('\n');
expect(selectFilings(csv, new Set(['020123456']))).toEqual([]);
});
it('zero-pads a short EIN before matching against the target set', () => {
const csv = [
HEADER,
row({
ein: '20123456', // 8 digits
taxPeriod: '202612',
returnType: '990PF',
objectId: '444',
batchId: 'BATCH1',
}),
].join('\n');
const filings = selectFilings(csv, new Set(['020123456']));
expect(filings).toHaveLength(1);
expect(filings[0]?.ein).toBe('020123456');
});
it('keeps only the LATEST TAX_PERIOD per EIN within the file', () => {
const csv = [
HEADER,
row({
ein: '020123456',
taxPeriod: '202512',
returnType: '990PF',
objectId: 'OLD',
batchId: 'BATCH1',
}),
row({
ein: '020123456',
taxPeriod: '202612',
returnType: '990PF',
objectId: 'NEW',
batchId: 'BATCH2',
}),
].join('\n');
const filings = selectFilings(csv, new Set(['020123456']));
expect(filings).toHaveLength(1);
expect(filings[0]?.objectId).toBe('NEW');
expect(filings[0]?.batchId).toBe('BATCH2');
});
it('skips rows with a blank OBJECT_ID or XML_BATCH_ID', () => {
const csv = [
HEADER,
row({
ein: '020123456',
taxPeriod: '202612',
returnType: '990PF',
objectId: '',
batchId: 'BATCH1',
}),
].join('\n');
expect(selectFilings(csv, new Set(['020123456']))).toEqual([]);
});
it('returns an empty array for a header-only file', () => {
expect(selectFilings(HEADER, new Set(['020123456']))).toEqual([]);
});
it('throws when required columns are missing from the header', () => {
expect(() => selectFilings('FOO,BAR\n1,2', new Set(['020123456']))).toThrow(
/missing required column/i,
);
});
});

143
apps/outreach-worker/src/sources/irs-990pf/index-csv.ts Normal file
View File

@@ -0,0 +1,143 @@
/**
* IRS e-file index CSV — maps EINs to the batch ZIP + XML object id
* containing their 990-PF filing for a given year.
* (`https://apps.irs.gov/pub/epostcard/990/xml/{year}/index_{year}.csv`).
*
* These files run ~28MB, so `selectFilings` walks the text line-by-line
* (via a manual index scan, not `String#split` — no ~300K-element line
* array materialized) and only ever retains the handful of rows that match
* our target EIN set, never a parsed copy of the whole file.
*/
import { parseCsvLine } from '#~/sources/irs-990pf/bmf.js';
export interface SelectedFiling {
/** Zero-padded to 9 digits. */
readonly ein: string;
readonly objectId: string;
readonly batchId: string;
/** Wire value from `TAX_PERIOD`, typically `YYYYMM`. */
readonly taxPeriod: string;
}
const REQUIRED_COLUMNS = [
'EIN',
'TAX_PERIOD',
'RETURN_TYPE',
'OBJECT_ID',
'XML_BATCH_ID',
] as const;
/** Yields each line of `text` without ever allocating a full line array. */
function* iterateLines(text: string): Generator<string> {
const len = text.length;
let start = 0;
while (start < len) {
let end = text.indexOf('\n', start);
if (end === -1) end = len;
let line = text.slice(start, end);
if (line.endsWith('\r')) line = line.slice(0, -1);
if (line.length > 0) yield line;
start = end + 1;
}
}
/**
* Selects `RETURN_TYPE === '990PF'` rows whose EIN is in `targetEins`,
* keeping only the LATEST `TAX_PERIOD` per EIN found in this file (an EIN
* can appear more than once in a year's index across amended/superseded
* filings — the largest `TAX_PERIOD` wins).
*/
export function selectFilings(
csvText: string,
targetEins: ReadonlySet<string>,
): SelectedFiling[] {
const latestByEin = new Map<
string,
SelectedFiling & { taxPeriodNum: number }
>();
let header: string[] | null = null;
let einIdx = -1;
let taxPeriodIdx = -1;
let returnTypeIdx = -1;
let objectIdIdx = -1;
let batchIdIdx = -1;
for (const line of iterateLines(csvText)) {
if (header == null) {
header = parseCsvLine(line).map((h) => h.trim().toUpperCase());
einIdx = header.indexOf('EIN');
taxPeriodIdx = header.indexOf('TAX_PERIOD');
returnTypeIdx = header.indexOf('RETURN_TYPE');
objectIdIdx = header.indexOf('OBJECT_ID');
batchIdIdx = header.indexOf('XML_BATCH_ID');
const missing = REQUIRED_COLUMNS.filter(
(col) => header!.indexOf(col) === -1,
);
if (missing.length > 0) {
throw new Error(
`selectFilings: e-file index CSV header is missing required column(s): ${missing.join(', ')}`,
);
}
continue;
}
const fields = parseCsvLine(line);
const returnType = fields[returnTypeIdx]?.trim();
if (returnType !== '990PF') continue;
const rawEin = fields[einIdx]?.trim() ?? '';
const digits = rawEin.replace(/\D/g, '');
if (digits === '') continue;
const ein = digits.padStart(9, '0');
if (!targetEins.has(ein)) continue;
const objectId = fields[objectIdIdx]?.trim() ?? '';
const batchId = fields[batchIdIdx]?.trim() ?? '';
if (objectId === '' || batchId === '') continue;
const taxPeriod = fields[taxPeriodIdx]?.trim() ?? '';
const taxPeriodNum = Number(taxPeriod);
const existing = latestByEin.get(ein);
if (
existing == null ||
(Number.isFinite(taxPeriodNum) && taxPeriodNum > existing.taxPeriodNum)
) {
latestByEin.set(ein, {
ein,
objectId,
batchId,
taxPeriod,
taxPeriodNum: Number.isFinite(taxPeriodNum) ? taxPeriodNum : -1,
});
}
}
return [...latestByEin.values()].map(
({ taxPeriodNum: _taxPeriodNum, ...rest }) => rest,
);
}
/**
* Fetches one year's e-file index CSV. Same Akamai-friendly headers as
* `fetchBmfCsv`/`fetchNhdojRegistryPdf` — apps.irs.gov is also
* Akamai-fronted (verified 2026-07-16).
*/
export async function fetchIndexCsv(year: number): Promise<string> {
const url = `https://apps.irs.gov/pub/epostcard/990/xml/${year}/index_${year}.csv`;
const res = await fetch(url, {
headers: {
'User-Agent':
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
Accept: 'text/csv,*/*',
},
});
if (!res.ok) {
throw new Error(
`fetchIndexCsv: ${url} responded ${res.status} ${res.statusText}`,
);
}
return res.text();
}

153
apps/outreach-worker/src/sources/irs-990pf/parse-990pf-xml.test.ts Normal file
View File

@@ -0,0 +1,153 @@
import { describe, expect, it } from 'vitest';
import { parse990PfXml } from './parse-990pf-xml.js';
// Realistic (trimmed) 990-PF e-file XML shape, namespace-prefixed the way
// the real IRS MeF documents are (`irs:` on IRS990PF-specific elements) —
// `removeNSPrefix: true` should strip these transparently.
const MULTI_GRANT_FIXTURE = `<?xml version="1.0" encoding="UTF-8"?>
<Return xmlns="http://www.irs.gov/efile">
<ReturnHeader>
<TaxYr>2026</TaxYr>
<TaxPeriodEndDt>2026-12-31</TaxPeriodEndDt>
</ReturnHeader>
<ReturnData>
<irs:IRS990PF xmlns:irs="http://www.irs.gov/efile">
<irs:SupplementaryInformationGrp>
<irs:ApplicationSubmissionInfoGrp>
<irs:RecipientNm>See attached list</irs:RecipientNm>
<irs:FormAndInfoAndMaterialsTxt>Submit a two-page letter of inquiry.</irs:FormAndInfoAndMaterialsTxt>
<irs:SubmissionDeadlinesTxt>March 1 and September 1 annually.</irs:SubmissionDeadlinesTxt>
<irs:RestrictionsOnAwardsTxt>Grants limited to 501(c)(3) organizations in New Hampshire.</irs:RestrictionsOnAwardsTxt>
</irs:ApplicationSubmissionInfoGrp>
<irs:GrantOrContributionPdDurYrGrp>
<irs:RecipientBusinessName>
<irs:BusinessNameLine1Txt>Granite State Youth Services</irs:BusinessNameLine1Txt>
</irs:RecipientBusinessName>
<irs:RecipientUSAddress>
<irs:CityNm>Manchester</irs:CityNm>
<irs:StateAbbreviationCd>NH</irs:StateAbbreviationCd>
</irs:RecipientUSAddress>
<irs:Amt>25000</irs:Amt>
<irs:GrantOrContributionPurposeTxt>General operating support</irs:GrantOrContributionPurposeTxt>
</irs:GrantOrContributionPdDurYrGrp>
<irs:GrantOrContributionPdDurYrGrp>
<irs:RecipientPersonNm>Jane Q. Public</irs:RecipientPersonNm>
<irs:RecipientUSAddress>
<irs:CityNm>Concord</irs:CityNm>
<irs:StateAbbreviationCd>NH</irs:StateAbbreviationCd>
</irs:RecipientUSAddress>
<irs:Amt>5000</irs:Amt>
<irs:GrantOrContributionPurposeTxt>Scholarship</irs:GrantOrContributionPurposeTxt>
</irs:GrantOrContributionPdDurYrGrp>
</irs:SupplementaryInformationGrp>
</irs:IRS990PF>
</ReturnData>
</Return>`;
// A filer with exactly one grant paid — fast-xml-parser collapses a
// single-element repeated group to a bare object without `isArray`.
const SINGLE_GRANT_FIXTURE = `<?xml version="1.0" encoding="UTF-8"?>
<Return xmlns="http://www.irs.gov/efile">
<ReturnHeader>
<TaxYr>2025</TaxYr>
</ReturnHeader>
<ReturnData>
<irs:IRS990PF xmlns:irs="http://www.irs.gov/efile">
<irs:SupplementaryInformationGrp>
<irs:GrantOrContributionPdDurYrGrp>
<irs:RecipientBusinessName>
<irs:BusinessNameLine1Txt>Lone Grantee Org</irs:BusinessNameLine1Txt>
</irs:RecipientBusinessName>
<irs:Amt>1200</irs:Amt>
</irs:GrantOrContributionPdDurYrGrp>
</irs:SupplementaryInformationGrp>
</irs:IRS990PF>
</ReturnData>
</Return>`;
const NO_GRANTS_FIXTURE = `<?xml version="1.0" encoding="UTF-8"?>
<Return xmlns="http://www.irs.gov/efile">
<ReturnHeader>
<TaxPeriodEndDt>2024-06-30</TaxPeriodEndDt>
</ReturnHeader>
<ReturnData>
<irs:IRS990PF xmlns:irs="http://www.irs.gov/efile">
<irs:SupplementaryInformationGrp/>
</irs:IRS990PF>
</ReturnData>
</Return>`;
describe('parse990PfXml', () => {
it('extracts grants paid, including a business-name recipient', () => {
const parsed = parse990PfXml(MULTI_GRANT_FIXTURE);
expect(parsed.grantsPaid).toHaveLength(2);
expect(parsed.grantsPaid[0]).toEqual({
recipientName: 'Granite State Youth Services',
recipientCity: 'Manchester',
recipientState: 'NH',
amount: 25000,
purpose: 'General operating support',
});
});
it('extracts a person-name recipient via RecipientPersonNm', () => {
const parsed = parse990PfXml(MULTI_GRANT_FIXTURE);
expect(parsed.grantsPaid[1]).toEqual({
recipientName: 'Jane Q. Public',
recipientCity: 'Concord',
recipientState: 'NH',
amount: 5000,
purpose: 'Scholarship',
});
});
it('coerces a single grant group (fast-xml-parser collapses one-element arrays)', () => {
const parsed = parse990PfXml(SINGLE_GRANT_FIXTURE);
expect(parsed.grantsPaid).toHaveLength(1);
expect(parsed.grantsPaid[0]).toMatchObject({
recipientName: 'Lone Grantee Org',
amount: 1200,
recipientCity: null,
recipientState: null,
purpose: null,
});
});
it('extracts application submission info into a plain object', () => {
const parsed = parse990PfXml(MULTI_GRANT_FIXTURE);
expect(parsed.applicationInfo).toEqual({
recipientName: 'See attached list',
formAndInfoAndMaterials: 'Submit a two-page letter of inquiry.',
submissionDeadlines: 'March 1 and September 1 annually.',
restrictionsOnAwards:
'Grants limited to 501(c)(3) organizations in New Hampshire.',
});
});
it('prefers ReturnHeader/TaxYr for the tax year', () => {
expect(parse990PfXml(MULTI_GRANT_FIXTURE).taxYear).toBe(2026);
expect(parse990PfXml(SINGLE_GRANT_FIXTURE).taxYear).toBe(2025);
});
it('falls back to the year in TaxPeriodEndDt when TaxYr is absent', () => {
expect(parse990PfXml(NO_GRANTS_FIXTURE).taxYear).toBe(2024);
});
it('returns an empty grants array and null applicationInfo when the group is empty', () => {
const parsed = parse990PfXml(NO_GRANTS_FIXTURE);
expect(parsed.grantsPaid).toEqual([]);
expect(parsed.applicationInfo).toBeNull();
});
it('is defensive against a totally missing IRS990PF/ReturnHeader shape', () => {
const parsed = parse990PfXml(
'<Return xmlns="http://www.irs.gov/efile"><ReturnData/></Return>',
);
expect(parsed).toEqual({
taxYear: null,
applicationInfo: null,
grantsPaid: [],
});
});
});

156
apps/outreach-worker/src/sources/irs-990pf/parse-990pf-xml.ts Normal file
View File

@@ -0,0 +1,156 @@
/**
* Pure parse of a 990-PF e-file XML document (`Return/ReturnData/IRS990PF`)
* into the fields the funder-precedent index needs: the filing's tax year,
* Part XV application-submission info, and Part XV line 3a grants paid.
*
* Namespace prefixes on these documents vary by filing software vendor, so
* the parser is configured with `removeNSPrefix: true` and every group is
* read defensively — a missing group resolves to an empty array/`null`
* rather than throwing, and single-element repeated groups (which
* `fast-xml-parser` collapses to a bare object) are coerced back to arrays.
*/
import { XMLParser } from 'fast-xml-parser';
export interface GrantPaid {
readonly recipientName: string;
readonly recipientCity: string | null;
readonly recipientState: string | null;
readonly amount: number | null;
readonly purpose: string | null;
}
export interface Parsed990Pf {
readonly taxYear: number | null;
readonly applicationInfo: Record<string, unknown> | null;
readonly grantsPaid: GrantPaid[];
}
const parser = new XMLParser({
ignoreAttributes: true,
removeNSPrefix: true,
trimValues: true,
isArray: (_name, jpath) =>
jpath.endsWith('GrantOrContributionPdDurYrGrp') ||
jpath.endsWith('ApplicationSubmissionInfoGrp'),
});
function coerceArray<T>(value: T | T[] | null | undefined): T[] {
if (value == null) return [];
return Array.isArray(value) ? value : [value];
}
function textOf(value: unknown): string | null {
if (value == null) return null;
const s = String(value).trim();
return s === '' ? null : s;
}
function amountOf(value: unknown): number | null {
if (value == null) return null;
const n = typeof value === 'number' ? value : Number(String(value).trim());
return Number.isFinite(n) ? Math.round(n) : null;
}
/**
* `ReturnHeader/TaxYr` when present; otherwise derived from the
* `TaxPeriodEndDt` (`YYYY-MM-DD`) date's calendar year.
*/
function extractTaxYear(header: Record<string, unknown> | null): number | null {
if (header == null) return null;
const taxYr = header.TaxYr;
if (taxYr != null) {
const n = Number(String(taxYr).trim());
if (Number.isFinite(n)) return n;
}
const endDt = header.TaxPeriodEndDt;
if (typeof endDt === 'string') {
const match = /^(\d{4})-\d{2}-\d{2}/.exec(endDt.trim());
if (match != null) return Number(match[1]);
}
return null;
}
function extractGrantsPaid(pf: Record<string, unknown> | null): GrantPaid[] {
const suppInfo = pf?.SupplementaryInformationGrp as
Record<string, unknown> | undefined;
const rows = coerceArray<Record<string, unknown>>(
suppInfo?.GrantOrContributionPdDurYrGrp as
Record<string, unknown>[] | Record<string, unknown> | undefined,
);
const grants: GrantPaid[] = [];
for (const row of rows) {
const businessName = row.RecipientBusinessName as
Record<string, unknown> | undefined;
const recipientName =
textOf(businessName?.BusinessNameLine1Txt) ??
textOf(row.RecipientPersonNm);
if (recipientName == null) continue; // no usable recipient identity
const address = row.RecipientUSAddress as
Record<string, unknown> | undefined;
grants.push({
recipientName,
recipientCity: textOf(address?.CityNm),
recipientState: textOf(address?.StateAbbreviationCd),
amount: amountOf(row.Amt),
purpose: textOf(row.GrantOrContributionPurposeTxt),
});
}
return grants;
}
/**
* Captures whatever Part XV "application submission info" text fields are
* present (form/deadline/address/restriction narrative) into a plain
* object. Filers vary widely in how much of this they fill in, so every
* field is optional and the result is `null` if none are present.
*/
function extractApplicationInfo(
pf: Record<string, unknown> | null,
): Record<string, unknown> | null {
const suppInfo = pf?.SupplementaryInformationGrp as
Record<string, unknown> | undefined;
const raw =
(suppInfo?.ApplicationSubmissionInfoGrp as
Record<string, unknown>[] | Record<string, unknown> | undefined) ??
(pf?.ApplicationSubmissionInfoGrp as
Record<string, unknown>[] | Record<string, unknown> | undefined);
const groups = coerceArray<Record<string, unknown>>(raw);
const first = groups[0];
if (first == null) return null;
const result: Record<string, unknown> = {};
const recipientName = textOf(first.RecipientNm);
if (recipientName != null) result.recipientName = recipientName;
const formAndInfo = textOf(first.FormAndInfoAndMaterialsTxt);
if (formAndInfo != null) result.formAndInfoAndMaterials = formAndInfo;
const deadlines = textOf(first.SubmissionDeadlinesTxt);
if (deadlines != null) result.submissionDeadlines = deadlines;
const restrictions = textOf(first.RestrictionsOnAwardsTxt);
if (restrictions != null) result.restrictionsOnAwards = restrictions;
return Object.keys(result).length === 0 ? null : result;
}
export function parse990PfXml(xml: string): Parsed990Pf {
const parsed = parser.parse(xml) as {
Return?: {
ReturnHeader?: Record<string, unknown>;
ReturnData?: { IRS990PF?: Record<string, unknown> };
};
};
const header = parsed.Return?.ReturnHeader ?? null;
const pf = parsed.Return?.ReturnData?.IRS990PF ?? null;
return {
taxYear: extractTaxYear(header),
applicationInfo: extractApplicationInfo(pf),
grantsPaid: extractGrantsPaid(pf),
};
}

51
apps/outreach-worker/src/sources/irs-990pf/zip.test.ts Normal file
View File

@@ -0,0 +1,51 @@
import { strToU8, zipSync } from 'fflate';
import { describe, expect, it } from 'vitest';
import { extractEntries } from './zip.js';
function buildFixtureZip(): Uint8Array {
return zipSync({
'111_public.xml': strToU8('<Return>wanted-one</Return>'),
'222_public.xml': strToU8('<Return>wanted-two</Return>'),
'333_public.xml': strToU8('<Return>not-wanted</Return>'),
'index.json': strToU8('{}'),
});
}
describe('extractEntries', () => {
it('extracts only the wanted entries, keyed by name', () => {
const zip = buildFixtureZip();
const wanted = new Set(['111_public.xml', '222_public.xml']);
const entries = extractEntries(zip, wanted);
expect(entries.size).toBe(2);
expect(new TextDecoder().decode(entries.get('111_public.xml'))).toBe(
'<Return>wanted-one</Return>',
);
expect(new TextDecoder().decode(entries.get('222_public.xml'))).toBe(
'<Return>wanted-two</Return>',
);
});
it('does not include entries outside the wanted set', () => {
const zip = buildFixtureZip();
const entries = extractEntries(zip, new Set(['111_public.xml']));
expect(entries.has('333_public.xml')).toBe(false);
expect(entries.has('index.json')).toBe(false);
expect(entries.size).toBe(1);
});
it('returns an empty map when none of the wanted names are present', () => {
const zip = buildFixtureZip();
const entries = extractEntries(zip, new Set(['999_public.xml']));
expect(entries.size).toBe(0);
});
it('returns an empty map for an empty wanted set', () => {
const zip = buildFixtureZip();
const entries = extractEntries(zip, new Set());
expect(entries.size).toBe(0);
});
});

28
apps/outreach-worker/src/sources/irs-990pf/zip.ts Normal file
View File

@@ -0,0 +1,28 @@
/**
* Thin wrapper over `fflate` for selective extraction from a 990-PF batch
* ZIP. Batches run 100-400MB and contain thousands of `{objectId}_public.xml`
* entries; we only ever want the handful matching this run's target object
* ids, so `unzipSync`'s `filter` is used to skip inflating everything else —
* memory stays bounded by the (already-in-memory) zip buffer plus the small
* set of extracted XMLs, not the full uncompressed archive.
*/
import { unzipSync } from 'fflate';
/**
* Extracts only the entries in `wantedNames` from `zipBuffer`, keyed by
* their in-archive filename.
*/
export function extractEntries(
zipBuffer: Uint8Array,
wantedNames: ReadonlySet<string>,
): Map<string, Uint8Array> {
const files = unzipSync(zipBuffer, {
filter: (file) => wantedNames.has(file.name),
});
const result = new Map<string, Uint8Array>();
for (const [name, data] of Object.entries(files)) {
result.set(name, data);
}
return result;
}

693
apps/outreach-worker/src/workflows/ingest-990pf.ts Normal file
View File

@@ -0,0 +1,693 @@
/**
* Monthly IRS 990-PF funder-precedent ingestion workflow.
*
* Private foundations rarely publish an open RFP — Grants.gov/PND/NHDOJ
* never see them. Instead this workflow builds a "precedent index": it
* discovers NH-registered private foundations from the IRS Business Master
* File, cross-references the IRS e-file index to find each one's latest
* 990-PF XML filing, parses out grants actually paid to NH nonprofits, and
* synthesizes a `grants` row per foundation with enough NH giving history
* ("this foundation funds orgs like you") — the only signal available for
* funders with no public application process.
*
* Flow:
* 1. BMF discovery — fetch the NH BMF extract, filter to 990-PF filers,
* upsert every one as a `funders` row (refreshes name/city/state/
* NTEE/assets in place; never clobbers filing-derived fields — see
* `serverUpsertFunder`'s COALESCE logic).
* 2. E-file index — fetch the current + previous year's index CSV(s),
* select the latest 990PF filing per target EIN.
* 3. Skip filter — drop (EIN, filing) pairs whose selected object id
* matches the funder's already-recorded `latestObjectId` (nothing
* changed since the last run).
* 4. Batch processing — group remaining filings by batch ZIP, process the
* batches with the most hits first (capped per run), each batch: fetch
* the ZIP to a temp file, extract only the wanted `{objectId}_public
* .xml` entries, parse each, upsert the funder's filing-derived fields
* and replace its grants-paid rows for that tax year, then delete the
* temp file.
* 5. Synthesis — for funders with enough NH giving history, build and
* upsert a `grants` row summarizing that history as a lead.
*
* 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. Values
* that DO cross a `DBOS.registerStep` boundary are kept JSON-serializable
* (plain objects/arrays, never `Map`/`Set`) for the same reason.
*/
import { createWriteStream } from 'node:fs';
import { mkdtemp, readFile, rm, stat } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { DBOS, SchedulerMode } from '@dbos-inc/dbos-sdk';
import type { schema } from '@novelpad/outreach-core';
import {
serverInsertGrants,
serverListFunderLatestObjectIds,
serverListFunderSynthesisData,
serverReplaceFunderGrantsForYear,
serverUpsertFunder,
type FunderSynthesisRow,
type NewGrantInput,
} from '@novelpad/outreach-core/server';
import type { NodePgDatabase } from 'drizzle-orm/node-postgres';
import {
fetchBmfCsv,
parseBmfFoundations,
type BmfFoundation,
} from '#~/sources/irs-990pf/bmf.js';
import {
fetchIndexCsv,
selectFilings,
type SelectedFiling,
} from '#~/sources/irs-990pf/index-csv.js';
import { parse990PfXml } from '#~/sources/irs-990pf/parse-990pf-xml.js';
import { extractEntries } from '#~/sources/irs-990pf/zip.js';
export type OutreachDb = NodePgDatabase<typeof schema>;
/**
* A `SelectedFiling` tagged with the e-file index year it was found under
* — the directory the filing's batch ZIP lives in
* (`.../xml/{indexYear}/{batchId}.zip`), which need not equal the filing's
* `taxPeriod` year (see the comment where this is assigned below).
*/
interface IndexedFiling extends SelectedFiling {
readonly indexYear: number;
}
/** The one state this vertical (and the rest of the outreach engine) targets. */
const BMF_STATE = 'nh';
const RECIPIENT_STATE = 'NH';
/** Minimum in-state grants-paid rows before a foundation is worth synthesizing a lead from. */
const MIN_STATE_GRANTS = 2;
/** Default batch-ZIP cap per run; overridable for backfills/supervised runs. */
const DEFAULT_MAX_BATCHES_PER_RUN = 4;
const BROWSER_HEADERS = {
'User-Agent':
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
} as const;
export interface Ingest990PfDeps {
readonly db: OutreachDb;
}
let registeredDeps: Ingest990PfDeps | null = null;
export function setIngest990pfDeps(deps: Ingest990PfDeps): void {
registeredDeps = deps;
}
function getIngest990pfDeps(): Ingest990PfDeps {
if (registeredDeps == null) {
throw new Error(
'Ingest990PfDeps not registered. Call setIngest990pfDeps() before DBOS.launch().',
);
}
return registeredDeps;
}
/** `IRS_990PF_YEARS` env override (comma-sep), else `[currentYear, currentYear - 1]`. */
function resolveYears(): number[] {
const override = process.env.IRS_990PF_YEARS;
if (override != null && override.trim() !== '') {
const years = override
.split(',')
.map((y) => Number(y.trim()))
.filter((y) => Number.isFinite(y));
if (years.length > 0) return years;
}
const currentYear = new Date().getUTCFullYear();
return [currentYear, currentYear - 1];
}
function resolveMaxBatches(): number {
const override = process.env.IRS_990PF_MAX_BATCHES_PER_RUN;
if (override != null && override.trim() !== '') {
const n = Number(override.trim());
if (Number.isFinite(n) && n > 0) return Math.floor(n);
}
return DEFAULT_MAX_BATCHES_PER_RUN;
}
/** Derives a tax year from an XML parse (preferred) or a `YYYYMM` `TAX_PERIOD` fallback. */
function resolveTaxYear(
parsedTaxYear: number | null,
taxPeriod: string,
): number | null {
if (parsedTaxYear != null) return parsedTaxYear;
const match = /^(\d{4})\d{2}$/.exec(taxPeriod);
return match != null ? Number(match[1]) : null;
}
// ---------------------------------------------------------------------------
// Step 1: BMF discovery
// ---------------------------------------------------------------------------
async function fetchBmf(state: string): Promise<string> {
const csv = await fetchBmfCsv(state);
console.log(
`[ingest-990pf] fetched BMF CSV for state=${state} (${csv.length} bytes)`,
);
return csv;
}
const fetchBmfStep = DBOS.registerStep(fetchBmf, {
name: 'fetchBmf990Pf',
retriesAllowed: true,
maxAttempts: 3,
});
async function upsertFoundation(
db: OutreachDb,
foundation: BmfFoundation,
): Promise<string> {
return serverUpsertFunder(db, {
ein: foundation.ein,
name: foundation.name,
city: foundation.city,
state: foundation.state,
nteeCode: foundation.nteeCode,
totalAssets: foundation.totalAssets,
});
}
const upsertFoundationStep = DBOS.registerStep(upsertFoundation, {
name: 'upsertBmfFoundation',
retriesAllowed: true,
maxAttempts: 3,
});
// ---------------------------------------------------------------------------
// Step 2: e-file index
// ---------------------------------------------------------------------------
async function fetchIndex(year: number): Promise<string> {
const csv = await fetchIndexCsv(year);
console.log(
`[ingest-990pf] fetched e-file index for year=${year} (${csv.length} bytes)`,
);
return csv;
}
const fetchIndexStep = DBOS.registerStep(fetchIndex, {
name: 'fetchIrs990PfIndex',
retriesAllowed: true,
maxAttempts: 3,
});
// ---------------------------------------------------------------------------
// Step 3: skip filter — funders' already-recorded latestObjectId
// ---------------------------------------------------------------------------
async function lookupLatestObjectIds(
db: OutreachDb,
eins: string[],
): Promise<Array<{ ein: string; latestObjectId: string | null }>> {
const rows = await serverListFunderLatestObjectIds(db, eins);
return rows.map((r) => ({ ein: r.ein, latestObjectId: r.latestObjectId }));
}
const lookupLatestObjectIdsStep = DBOS.registerStep(lookupLatestObjectIds, {
name: 'lookupFunderLatestObjectIds',
retriesAllowed: true,
maxAttempts: 3,
});
// ---------------------------------------------------------------------------
// Step 4: batch processing
// ---------------------------------------------------------------------------
/**
* Streams a batch ZIP straight to a temp file (never materializing the full
* 100-400MB response in a JS `ArrayBuffer`), logs its size, and returns the
* temp dir + file path for the caller to read/extract from and clean up.
*/
async function downloadBatchZip(
year: number,
batchId: string,
): Promise<{ dir: string; zipPath: string }> {
const url = `https://apps.irs.gov/pub/epostcard/990/xml/${year}/${batchId}.zip`;
const res = await fetch(url, { headers: BROWSER_HEADERS });
if (!res.ok) {
throw new Error(
`downloadBatchZip: ${url} responded ${res.status} ${res.statusText}`,
);
}
if (res.body == null) {
throw new Error(`downloadBatchZip: ${url} returned no response body`);
}
const dir = await mkdtemp(path.join(tmpdir(), 'irs990pf-'));
const zipPath = path.join(dir, `${batchId}.zip`);
await pipeline(
Readable.fromWeb(res.body as never),
createWriteStream(zipPath),
);
const { size } = await stat(zipPath);
console.log(
`[ingest-990pf] downloaded batch ${batchId} (year ${year}): ${(size / (1024 * 1024)).toFixed(1)} MB -> ${zipPath}`,
);
return { dir, zipPath };
}
interface BatchProcessResult {
readonly xmlsParsed: number;
readonly grantsPaidRows: number;
readonly xmlsMissing: number;
}
/**
* Processes one batch ZIP durably: download, extract only the wanted
* `{objectId}_public.xml` entries, parse + persist each filing, then delete
* the temp file (`finally`, so a mid-loop failure still cleans up).
*/
async function processBatch(
db: OutreachDb,
year: number,
batchId: string,
filings: SelectedFiling[],
foundationByEin: Record<string, BmfFoundation>,
): Promise<BatchProcessResult> {
const { dir, zipPath } = await downloadBatchZip(year, batchId);
let xmlsParsed = 0;
let grantsPaidRows = 0;
let xmlsMissing = 0;
try {
const zipBuffer = new Uint8Array(await readFile(zipPath));
const wantedNames = new Set(filings.map((f) => `${f.objectId}_public.xml`));
const entries = extractEntries(zipBuffer, wantedNames);
console.log(
`[ingest-990pf] batch ${batchId}: wanted ${wantedNames.size} filing(s), extracted ${entries.size}`,
);
const decoder = new TextDecoder('utf-8');
for (const filing of filings) {
const entryName = `${filing.objectId}_public.xml`;
const bytes = entries.get(entryName);
if (bytes == null) {
xmlsMissing++;
console.warn(
`[ingest-990pf] batch ${batchId}: entry ${entryName} (EIN ${filing.ein}) not found in ZIP; skipping`,
);
continue;
}
const foundation = foundationByEin[filing.ein];
if (foundation == null) {
console.warn(
`[ingest-990pf] batch ${batchId}: no BMF foundation record for EIN ${filing.ein}; skipping`,
);
continue;
}
try {
const xml = decoder.decode(bytes);
const parsed = parse990PfXml(xml);
const taxYear = resolveTaxYear(parsed.taxYear, filing.taxPeriod);
if (taxYear == null) {
console.warn(
`[ingest-990pf] batch ${batchId}: could not resolve a tax year for EIN ${filing.ein} (object ${filing.objectId}); skipping`,
);
continue;
}
const funderId = await serverUpsertFunder(db, {
ein: filing.ein,
name: foundation.name,
city: foundation.city,
state: foundation.state,
nteeCode: foundation.nteeCode,
totalAssets: foundation.totalAssets,
applicationInfo: parsed.applicationInfo,
latestTaxYear: taxYear,
latestObjectId: filing.objectId,
});
await serverReplaceFunderGrantsForYear(
db,
funderId,
taxYear,
parsed.grantsPaid,
);
xmlsParsed++;
grantsPaidRows += parsed.grantsPaid.length;
} catch (err) {
console.error(
`[ingest-990pf] batch ${batchId}: failed to parse/persist EIN ${filing.ein} (object ${filing.objectId}):`,
err,
);
}
}
} finally {
await rm(dir, { recursive: true, force: true }).catch((err) => {
console.warn(`[ingest-990pf] failed to clean up temp dir ${dir}:`, err);
});
}
return { xmlsParsed, grantsPaidRows, xmlsMissing };
}
// Two attempts, not three like the lighter-weight steps elsewhere in this
// worker — a batch download is 100-400MB, so retrying a real outage 3x is
// expensive; a transient blip gets one retry and otherwise waits for next
// month's run (batches are re-selected fresh every run, nothing is lost).
const processBatchStep = DBOS.registerStep(processBatch, {
name: 'processIrs990PfBatch',
retriesAllowed: true,
maxAttempts: 2,
});
// ---------------------------------------------------------------------------
// Step 5: synthesis
// ---------------------------------------------------------------------------
function locationClause(city: string | null, state: string | null): string {
if (city != null && state != null) return ` based in ${city}, ${state}`;
if (state != null) return ` based in ${state}`;
if (city != null) return ` based in ${city}`;
return '';
}
function applicationInfoSentence(info: unknown): string | null {
if (info == null || typeof info !== 'object') return null;
const rec = info as Record<string, unknown>;
const parts: string[] = [];
if (
typeof rec.formAndInfoAndMaterials === 'string' &&
rec.formAndInfoAndMaterials.trim() !== ''
) {
parts.push(rec.formAndInfoAndMaterials.trim());
}
if (
typeof rec.submissionDeadlines === 'string' &&
rec.submissionDeadlines.trim() !== ''
) {
parts.push(`Deadlines: ${rec.submissionDeadlines.trim()}`);
}
if (
typeof rec.restrictionsOnAwards === 'string' &&
rec.restrictionsOnAwards.trim() !== ''
) {
parts.push(`Restrictions: ${rec.restrictionsOnAwards.trim()}`);
}
return parts.length === 0
? null
: `Application info from its most recent filing: ${parts.join(' ')}`;
}
/** Pure: builds a synthesized precedent-lead `grants` row from one funder's aggregated NH giving history. */
export function buildSynthesizedGrant(
row: FunderSynthesisRow,
now: Date,
): NewGrantInput {
const medianText =
row.medianAmount != null
? ` typically around $${row.medianAmount.toLocaleString('en-US')}`
: '';
const purposesText = row.purposes.slice(0, 8).join('; ');
const sentences: string[] = [
`Private foundation${locationClause(row.city, row.state)}.`,
`Has made ${row.stateGrantCount} grant${row.stateGrantCount === 1 ? '' : 's'} to New Hampshire organizations in recent filings${medianText}.`,
];
if (purposesText.length > 0) {
sentences.push(`Recent grant purposes include: ${purposesText}.`);
}
const appInfo = applicationInfoSentence(row.applicationInfo);
if (appInfo != null) sentences.push(appInfo);
const awardCeiling = row.maxAmount ?? row.medianAmount ?? null;
const awardFloor =
row.medianAmount != null && row.maxAmount != null
? Math.min(row.medianAmount, row.maxAmount)
: (row.medianAmount ?? row.maxAmount ?? null);
return {
sourceUrl: `https://projects.propublica.org/nonprofits/organizations/${row.ein}`,
source: 'irs_990pf',
funderEin: row.ein,
funder: row.name,
title: `${row.name} — grants for New Hampshire nonprofits`,
synopsis: sentences.join(' '),
eligibilityEntityTypes: null,
// `serverListFunderSynthesisData` aggregates grants paid INTO
// recipientState only — we have no visibility into this funder's
// TOTAL giving footprint from this query, so we can't tell whether NH
// is an exclusive restriction or just where it happens to have given
// before. Asserting 'New Hampshire' here would fake a hard geography
// gate off data that doesn't support it; leaving it null keeps the NH
// signal in the precedent subscore instead. (Would need a
// total-grant-count column added to the synthesis query to assert
// this safely — out of scope: core changes weren't part of this task.)
geographicScope: null,
programAreas: null,
awardFloor,
awardCeiling,
expectedAwardsCount: null,
openDate: null,
closeDate: null,
matchRequirement: false,
applicationEffortEstimate: 'unknown',
applicationFormSupported: false,
status: 'open',
lastVerifiedAt: now,
};
}
async function synthesizeFunderGrants(db: OutreachDb): Promise<number> {
const rows = await serverListFunderSynthesisData(db, {
recipientState: RECIPIENT_STATE,
minStateGrants: MIN_STATE_GRANTS,
});
if (rows.length === 0) return 0;
const now = new Date();
const grants = rows.map((row) => buildSynthesizedGrant(row, now));
await serverInsertGrants(db, grants);
return grants.length;
}
const synthesizeFunderGrantsStep = DBOS.registerStep(synthesizeFunderGrants, {
name: 'synthesizeFunderGrants',
retriesAllowed: true,
maxAttempts: 3,
});
// ---------------------------------------------------------------------------
// Workflow
// ---------------------------------------------------------------------------
async function runIngest990Pf(): Promise<void> {
const { db } = getIngest990pfDeps();
// 1. BMF discovery.
const bmfCsv = await fetchBmfStep(BMF_STATE);
const foundations = parseBmfFoundations(bmfCsv);
console.log(
`[ingest-990pf] BMF foundations (990-PF filers): ${foundations.length}`,
);
const funderIdByEin = new Map<string, string>();
const foundationByEin: Record<string, BmfFoundation> = {};
for (const foundation of foundations) {
const funderId = await upsertFoundationStep(db, foundation);
funderIdByEin.set(foundation.ein, funderId);
foundationByEin[foundation.ein] = foundation;
}
console.log(`[ingest-990pf] foundations upserted: ${funderIdByEin.size}`);
if (funderIdByEin.size === 0) {
console.log(
'[ingest-990pf] no 990-PF filers found in BMF this run; skipping filings + synthesis',
);
return;
}
// 2. e-file index — current + previous year(s), latest filing per EIN.
// `indexYear` is the directory the filing's index CSV (and therefore its
// batch ZIP) was fetched from — NOT necessarily the filing's tax period
// year (a filing e-filed/processed in 2027 can carry TAX_PERIOD 202612)
// — so it's tracked alongside each filing rather than re-derived later.
const targetEins = new Set(funderIdByEin.keys());
const years = resolveYears();
const selectedByEin = new Map<string, IndexedFiling>();
for (const year of years) {
let indexCsv: string;
try {
indexCsv = await fetchIndexStep(year);
} catch (err) {
console.warn(
`[ingest-990pf] e-file index fetch failed for year ${year}; skipping that year:`,
err,
);
continue;
}
const filings = selectFilings(indexCsv, targetEins);
console.log(
`[ingest-990pf] year ${year}: ${filings.length} matching 990PF filing(s)`,
);
for (const filing of filings) {
const existing = selectedByEin.get(filing.ein);
if (
existing == null ||
Number(filing.taxPeriod) > Number(existing.taxPeriod)
) {
selectedByEin.set(filing.ein, { ...filing, indexYear: year });
}
}
}
console.log(
`[ingest-990pf] filings selected across ${years.length} year(s): ${selectedByEin.size}`,
);
if (selectedByEin.size > 0) {
// 3. Skip filter.
const latestObjectIdRows = await lookupLatestObjectIdsStep(db, [
...selectedByEin.keys(),
]);
const latestObjectIdByEin = new Map(
latestObjectIdRows.map((r) => [r.ein, r.latestObjectId]),
);
const toProcess: IndexedFiling[] = [];
let skippedUnchanged = 0;
for (const filing of selectedByEin.values()) {
const known = latestObjectIdByEin.get(filing.ein);
if (known != null && known === filing.objectId) {
skippedUnchanged++;
continue;
}
toProcess.push(filing);
}
if (skippedUnchanged > 0) {
console.log(
`[ingest-990pf] skipping ${skippedUnchanged} filing(s) already parsed (unchanged latestObjectId)`,
);
}
// 4. Group by batch (scoped to the index year the batch ZIP actually
// lives under), most-hits-first, capped.
const batchesByYear = new Map<
string,
{ year: number; batchId: string; filings: SelectedFiling[] }
>();
for (const filing of toProcess) {
const key = `${filing.indexYear}::${filing.batchId}`;
const entry = batchesByYear.get(key) ?? {
year: filing.indexYear,
batchId: filing.batchId,
filings: [],
};
entry.filings.push(filing);
batchesByYear.set(key, entry);
}
const sortedBatches = [...batchesByYear.values()].sort(
(a, b) => b.filings.length - a.filings.length,
);
const maxBatches = resolveMaxBatches();
const batchesToProcess = sortedBatches.slice(0, maxBatches);
const deferredBatches = sortedBatches.slice(maxBatches);
if (deferredBatches.length > 0) {
const deferredFilings = deferredBatches.reduce(
(n, b) => n + b.filings.length,
0,
);
console.warn(
`[ingest-990pf] batch cap ${maxBatches} reached: deferring ${deferredBatches.length} batch(es) (${deferredFilings} filing(s)) to a future run — ${deferredBatches.map((b) => `${b.batchId}(${b.filings.length})`).join(', ')}`,
);
}
let xmlsParsedTotal = 0;
let grantsPaidTotal = 0;
let xmlsMissingTotal = 0;
for (const batch of batchesToProcess) {
try {
const result = await processBatchStep(
db,
batch.year,
batch.batchId,
batch.filings,
foundationByEin,
);
xmlsParsedTotal += result.xmlsParsed;
grantsPaidTotal += result.grantsPaidRows;
xmlsMissingTotal += result.xmlsMissing;
} catch (err) {
console.error(`[ingest-990pf] batch ${batch.batchId} failed:`, err);
}
}
console.log(
`[ingest-990pf] batches processed=${batchesToProcess.length} deferred=${deferredBatches.length} xmlsParsed=${xmlsParsedTotal} xmlsMissing=${xmlsMissingTotal} grantsPaidRows=${grantsPaidTotal}`,
);
} else {
console.log(
'[ingest-990pf] no filings selected this run; skipping batch processing',
);
}
// 5. Synthesis.
const synthesized = await synthesizeFunderGrantsStep(db);
console.log(
`[ingest-990pf] synthesized funder-precedent grants upserted: ${synthesized}`,
);
}
const g = globalThis as unknown as {
__outreachIngest990PfRegistered?: boolean;
__outreachIngest990PfHandle?: (
scheduledTime: Date,
startedAt: Date,
) => Promise<void>;
};
if (!g.__outreachIngest990PfRegistered) {
g.__outreachIngest990PfRegistered = true;
const ingest990pf = async (_scheduledTime: Date, _startedAt: Date) => {
try {
await runIngest990Pf();
} catch (err) {
console.error('[ingest-990pf] 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.
g.__outreachIngest990PfHandle = DBOS.registerWorkflow(ingest990pf, {
name: 'ingest990pf',
});
DBOS.registerScheduled(ingest990pf, {
crontab: '0 6 2 * *',
name: 'ingest990pf',
mode: SchedulerMode.ExactlyOncePerInterval,
});
}
/**
* Starts one durable run of this workflow immediately through DBOS —
* the exact production path (workflow + checkpointed steps), used by
* `run-once.ts` for supervised/manual passes. Requires deps injected and
* `DBOS.launch()` completed.
*/
export function runIngest990PfNow(): Promise<void> {
const handle = g.__outreachIngest990PfHandle;
if (handle == null) {
throw new Error(
'ingest990pf is not registered; was this module imported before DBOS.launch()?',
);
}
return handle(new Date(), new Date());
}

31
apps/outreach-worker/src/workflows/match-grants.ts
View File

@@ -108,8 +108,10 @@ const ensureOrgEmbeddingStep = DBOS.registerStep(ensureOrgEmbedding, {
async function retrieveGrants(
db: OutreachDb,
embedding: number[],
orgState: string,
): Promise<EligibleGrantWithSimilarity[]> {
return serverListEligibleGrantsForOrg(db, embedding, {
orgState,
minDaysToDeadline: MIN_DAYS_TO_DEADLINE,
minAwardCeiling: MIN_AWARD_CEILING,
limit: GRANTS_PER_ORG,
@@ -121,6 +123,14 @@ const retrieveGrantsStep = DBOS.registerStep(retrieveGrants, {
maxAttempts: 3,
});
/** Case/punctuation/suffix-insensitive equality for self-match detection. */
function normalizeSelfMatchName(name: string): string {
return name
.toUpperCase()
.replace(/\b(INC|INCORPORATED|TTEE|TRUSTEE|FUND|FOUNDATION|CHARITABLE|TRUST)\b/g, '')
.replace(/[^A-Z0-9]/g, '');
}
async function scoreAndStoreOrgMatches(
db: OutreachDb,
org: MatchCandidateOrg,
@@ -131,6 +141,22 @@ async function scoreAndStoreOrgMatches(
let gated = 0;
for (const grant of grants) {
// A foundation's synthesized grant must never match the foundation's
// own org row ("we found you $50K — from yourself"). EIN when both
// sides have one; normalized-name fallback because many NHDOJ org rows
// haven't resolved an EIN yet — a self-match is worse than a missed
// match here.
const isSelfByEin =
grant.funderEin != null &&
org.ein != null &&
grant.funderEin === org.ein;
const isSelfByName =
grant.funderEin != null &&
normalizeSelfMatchName(grant.funder) === normalizeSelfMatchName(org.name);
if (isSelfByEin || isSelfByName) {
gated++;
continue;
}
const gates = evaluateHardGates(
{ entityType: ASSUMED_ENTITY_TYPE, state: org.state },
{
@@ -153,6 +179,7 @@ async function scoreAndStoreOrgMatches(
const scored = scoreMatch({
similarity: grant.similarity,
funderStateGrantCount: grant.funderStateGrantCount,
orgTotalRevenue: org.totalRevenue,
awardCeiling: grant.awardCeiling,
geographicScope: grant.geographicScope,
@@ -173,7 +200,7 @@ async function scoreAndStoreOrgMatches(
gateFailures: gates.failures,
ignoredGates: [...IGNORED_GATES],
scoredAt: now.toISOString(),
scoringVersion: 'v1-no-precedent',
scoringVersion: 'v2-state-precedent',
},
});
stored++;
@@ -203,7 +230,7 @@ async function runMatchGrants(): Promise<void> {
for (const org of orgs) {
try {
const embedding = await ensureOrgEmbeddingStep(db, org);
const grants = await retrieveGrantsStep(db, embedding);
const grants = await retrieveGrantsStep(db, embedding, org.state);
const { stored, gated } = await scoreAndStoreOrgMatchesStep(
db,
org,