fix(ingestion): survive first contact with real data sources

NHDOJ: parser rebuilt for the real 8-column registry layout (Reg. No. |
Charity Name | Address | City | State | Zip | Status | Report Due) with
single-letter G/X/S statuses; Reg. No. is the stable upsert key (new
orgs.registration_number column + partial unique index, enum gains
'suspended' via idempotent ADD VALUE); out-of-state registrants keep
their real state. Akamai-safe fetch headers + NHDOJ_REGISTRY_PDF_PATH
local-file override.

ProPublica: zero-hit state-scoped searches return 404, not an empty
list — map to no-candidates instead of failure (tripped the systemic-
failure breaker at 60/200 on first contact). Enrichment queue now
prioritizes NH good-standing orgs over the out-of-state tail.

PND: feed retired upstream (HTML shell on every historical path) —
documented as rework candidate, low priority.

run-once.ts: supervised one-off runner through the durable DBOS
handles (workflow modules now export run*Now accessors); drop the
double pool.end() after DBOS.shutdown().

First supervised run: 200 Grants.gov opportunities (1 auto-expired),
13,632 orgs from the 427-page registry, enrichment at failed=0 with
121/200 EIN resolution in the NH-priority batch.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

packages/outreach-core/src/db/schema.ts

@@ -43,6 +43,7 @@ export const grantStatusEnum = pgEnum('grant_status', [
 export const registrationStatusEnum = pgEnum('registration_status', [
   'good_standing',
   'lapsed',
+  'suspended',
   'unknown',
 ]);
 
@@ -174,6 +175,9 @@ export const orgs = pgTable(
     nteeCode: text('ntee_code'),
     totalRevenue: integer('total_revenue'),
     fiscalYearEnd: text('fiscal_year_end'),
+    // Registry-assigned identifier (e.g. NHDOJ "Reg. No.") — stable across
+    // republishes, unlike org names; the preferred upsert key when present.
+    registrationNumber: text('registration_number'),
     registrationStatus: registrationStatusEnum('registration_status')
       .notNull()
       .default('unknown'),
@@ -186,6 +190,11 @@ export const orgs = pgTable(
     // Partial unique index: EIN is unique when present, but many small
     // orgs in early ingestion won't have one resolved yet.
     uniqueIndex('idx_orgs_ein').on(t.ein).where(sql`${t.ein} IS NOT NULL`),
+    uniqueIndex('idx_orgs_registry_reg_no')
+      .on(t.sourceRegistry, t.registrationNumber)
+      .where(
+        sql`${t.sourceRegistry} IS NOT NULL AND ${t.registrationNumber} IS NOT NULL`,
+      ),
     index('idx_orgs_icp_band').on(t.icpBand),
     index('idx_orgs_state').on(t.state),
   ],

packages/outreach-core/src/orgs/actions/upsert-org-from-registry.server.ts

@@ -6,8 +6,10 @@ 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 state?: string | null;
+  /** Registry-assigned id (e.g. NHDOJ Reg. No.) — preferred identity when present. */
+  readonly registrationNumber?: string | null;
+  readonly registrationStatus: 'good_standing' | 'lapsed' | 'suspended' | 'unknown';
   readonly sourceRegistry: string;
 }
 
@@ -18,12 +20,17 @@ export interface RegistryUpsertResult {
 
 /**
  * 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.
+ * Charitable Trusts list).
+ *
+ * Identity, in preference order:
+ *  1. (sourceRegistry, registrationNumber) — the registry's own stable id;
+ *     survives renames and relocations. Name/city/state are refreshed from
+ *     the incoming row on this path.
+ *  2. Case-insensitive (name, city, state) — fallback for registries (or
+ *     historical rows) without a registration number. Only status and
+ *     provenance are refreshed on this path.
+ *
+ * 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).
@@ -33,8 +40,37 @@ export async function serverUpsertOrgFromRegistry(
   org: RegistryOrgInput,
 ): Promise<RegistryUpsertResult> {
   const state = org.state ?? 'NH';
+  const registrationNumber = org.registrationNumber ?? null;
 
-  const existing = await db
+  if (registrationNumber != null) {
+    const byRegNo = await db
+      .select({ id: schema.orgs.id })
+      .from(schema.orgs)
+      .where(
+        and(
+          eq(schema.orgs.sourceRegistry, org.sourceRegistry),
+          eq(schema.orgs.registrationNumber, registrationNumber),
+        ),
+      )
+      .limit(1);
+
+    const found = byRegNo[0];
+    if (found != null) {
+      await db
+        .update(schema.orgs)
+        .set({
+          name: org.name,
+          city: org.city,
+          state,
+          registrationStatus: org.registrationStatus,
+          updatedAt: sql`now()`,
+        })
+        .where(eq(schema.orgs.id, found.id));
+      return { id: found.id, inserted: false };
+    }
+  }
+
+  const byName = await db
     .select({ id: schema.orgs.id })
     .from(schema.orgs)
     .where(
@@ -48,13 +84,15 @@ export async function serverUpsertOrgFromRegistry(
     )
     .limit(1);
 
-  const found = existing[0];
+  const found = byName[0];
   if (found != null) {
     await db
       .update(schema.orgs)
       .set({
         registrationStatus: org.registrationStatus,
         sourceRegistry: org.sourceRegistry,
+        // Adopt the registry id so future re-scans match on the stable key.
+        registrationNumber,
         updatedAt: sql`now()`,
       })
       .where(eq(schema.orgs.id, found.id));
@@ -67,6 +105,7 @@ export async function serverUpsertOrgFromRegistry(
       name: org.name,
       city: org.city,
       state,
+      registrationNumber,
       registrationStatus: org.registrationStatus,
       sourceRegistry: org.sourceRegistry,
     })

packages/outreach-core/src/orgs/queries/list-orgs-needing-enrichment.server.ts

@@ -1,4 +1,4 @@
-import { and, asc, isNull } from 'drizzle-orm';
+import { and, asc, isNull, sql } from 'drizzle-orm';
 
 import type { NpOutreachDatabase, NpOutreachTransaction } from '#~/db/db.js';
 import { schema } from '#~/db/db.js';
@@ -31,6 +31,12 @@ export async function serverListOrgsNeedingEnrichment(
     })
     .from(schema.orgs)
     .where(and(isNull(schema.orgs.ein), isNull(schema.orgs.totalRevenue)))
-    .orderBy(asc(schema.orgs.updatedAt))
+    // NH good-standing orgs are the ICP — enrich them before the long tail
+    // of out-of-state registrants; oldest-touched first within each tier.
+    .orderBy(
+      sql`(${schema.orgs.state} = 'NH') DESC`,
+      sql`(${schema.orgs.registrationStatus} = 'good_standing') DESC`,
+      asc(schema.orgs.updatedAt),
+    )
     .limit(limit);
 }