Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -984,6 +984,12 @@ Org-less lookups store `cost_name`; a treg phone reveal is `reveal_route = treg`
(callback + reconciler cancel the 8-credit hold, declare no Apollo credit). treg
meters a phone reveal at 1 credit (its documented under-metering of the +8).
2026-10-09: credits ran out 3 days before the Oct 12 cycle.
**The staff alert fires only when the client lost data** (MON-APOLLO-CREDITS-1111,
25 alerts in 30 days, every call served by treg): inside a platform
`withTregFallback`, chokepoint alerts are HELD (`withCreditAlertsDeferred`,
AsyncLocalStorage in `credit-alert.ts`), dropped when treg serves, sent
(`flushCreditAlerts`, reason + treg's error) when treg fails. Recheck window +
treg failure alerts on the caller's `alertIdentity`. BYOK alerts at once.

## Running out of Apollo credits raises a STAFF EMAIL — never let it stay silent

Expand Down
187 changes: 187 additions & 0 deletions openapi.json
Original file line number Diff line number Diff line change
Expand Up @@ -5506,6 +5506,103 @@
"quickenrich"
]
},
"CreateAudienceFromFiltersResponse": {
"type": "object",
"properties": {
"apolloAudienceId": {
"type": "string",
"format": "uuid"
},
"name": {
"type": "string"
},
"filters": {
"allOf": [
{
"$ref": "#/components/schemas/SearchFilters"
},
{
"description": "Apollo People API Search filters using Apollo-native field names. All filters are combined using AND. Start broad and narrow down to avoid empty results."
}
]
},
"count": {
"type": "integer",
"description": "Live Apollo match-count for the filters, from a free dry-run."
},
"serveSource": {
"type": "string",
"enum": [
"apollo",
"quickenrich"
]
},
"quickenrich": {
"type": "object",
"properties": {
"expressible": {
"type": "boolean"
},
"reasons": {
"type": "array",
"items": {
"type": "string"
}
}
},
"required": [
"expressible",
"reasons"
]
}
},
"required": [
"apolloAudienceId",
"name",
"filters",
"count",
"serveSource",
"quickenrich"
]
},
"CreateAudienceFromFiltersRequest": {
"type": "object",
"properties": {
"name": {
"type": "string",
"minLength": 1,
"maxLength": 200,
"description": "Audience name (English).",
"example": "US Chiropractors (by title)"
},
"description": {
"type": "string",
"minLength": 1,
"maxLength": 2000,
"description": "What the audience is, in one or two sentences (English)."
},
"brandId": {
"type": "string",
"description": "Brand the audience belongs to."
},
"filters": {
"allOf": [
{
"$ref": "#/components/schemas/SearchFilters"
},
{
"description": "Apollo-native People Search filters, stored VERBATIM. A consumer that forwards the same object to /search/next is served from this audience (it is the audience's identity on that route, e.g. for its QuickEnrich serve source). A buying signal goes through POST /audiences/signal."
}
]
}
},
"required": [
"name",
"description",
"filters"
],
"additionalProperties": false
},
"ServeSourceResponse": {
"type": "object",
"properties": {
Expand Down Expand Up @@ -8513,6 +8610,96 @@
}
}
},
"/audiences": {
"post": {
"summary": "Persist an Apollo audience from given Apollo-native filters",
"description": "Stores the filters exactly as given (no LLM, no rewrite), counted by a free Apollo dry-run. The returned apolloAudienceId can then be switched to the QuickEnrich serve source (PATCH /audiences/{apolloAudienceId}/serve-source) and bound to a consumer audience whose stored filters are the same object.",
"parameters": [
{
"schema": {
"type": "string"
},
"required": false,
"name": "x-api-key",
"in": "header"
},
{
"schema": {
"type": "string"
},
"required": true,
"name": "x-org-id",
"in": "header"
},
{
"schema": {
"type": "string"
},
"required": true,
"name": "x-user-id",
"in": "header"
},
{
"schema": {
"type": "string"
},
"required": false,
"name": "x-brand-id",
"in": "header"
},
{
"schema": {
"type": "string"
},
"required": false,
"name": "x-run-id",
"in": "header"
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/CreateAudienceFromFiltersRequest"
}
}
}
},
"responses": {
"201": {
"description": "Persisted",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/CreateAudienceFromFiltersResponse"
}
}
}
},
"400": {
"description": "Validation error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"500": {
"description": "Internal server error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
}
}
}
},
"/audiences/{apolloAudienceId}/serve-source": {
"patch": {
"summary": "Switch an audience between the Apollo and the QuickEnrich serve path",
Expand Down
48 changes: 43 additions & 5 deletions src/lib/apollo-reveal-route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,13 @@

import { decryptPlatformKey } from "./keys-client.js";
import { ApolloCreditsExhaustedError } from "./provider-error.js";
import {
flushCreditAlerts,
reportApolloCreditsExhausted,
withCreditAlertsDeferred,
type CreditAlertIdentity,
type CreditAlertScope,
} from "./credit-alert.js";
import type { ApolloPerson } from "./apollo-client.js";

export const TREG_CALL_BASE = "https://treg.to/call/";
Expand Down Expand Up @@ -135,6 +142,11 @@ export interface FallbackDeps<T> {
credentials?: () => Promise<TregCredentials>;
fetchImpl?: typeof fetch;
now?: () => number;
/**
* Who to alert when treg fails while our key is known to be out (the recheck
* window skips our key, so no Apollo answer carries an identity to alert on).
*/
alertIdentity?: CreditAlertIdentity;
}

/**
Expand All @@ -143,18 +155,44 @@ export interface FallbackDeps<T> {
*/
export async function withTregFallback<T>(keySource: "org" | "platform", deps: FallbackDeps<T>): Promise<RevealResult<T>> {
const now = deps.now ?? Date.now;
// Platform key only: an exhaustion alert raised by our own call is HELD until
// we know whether treg covered it (MON-APOLLO-CREDITS-1111). BYOK has no
// fallback, so its alerts go out at once as before.
const scope: CreditAlertScope = { deferred: [] };
if (keySource !== "platform" || !ownKeyExhausted(now())) {
try {
return { response: await deps.own(), via: "apollo", tregCostMicro: null, tregCallId: null };
const response = keySource === "platform" ? await withCreditAlertsDeferred(scope, deps.own) : await deps.own();
flushCreditAlerts(scope); // a 200 sentinel (email not unlocked) is not covered by treg
return { response, via: "apollo", tregCostMicro: null, tregCallId: null };
} catch (error) {
if (keySource !== "platform" || !(error instanceof ApolloCreditsExhaustedError)) throw error;
if (keySource !== "platform" || !(error instanceof ApolloCreditsExhaustedError)) {
flushCreditAlerts(scope);
throw error;
}
ownKeyExhaustedUntil = now() + OWN_KEY_EXHAUSTED_RECHECK_MS;
console.warn(`[Apollo Service][reveal-route] own Apollo key out of credits, calling Apollo through treg for ${OWN_KEY_EXHAUSTED_RECHECK_MS / 60000} min`);
}
}
const creds = await (deps.credentials ?? (() => platformTregCredentials(deps.callerPath)))();
const r = await callApolloViaTreg<T>(deps.treg, creds, deps.fetchImpl);
return { response: r.response, via: "treg", tregCostMicro: r.costMicro, tregCallId: r.callId };
try {
const creds = await (deps.credentials ?? (() => platformTregCredentials(deps.callerPath)))();
const r = await callApolloViaTreg<T>(deps.treg, creds, deps.fetchImpl);
if (scope.deferred.length > 0) {
console.log(`[Apollo Service][reveal-route] own Apollo key out of credits, served through treg ${deps.treg.endpoint}: no staff alert`);
}
return { response: r.response, via: "treg", tregCostMicro: r.costMicro, tregCallId: r.callId };
} catch (tregError) {
// The client path really lost data: our key is out AND treg could not answer.
const why = tregError instanceof Error ? tregError.message : String(tregError);
if (scope.deferred.length > 0) {
flushCreditAlerts(scope, why);
} else {
reportApolloCreditsExhausted(deps.alertIdentity, {
operation: deps.treg.endpoint,
reason: `Apollo is out of credits (our key skipped for the recheck window) and the treg fallback failed too: ${why}`,
});
}
throw tregError;
}
}

export interface RevealDeps<T> extends Omit<FallbackDeps<T>, "treg"> {
Expand Down
1 change: 1 addition & 0 deletions src/lib/buying-signals.ts
Original file line number Diff line number Diff line change
Expand Up @@ -435,6 +435,7 @@ async function fetchJobPostingsUnderLock(apolloOrganizationId: string, ctx: Evid
own: () => getOrganizationJobPostings(key, apolloOrganizationId, ctx.alertIdentity),
treg: { endpoint: "apollo.companies.jobs", method: "GET", query: { organization_id: apolloOrganizationId } },
callerPath: "/enrich",
alertIdentity: ctx.alertIdentity,
});
} catch (error) {
await releaseHold("failed");
Expand Down
58 changes: 58 additions & 0 deletions src/lib/credit-alert.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
* or any existing endpoint — a missing var surfaces as a logged alert failure.
*/

import { AsyncLocalStorage } from "node:async_hooks";
import { fetchWithRetry } from "./fetch-retry.js";
import type { CreditProvider } from "./provider-error.js";

Expand Down Expand Up @@ -191,6 +192,15 @@ export function reportProviderCreditsExhausted(
identity: CreditAlertIdentity | undefined,
detail: CreditAlertDetail,
): void {
const scope = fallbackScope.getStore();
if (scope) {
// A fallback stands behind this call: only raise the alert if the fallback
// fails too (flushCreditAlerts). An exhaustion the fallback absorbed lost
// nothing for the client.
scope.deferred.push({ provider, identity, detail });
return;
}

if (!identity?.orgId) {
console.warn(
`[apollo-service][credit-alert] ${provider} credits look exhausted (${detail.operation}: ${detail.reason}) but the caller carried no org — no staff alert sent`,
Expand All @@ -206,3 +216,51 @@ export function reportProviderCreditsExhausted(
console.error("[apollo-service][credit-alert] Failed to send staff alert:", err);
});
}

// ─── Fallback scope ──────────────────────────────────────────────────────────

/**
* MON-APOLLO-CREDITS-1111: when our platform Apollo key is out of credits, the
* same call is answered by Apollo THROUGH treg (`withTregFallback`), so the
* client loses nothing. The alert used to fire at the Apollo chokepoint BEFORE
* the fallback ran, so every covered call still raised « A paid data provider
* is out of credits » (25 times in 30 days, all served by treg). An alert now
* means the client path really lost data.
*
* Inside `withCreditAlertsDeferred`, every alert the call raises is HELD in
* the scope instead of sent. The fallback owner then either drops them (the
* fallback served) or sends them with `flushCreditAlerts` (it did not).
*/
interface DeferredCreditAlert {
provider: CreditProvider;
identity: CreditAlertIdentity | undefined;
detail: CreditAlertDetail;
}

export interface CreditAlertScope {
deferred: DeferredCreditAlert[];
}

const fallbackScope = new AsyncLocalStorage<CreditAlertScope>();

/** Run `fn` with its credit alerts held in `scope` (see above). */
export function withCreditAlertsDeferred<T>(scope: CreditAlertScope, fn: () => Promise<T>): Promise<T> {
return fallbackScope.run(scope, fn);
}

/**
* Send the alerts held in `scope`, outside any scope. `fallbackFailure`, when
* given, says why the fallback could not cover them; it is appended to the
* reason so the staff alert names both walls.
*/
export function flushCreditAlerts(scope: CreditAlertScope, fallbackFailure?: string): void {
const held = scope.deferred.splice(0);
fallbackScope.exit(() => {
for (const a of held) {
reportProviderCreditsExhausted(a.provider, a.identity, {
...a.detail,
reason: fallbackFailure ? `${a.detail.reason}; the treg fallback failed too: ${fallbackFailure}` : a.detail.reason,
});
}
});
}
Loading
Loading