DEV Community: Nimbryxo-studio

Your moneyline parser can reverse the odds ranking

Nimbryxo-studio — Fri, 13 Nov 2026 10:01:24 +0000

Your dashboard puts a long shot above the favorite. You check the sorting function. It's fine. The damage happened before the rows reached it.

A field named homeMoneyline looks reassuringly numeric. But the Actor preserves published prices, including fractional strings. Applying parseFloat before identifying the notation can reverse your implied-probability ranking.

Example: this assertion shows a fractional long shot incorrectly outranking an American-priced favorite after a permissive conversion.

import assert from 'node:assert/strict';

const americanProbability = (price: number) =>
  price > 0
    ? 100 / (price + 100)
    : -price / (-price + 100);

const fractionalLongShot = '20/1';
const americanFavorite = -110;

assert.ok(
  americanProbability(parseFloat(fractionalLongShot)) >
  americanProbability(americanFavorite),
);

Nothing throws. That's the annoying part.

The reusable pattern is to preserve the source value, normalize only recognized representations, and give ingestion its own replay identity. Your dashboard should receive an explicit review state whenever those steps can't produce a trustworthy comparison.

Fetch the prices without interpreting them

Use odds mode because it returns a record per sportsbook line. Keep the request narrow while you establish the contract.

Example: this input selects a soccer competition so the workflow also handles a draw price.

import { ApifyClient } from 'apify-client';
import { z } from 'zod';
import pg from 'pg';

const input = {
  mode: 'odds',
  league: 'epl',
  date: '2024-01-15',
  maxItems: 100,
};

const client = new ApifyClient({
  token: process.env.APIFY_TOKEN,
});

const started = process.env.REPLAY_RUN_ID
  ? { id: process.env.REPLAY_RUN_ID }
  : await client
      .actor('nimbryxo-studio/espn-live-scores-schedules-and-game-odds-scraper')
      .start(input);

const run = await client.run(started.id).waitForFinish();

if (
  !run ||
  !['SUCCEEDED', 'FAILED', 'TIMED-OUT', 'ABORTED'].includes(run.status)
) {
  throw new Error('A terminal run is required before dataset ingestion');
}

Set REPLAY_RUN_ID when reprocessing an existing dataset. Leave it unset when you want a fresh capture.

Those actions have different meanings. A fresh capture reads the live source again. It can't reconstruct an earlier price that has disappeared.

Waiting for a terminal state also gives pagination a fixed ingestion boundary. A failed run may still have useful records; the code below preserves available items and raises a separate run-status signal.

How do you normalize fractional moneylines?

Fractional odds express profit relative to stake. Their implied probability is the denominator divided by the sum of numerator and denominator.

Decimal odds express total return. American odds use a signed price. These representations need separate branches, even when their values all fit inside a JavaScript string.

Example: the parser below checks the published fractional forms and leaves bare numbers behind an explicit provider policy.

type Notation = 'american' | 'decimal';
type Price = string | number | undefined;
type NormalizedPrice = {
  probability: number | null;
  state: 'ok' | 'missing' | 'unrecognized';
};

const providerNotation = new Map<string, Notation>();

function normalizePrice(
  value: Price,
  notation?: Notation,
): NormalizedPrice {
  if (value === undefined) {
    return { probability: null, state: 'missing' };
  }

  const text = String(value).trim();
  const fraction = /^(\d+(?:\.\d+)?)\/(\d+(?:\.\d+)?)$/.exec(text);
  let probability: number | undefined;

  if (fraction) {
    const numerator = Number(fraction[1]);
    const denominator = Number(fraction[2]);
    const total = numerator + denominator;

    if (
      numerator >= 0 &&
      denominator > 0 &&
      Number.isFinite(total)
    ) {
      probability = denominator / total;
    }
  } else if (/^[+-]?(?:\d+(?:\.\d*)?|\.\d+)$/.test(text)) {
    const numeric = Number(text);

    if (Number.isFinite(numeric)) {
      if (notation === 'american' && Math.abs(numeric) >= 100) {
        probability = americanProbability(numeric);
      } else if (notation === 'decimal' && numeric > 1) {
        probability = 1 / numeric;
      }
    }
  }

  if (
    probability === undefined ||
    !Number.isFinite(probability) ||
    probability <= 0 ||
    probability > 1
  ) {
    return { probability: null, state: 'unrecognized' };
  }

  return { probability, state: 'ok' };
}

assert.equal(normalizePrice(fractionalLongShot).probability, 1 / 21);
assert.equal(normalizePrice('1/28').probability, 28 / 29);
assert.ok(
  normalizePrice(fractionalLongShot).probability! <
  americanProbability(americanFavorite),
);

The provider map starts empty deliberately. Configure it only after establishing the representation used by that feed. A field name, a familiar magnitude, or a successful cast is insufficient evidence.

A provider-wide policy also has limits. If representations vary by market, your policy needs that context before you enable numeric normalization.

These probabilities provide a comparison scale. Bookmaker margin means they aren't calibrated forecasts. Compare matching outcomes and market definitions before interpreting a difference as an opportunity.

Preserve additions, quarantine broken contracts

Use passthrough validation for fields you consume. Extra fields remain available without breaking ingestion.

Strict validation would reject every addition, including harmless enrichment. Passthrough accepts that evolution, while still rejecting changed types in the identifiers and prices your calculations depend on.

Condition	Storage decision	Dashboard decision
Recognized prices and valid identifiers	Keep raw and normalized values	Eligible for comparison
Unrecognized or missing market prices	Keep row with review reasons	Exclude from ranking
Invalid consumed fields	Keep raw row with validation issues	Exclude from ranking

For this soccer request, eligibility requires the draw price as well as the team prices. Otherwise, your apparent full-market comparison silently omits an outcome.

This schema checks only consumed fields, and the mapper retains every original record.

const identifier = z.union([
  z.string().min(1),
  z.number().finite(),
]).transform(String);

const sourcePrice = z.union([
  z.string().min(1),
  z.number().finite(),
]);

const oddsSchema = z.object({
  gameId: identifier,
  providerId: identifier,
  recordType: z.string().min(1),
  scrapedAt: z.string().datetime({ offset: true }),
  homeMoneyline: sourcePrice.optional(),
  awayMoneyline: sourcePrice.optional(),
  drawMoneyline: sourcePrice.optional(),
}).passthrough();

function makeObservation(raw: unknown, itemIndex: number) {
  const base = {
    run_id: run.id,
    item_index: itemIndex,
    run_status: run.status,
    game_id: null as string | null,
    provider_id: null as string | null,
    scraped_at: null as string | null,
    home_probability: null as number | null,
    away_probability: null as number | null,
    draw_probability: null as number | null,
    normalizer: 'fractional-v1',
    raw,
  };

  const checked = oddsSchema.safeParse(raw);

  if (!checked.success) {
    return {
      ...base,
      ingest_state: 'schema_review',
      issues: checked.error.issues,
    };
  }

  const row = checked.data;
  const notation = providerNotation.get(row.providerId);
  const prices = [
    normalizePrice(row.homeMoneyline, notation),
    normalizePrice(row.awayMoneyline, notation),
    normalizePrice(row.drawMoneyline, notation),
  ];
  const outcomes = ['home', 'away', 'draw'];
  const issues = prices.flatMap((price, index) =>
    price.state === 'ok'
      ? []
      : [{ outcome: outcomes[index], reason: price.state }],
  );

  return {
    ...base,
    game_id: row.gameId,
    provider_id: row.providerId,
    scraped_at: row.scrapedAt,
    home_probability: prices[0].probability,
    away_probability: prices[1].probability,
    draw_probability: prices[2].probability,
    ingest_state: issues.length ? 'price_review' : 'ready',
    issues,
  };
}

Missing values and unfamiliar representations receive different reasons. Absence can be normal source behavior. An unfamiliar notation needs investigation before normalization resumes.

The schema intentionally leaves nested side-detail objects untouched. Guessing their internal structure would merely move the permissive parser problem elsewhere.

Give replay an ingestion identity

gameId identifies the event. providerId identifies the sportsbook provider. Their combination is useful for grouping observations, but it can't identify a historical price snapshot.

Upserting on that pair would overwrite movement you intended to retain.

Use the run ID and absolute dataset position for replay deduplication instead. This identifies an ingestion occurrence, not a source-level odds line.

The distinction looks like this:

same terminal dataset + same position -> replace derived observation
fresh capture + same game/provider    -> retain another observation
invalid price or shape               -> retain raw row for review

That key depends on replaying the same terminal dataset in the same order. Don't change filtering, sorting, or dataset contents underneath it. Exported or reordered data needs a persisted ingestion identity of its own.

Keep typed columns for joins, filtering, and ranking. Keep raw JSONB for audit and reprocessing. JSONB alone makes every dashboard query responsible for normalization; typed columns alone discard the evidence needed to fix it.

This table stores both, with a normalizer version attached to derived values.

const pool = new pg.Pool();

const createTable = `
  CREATE TABLE IF NOT EXISTS odds_observations (
    run_id text NOT NULL,
    item_index bigint NOT NULL,
    run_status text NOT NULL,
    game_id text,
    provider_id text,
    scraped_at timestamptz,
    home_probability double precision,
    away_probability double precision,
    draw_probability double precision,
    normalizer text NOT NULL,
    raw jsonb NOT NULL,
    ingest_state text NOT NULL,
    issues jsonb NOT NULL,
    PRIMARY KEY (run_id, item_index)
  )
`;

const upsertBatch = `
  INSERT INTO odds_observations
  SELECT *
  FROM jsonb_populate_recordset(
    NULL::odds_observations,
    $1::jsonb
  )
  ON CONFLICT (run_id, item_index) DO UPDATE SET
    run_status = EXCLUDED.run_status,
    game_id = EXCLUDED.game_id,
    provider_id = EXCLUDED.provider_id,
    scraped_at = EXCLUDED.scraped_at,
    home_probability = EXCLUDED.home_probability,
    away_probability = EXCLUDED.away_probability,
    draw_probability = EXCLUDED.draw_probability,
    normalizer = EXCLUDED.normalizer,
    raw = EXCLUDED.raw,
    ingest_state = EXCLUDED.ingest_state,
    issues = EXCLUDED.issues
`;

Reprocessing replaces the current derived interpretation. If you also need an audit of normalization changes, retain those versions separately.

Commit every dataset page as a batch

Read until pagination is exhausted. The input cap limits returned records; it doesn't replace dataset pagination.

Each page becomes a single database statement. A database failure rolls back that statement, while previously committed pages remain available. Replay starts from the beginning and safely replaces them.

This loop retains rejected records, commits valid observations, and emits a structured attention event after ingestion.

let offset = 0;
let reviewCount = 0;
const pageSize = 100;

try {
  await pool.query(createTable);

  while (true) {
    const { items } = await client
      .dataset(run.defaultDatasetId)
      .listItems({ offset, limit: pageSize });

    if (items.length === 0) break;

    const batch = items.map((item, index) =>
      makeObservation(item, offset + index),
    );

    await pool.query(upsertBatch, [JSON.stringify(batch)]);

    reviewCount += batch.filter(
      row => row.ingest_state !== 'ready',
    ).length;
    offset += items.length;
  }

  const capReached = offset >= input.maxItems;

  if (
    reviewCount > 0 ||
    run.status !== 'SUCCEEDED' ||
    capReached ||
    offset === 0
  ) {
    console.error(JSON.stringify({
      event: 'odds_ingest_attention',
      runId: run.id,
      runStatus: run.status,
      reviewCount,
      capReached,
      emptyDataset: offset === 0,
    }));
  }
} finally {
  await pool.end();
}

Route that event to your job runner's alert rule. Let database and dataset-read exceptions fail the job normally.

Hitting the cap signals possible truncation, not proof that more records exist. An empty dataset needs context rather than an automatic outage page. A completed ingestion with quarantined prices deserves attention because the dashboard now has a known coverage gap.

Your ranking query must require ingest_state = 'ready'. For published snapshots, also require an accepted run status and an explicit freshness policy.

Missing prices aren't necessarily schema drift

The worked source is ESPN Live Scores, Schedules & Game Odds Scraper.

It returns published values and omits unavailable fields. Provider coverage varies by event, and lines can disappear after a game ends. Soccer lines may supply moneyline and draw prices instead of spread and total.

Those behaviors explain several failures that can look like parser bugs. Re-running a historical request can't restore a removed line, and a missing spread doesn't make a soccer market unusable.

League-response failures produce diagnostic records rather than replacement game records. Those diagnostics fail this odds-specific schema and remain in raw storage with review issues. They don't become blank sportsbook observations.

Keep availability alerts separate from contract alerts. A changed price type can require a code fix; a removed source price requires a coverage decision.

Before you trust the ranking

Confirm notation before enabling a provider's numeric prices.
Inspect review reasons before admitting quarantined rows.
Include the draw when comparing the full soccer market.
Replay an existing dataset to repair normalization.
Start a fresh capture when you want a new observation.
Check run status, freshness, and cap signals before publishing.

A number parser accepting a price isn't evidence that it understood it.

Smaller scheduled batches can raise your event budget

Nimbryxo-studio — Thu, 12 Nov 2026 09:51:02 +0000

Your row budget stayed flat. Your scheduled starts multiplied.

You split a weekly company-directory refresh into smaller jobs, hoping to make failures less annoying. Then the event budget grows. The row allowance still looks innocent.

Splitting the same company-record allowance across more runs increases start-event fees. For Nimbryxo-studio's directory Actor, starts are memory-dependent, result events follow dataset items, and platform usage is charged separately.

Make the input part of the budget

Start with a restricted input your estimator actually understands. Supporting every mode would widen the billing assumptions before you need them.

Example: this input selects a bulk listing path, with enrichment explicitly disabled.

import assert from 'node:assert/strict';
import { z } from 'zod';
import pg from 'pg';

const BrowseInput = z.object({
  mode: z.literal('browse'),
  location: z.literal('chicago'),
  includeDetails: z.boolean(),
  includeProfile: z.boolean(),
  includeBenefits: z.boolean(),
  maxItems: z.number().int().min(1).max(10000),
}).strict();

const run_input = BrowseInput.parse({
  mode: 'browse',
  location: 'chicago',
  includeDetails: false,
  includeProfile: false,
  includeBenefits: false,
  maxItems: 100,
});

This schema deliberately accepts only a subset of the Actor's inputs. Changes outside that subset require another budgeting review.

The explicit flags matter. A saved task shouldn't quietly inherit enrichment behavior your cost model never considered.

What actually creates a charge?

The Store prices describe separate events with different triggers.

Charge	What drives it	Budgeting consequence
`apify-default-dataset-item`	A result in the default dataset	Use the company-record limit for this restricted browse path
`apify-actor-start`	Starting the Actor, with event count dependent on memory	Reserve a known start-event allowance for every attempt
Platform usage	Resources consumed during execution	Keep a separate allowance; event rates cannot price it

The start charge uses one event per GB, with a minimum of one event. A run boundary therefore has a price even when your total record allowance stays unchanged.

That explains why smaller batches can cost more in event fees. You're buying additional starts, alongside the records.

Memory belongs in the scheduling configuration. If it changes, invalidate the old estimate rather than carrying its start-event allowance forward.

Enrichment changes work before it changes rows

maxItems caps company records returned. It doesn't describe all the requests needed to produce them.

includeProfile and includeBenefits each add one request per company. includeDetails fetches detail records and can fall back to profile pages when details are missing or lack a tech stack.

Large runs with details enabled are slow. Use includeDetails: false for bulk listing when those fields aren't needed.

For a recurring directory refresh, separate discovery from selective enrichment. Refresh the listing broadly, then enrich companies whose additional fields support an actual decision.

That separation makes spend easier to explain. It doesn't make enrichment free, and the published event rates can't bound its platform usage.

Calculate an event-fee ceiling before launch

Use integer money internally. Floating-point rounding is an unnecessary guest in a budget ledger.

This function prices the restricted input, requires an explicit start-event allowance, and checks the published arithmetic without printing output. Its allowance arguments belong to your scheduler, not the Actor input.

const ratesMicroUsd = {
  FREE: { item: 990, start: 100000 },
  GOLD: { item: 490, start: 30000 },
} as const;

type Tier = keyof typeof ratesMicroUsd;

const PositiveInteger = z.number().int().min(1)
  .max(Number.MAX_SAFE_INTEGER);

function eventCeilingMicroUsd(
  input: unknown,
  tier: Tier,
  plannedStarts: unknown,
  startEventAllowance: unknown,
): number {
  const parsed = BrowseInput.parse(input);
  const starts = PositiveInteger.parse(plannedStarts);
  const startEvents = PositiveInteger.parse(startEventAllowance);
  const rate = ratesMicroUsd[tier];

  const perRun =
    parsed.maxItems * rate.item + startEvents * rate.start;
  const total = perRun * starts;

  if (!Number.isSafeInteger(perRun) || !Number.isSafeInteger(total)) {
    throw new Error('Event reservation exceeds safe integer precision');
  }

  return total;
}

function usdString(microUsd: number): string {
  return `${Math.trunc(microUsd / 1_000_000)}.${
    String(microUsd % 1_000_000).padStart(6, '0')
  }`;
}

for (const [maxItems, tier, expectedUsd] of [
  [100, 'FREE', 0.199],
  [100, 'GOLD', 0.079],
  [1000, 'FREE', 1.09],
  [1000, 'GOLD', 0.52],
] as const) {
  assert.equal(
    eventCeilingMicroUsd({ ...run_input, maxItems }, tier, 1, 1)
      / 1_000_000,
    expectedUsd,
  );
}

assert.throws(() =>
  eventCeilingMicroUsd(run_input, 'FREE', 1, undefined),
);

Treat the result as a conditional event-fee ceiling. It assumes the configured memory fits the supplied start-event allowance and the submitted input follows this company-record path.

It isn't a whole-run spending ceiling. Adding an enrichment flag leaves this event formula unchanged while adding work that consumes platform usage.

The function can't verify your memory configuration. Keep that configuration and its verified event allowance together, and reject scheduling changes that update only one.

Budget attempts, including scheduled retries

A recurring budget needs a finite planning window. “Weekly” describes timing; it doesn't specify how many launches you authorize.

Keep planned starts and retry starts separate. Otherwise, retries become the place where the budget quietly acquires imagination.

Example: these plans compare equal scheduled record allowances, then reserve additional attempts explicitly. The start-event allowance is illustrative; replace it with an allowance verified for your configured memory.

const schedulePlans = {
  bulk: {
    input: BrowseInput.parse({ ...run_input, maxItems: 1000 }),
    plannedStarts: 1,
    retryStarts: 2,
  },
  split: {
    input: run_input,
    plannedStarts: 10,
    retryStarts: 2,
  },
};

const configuredStartEventsPerRun = 1;

const comparison = Object.fromEntries(
  Object.entries(schedulePlans).map(([name, plan]) => [
    name,
    {
      scheduledRecords: plan.input.maxItems * plan.plannedStarts,
      scheduledEventMicroUsd: eventCeilingMicroUsd(
        plan.input,
        'FREE',
        plan.plannedStarts,
        configuredStartEventsPerRun,
      ),
      reservedEventMicroUsd: eventCeilingMicroUsd(
        plan.input,
        'FREE',
        plan.plannedStarts + plan.retryStarts,
        configuredStartEventsPerRun,
      ),
    },
  ]),
);

assert.equal(
  comparison.bulk.scheduledRecords,
  comparison.split.scheduledRecords,
);
assert.ok(
  comparison.split.scheduledEventMicroUsd
    > comparison.bulk.scheduledEventMicroUsd,
);

The comparison separates scheduled work from permission to repeat it. Equal scheduled allowances don't imply equal retry allowances: repeating a larger batch authorizes more records.

Retries can emit records already collected by an earlier attempt. Deduplicating those records downstream doesn't undo their event charges.

Choose smaller batches for freshness or failure isolation when those benefits justify repeated starts. A row allowance also doesn't guarantee progress through the directory; batch boundaries need their own coverage design.

Reserve locally, cap remotely

Preflight admission answers: can this planning window afford another attempt?

A platform-side cap answers a different question: how much may this particular pay-per-event run charge?

Use both. A local estimate can reject work before it starts and coordinate concurrent workers. It can't enforce spending inside an already running process.

The following functions create your application's accounting tables and reserve an attempt atomically. The table columns are local bookkeeping, separate from the Actor's input and output.

async function initializeEventBudget(
  pool: pg.Pool,
  period: string,
  periodMicroUsd: number,
): Promise<void> {
  PositiveInteger.parse(periodMicroUsd);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS directory_event_budget (
      period text PRIMARY KEY,
      remaining_micro_usd bigint NOT NULL
        CHECK (remaining_micro_usd >= 0)
    )
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS directory_event_attempt (
      period text NOT NULL REFERENCES directory_event_budget(period),
      attempt_key text NOT NULL,
      amount_micro_usd bigint NOT NULL CHECK (amount_micro_usd > 0),
      submitted_input jsonb NOT NULL,
      submission_status integer,
      submission_body text,
      PRIMARY KEY (period, attempt_key)
    )
  `);

  await pool.query(`
    INSERT INTO directory_event_budget (period, remaining_micro_usd)
    VALUES ($1, $2)
    ON CONFLICT (period) DO NOTHING
  `, [period, periodMicroUsd]);
}

async function reserveAndStart(
  pool: pg.Pool,
  period: string,
  attemptKey: string,
  input: unknown,
  tier: Tier,
  startEventAllowance: unknown,
): Promise<Response> {
  const token = process.env.APIFY_TOKEN;
  if (!token) throw new Error('APIFY_TOKEN is required');
  if (!attemptKey) throw new Error('An attempt key is required');

  const parsed = BrowseInput.parse(input);
  const reserved = eventCeilingMicroUsd(
    parsed, tier, 1, startEventAllowance,
  );

  const url = new URL(
    'https://api.apify.com/v2/acts/' +
    'nimbryxo-studio~builtin-tech-company-directory-and-hiring-signal/runs',
  );
  url.searchParams.set('maxTotalChargeUsd', usdString(reserved));

  const admission = await pool.query(`
    WITH debited AS (
      UPDATE directory_event_budget
      SET remaining_micro_usd = remaining_micro_usd - $2::bigint
      WHERE period = $1
        AND remaining_micro_usd >= $2::bigint
      RETURNING period
    )
    INSERT INTO directory_event_attempt (
      period, attempt_key, amount_micro_usd, submitted_input
    )
    SELECT period, $3::text, $2::bigint, $4::jsonb
    FROM debited
    RETURNING attempt_key
  `, [period, reserved, attemptKey, JSON.stringify(parsed)]);

  if (admission.rowCount !== 1) {
    throw new Error('Budget unavailable');
  }

  const response = await fetch(url, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${token}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify(parsed),
  });

  await pool.query(`
    UPDATE directory_event_attempt
    SET submission_status = $3::integer,
        submission_body = $4::text
    WHERE period = $1 AND attempt_key = $2
  `, [
    period,
    attemptKey,
    response.status,
    await response.clone().text(),
  ]);

  if (!response.ok) throw new Error('Run submission needs review');
  return response;
}

Initialize the budget once per planning period. The conflict clause deliberately prevents a worker restart from replenishing an existing allowance.

The atomic statement prevents concurrent workers from spending the same remaining balance. A duplicate attempt key raises a database error and rolls back that statement, including its debit.

Use a stable key for each authorized attempt. A worker restart reuses it; an explicitly authorized retry gets a fresh key.

A lost response creates uncertainty: the request may have started a run. Keep that reservation until you reconcile it. Automatically refunding it would turn a transport failure into permission to spend again.

Persist the returned run identifier in your worker's job record. Reconcile each reservation against actual event charges afterward, releasing unused allowance only once.

This code handles admission and submission, not that reconciliation workflow. Keep platform usage in its separate ledger, and don't treat maxTotalChargeUsd as a ceiling for that component.

A capped extraction needs a publication gate

The worked example is BuiltIn Tech Company Directory & Hiring Signal Scraper. The budgeting pattern applies whenever starts, results, and execution resources have different charging rules.

The operational catch arrives downstream: a stopped run may leave a partial dataset. Treat retained rows as observations, not proof that a directory refresh completed.

Keep the flow explicit.

reserve attempt -> submit capped run -> retain run evidence
                                           |
                                           v
                                    stage dataset
                                           |
                                    publication gate
                                           |
                                    current snapshot

Retain the submitted input, run outcome, warnings, and dataset reference together. Stage all dataset pages before deciding whether the result can replace your current snapshot.

A successful status alone can't establish coverage. The Actor can warn when its page safety cap stops browsing before maxItems. Conversely, returning fewer records than the limit doesn't prove a safety stop.

Your publication gate needs those distinctions. A failed, interrupted, or early-stopped extraction shouldn't erase companies from the previous snapshot merely because they weren't observed.

Even reaching the requested limit establishes a bounded observation, not directory completeness. Your downstream deletion policy needs stronger evidence than “the job returned rows.”

For a bounded refresh, updating observed companies can be reasonable while withholding deletion decisions. Keep the coverage state visible to whatever consumes the snapshot.

Before the next schedule fires

Fix the planning window, planned starts, and retry allowance.
Supply a start-event allowance tied to the configured memory.
Reserve event fees atomically before submitting each attempt.
Submit the validated maxItems and remote charge cap together.
Keep uncertain submissions reserved until reconciliation.
Budget platform usage separately.
Publish staged records only after checking outcome, warnings, and coverage.

A cheaper-looking row target isn't a schedule budget. Count the attempts you're authorizing.

Your partners array can reverse the buyer relationship

Nimbryxo-studio — Wed, 11 Nov 2026 10:08:27 +0000

Your next dashboard refresh can turn a customer into a sourcing lead.

You're merging importer and supplier profiles into a relationship table. Treating every profile owner as the buyer looks convenient. It also puts the mistake somewhere nobody enjoys debugging: inside otherwise valid rows.

For a US importer, partners contains suppliers; for a supplier, it contains customers. Flattening that field with a fixed owner-to-supplier rule reverses supplier-profile relationships.

The repair starts before the upsert. Preserve the profile context, resolve direction explicitly, and keep unidentified partners out of the graph.

Start with a bounded profile job

Use profile mode when you already know the pages you need. Disable shipment rows here because a profile relationship and a bill of lading need different ingestion contracts.

Install apify-client, zod, and pg. The following cells form a single TypeScript module with top-level await.

Example input: these company and supplier paths make the intended profile contexts explicit.

import { ApifyClient } from 'apify-client';
import { z } from 'zod';
import pg from 'pg';

const Input = z.object({
  mode: z.literal('companyProfile'),
  startUrls: z.array(
    z.object({ url: z.string().min(1) }).strict()
  ).min(1),
  includeShipments: z.literal(false),
  maxItems: z.number().int().min(1).max(500),
}).strict();

const input = Input.parse({
  mode: 'companyProfile',
  startUrls: [
    { url: 'company/samsung-electronics-america' },
    { url: 'supplier/foxconn' },
  ],
  includeShipments: false,
  maxItems: 20,
});

const client = new ApifyClient({
  token: process.env.APIFY_TOKEN,
});

This input validator deliberately covers this job's configuration. It isn't an attempt to mirror every mode.

Avoid adding recency filters to a relationship refresh without deciding what absence means. A date filter excludes records without a date, which can make an existing account disappear from your observation set.

Wait once, then ingest the dataset

Starting the source job and committing its records are separate operations. Keep the run identifier in your scheduler so an ingestion retry can return to the existing dataset.

This cell starts the run, waits, and reads its dataset.

const started = await client
  .actor('nimbryxo-studio/importyeti-us-importer-and-global-supplier-search')
  .start(input);

const run = await client.run(started.id).waitForFinish();

if (!run?.defaultDatasetId) {
  throw new Error(`Run ${started.id} has no readable dataset`);
}

const { items } = await client
  .dataset(run.defaultDatasetId)
  .listItems();

Don't throw away collected records solely because the final status isn't SUCCEEDED. Ingest usable observations and record that status alongside the ingestion report.

Conversely, a successful status doesn't establish that every requested profile arrived. That needs a separate comparison.

Validate the envelope and keep the evidence

The record discriminator tells you which ingestion stream owns a row. This job accepts profile records; anything else goes into review.

Validate fields used for identity and direction. Keep partners opaque because its nested identity fields aren't specified here. Guessing a convenient partner.id would move uncertainty into your primary key.

This cell validates profiles while preserving their original payloads.

const Profile = z.object({
  recordType: z.enum(['company', 'supplier']),
  dataset: z.enum(['us-import', 'us-export', 'mx']),
  companyUrl: z.string().url(),
  sourceDomain: z.string().optional(),
  type: z.string().optional(),
  partnerRole: z.string().min(1).optional(),
  partners: z.unknown().optional(),
  scrapedAt: z.string().optional(),
}).passthrough();

type ProfileRecord = z.infer<typeof Profile>;

const valid: Array<{
  profile: ProfileRecord;
  raw: unknown;
}> = [];

const rejected: Array<{
  raw: unknown;
  reason: string;
}> = [];

for (const raw of items as unknown[]) {
  const parsed = Profile.safeParse(raw);

  if (!parsed.success) {
    rejected.push({
      raw,
      reason: JSON.stringify(parsed.error.issues),
    });
    continue;
  }

  valid.push({ profile: parsed.data, raw });
}

A failed record stays available for inspection. It doesn't become an empty profile, and it doesn't overwrite a previously usable observation.

Choice	What it protects	What it costs
Strict outer validation	Detects every added field	Harmless enrichment can stop ingestion
Passthrough outer validation	Checks consumed fields while retaining additions	New fields need a separate discovery process
Opaque partner payload	Prevents invented identity mappings	Relationship materialization waits for review

We'd keep typed keys and direction decisions beside raw JSONB. Typed columns make joins predictable; raw payloads let you repair a mapping without fetching the source again.

Raw JSONB also needs retention rules. Otherwise your evidence archive quietly becomes your storage policy.

Which way does the relationship point?

Choose a canonical graph direction before naming database columns. Here, every eventual edge means supplier → buyer.

The owner's position changes with the profile context.

Importer profile:  partner ──supplies──▶ owner
Supplier profile:  owner   ──supplies──▶ partner

Stored edge:       supplier ──────────▶ buyer

partnerRole accompanies the breakdown, but its exact values aren't enumerated here. The resolver therefore requires a reviewed vocabulary instead of assuming singulars, plurals, or capitalization.

The empty map below is intentional. Populate it only after reviewing actual role values. Until then, profiles persist with a review decision.

This cell checks role agreement against the supported US import profile contexts.

type PartnerMeaning = 'supplier' | 'buyer';
const reviewedRoles = new Map<string, PartnerMeaning>();

function profilePath(value: string): string {
  const url = new URL(value, 'https://www.importyeti.com/');

  if (
    url.protocol !== 'https:' ||
    !['importyeti.com', 'www.importyeti.com'].includes(url.hostname)
  ) {
    throw new Error('Unexpected profile origin');
  }

  return url.pathname.replace(/\/+$/, '') || '/';
}

function resolveDirection(profile: ProfileRecord, path: string) {
  const review = (reason: string) => ({
    direction: 'review',
    reason,
  });

  if (profile.partners == null) {
    return review('missing_partner_payload');
  }

  const expected: PartnerMeaning | undefined =
    profile.dataset === 'us-import' &&
    profile.recordType === 'company' &&
    path.startsWith('/company/')
      ? 'supplier'
      : profile.dataset === 'us-import' &&
          profile.recordType === 'supplier' &&
          path.startsWith('/supplier/')
        ? 'buyer'
        : undefined;

  if (!expected) return review('unsupported_profile_context');
  if (!profile.partnerRole) return review('missing_partner_role');

  const meaning = reviewedRoles.get(profile.partnerRole);
  if (!meaning) return review('unreviewed_partner_role');
  if (meaning !== expected) return review('role_context_conflict');

  return {
    direction: meaning === 'supplier'
      ? 'partner_to_owner'
      : 'owner_to_partner',
    reason: 'role_and_context_agree',
  };
}

Other dataset contexts remain in review until you establish their semantics. type is retained, but the resolver doesn't assign meanings to unspecified values.

Even an agreed direction can't identify a partner. Before creating edges, you still need an explicit, reviewed mapping from the nested payload to partner profile identities. Names alone aren't enough: spelling variants can split a business, while shared names can merge unrelated businesses.

That boundary is useful. A review queue is cheaper than a confidently backward sourcing dashboard.

Give observations a practical upsert key

Use (dataset, normalized profile path) as the observation slot. Keep dataset prefixes, path case, and slug spelling intact; remove query strings, fragments, and trailing slashes.

This is a practical URL identity, not a guaranteed permanent company identifier. Redirects or renamed profiles can create another slot. Resolve those through a reviewed alias table rather than increasingly enthusiastic URL normalization.

The schema separates source payloads from direction decisions and retains the supplying run.

const ddl = `
CREATE TABLE IF NOT EXISTS trade_profile (
  dataset text NOT NULL,
  profile_path text NOT NULL,
  run_id text NOT NULL,
  raw jsonb NOT NULL,
  PRIMARY KEY (dataset, profile_path)
);

CREATE TABLE IF NOT EXISTS trade_profile_direction (
  dataset text NOT NULL,
  profile_path text NOT NULL,
  partner_role text,
  direction text NOT NULL CHECK (
    direction IN ('partner_to_owner', 'owner_to_partner', 'review')
  ),
  reason text NOT NULL,
  PRIMARY KEY (dataset, profile_path),
  FOREIGN KEY (dataset, profile_path)
    REFERENCES trade_profile (dataset, profile_path)
);

CREATE TABLE IF NOT EXISTS trade_ingestion_run (
  run_id text PRIMARY KEY,
  status text NOT NULL,
  report jsonb NOT NULL
);
`;

Keep this DDL in a migration for deployment. It's included here so the recipe can initialize an empty database.

The next cell stages unique observations, quarantines conflicting duplicate keys, and writes batches inside a transaction. Configure pg through your usual database environment variables.

type StagedRow = {
  dataset: string;
  profile_path: string;
  partner_role: string | null;
  direction: string;
  reason: string;
  raw: unknown;
};

const staged = new Map<string, StagedRow>();
const conflicted = new Set<string>();

for (const { profile, raw } of valid) {
  try {
    const path = profilePath(profile.companyUrl);
    const key = JSON.stringify([profile.dataset, path]);

    if (conflicted.has(key)) {
      rejected.push({ raw, reason: 'duplicate_observation_key' });
      continue;
    }

    const previous = staged.get(key);
    if (previous) {
      rejected.push(
        { raw: previous.raw, reason: 'duplicate_observation_key' },
        { raw, reason: 'duplicate_observation_key' },
      );
      staged.delete(key);
      conflicted.add(key);
      continue;
    }

    staged.set(key, {
      dataset: profile.dataset,
      profile_path: path,
      partner_role: profile.partnerRole ?? null,
      ...resolveDirection(profile, path),
      raw,
    });
  } catch (error) {
    rejected.push({ raw, reason: String(error) });
  }
}

const rows = [...staged.values()];
const missingRequested = input.startUrls
  .map(target => profilePath(target.url))
  .filter(path => !rows.some(row =>
    row.dataset === 'us-import' && row.profile_path === path
  ));

const report = {
  requested: input.startUrls,
  missingRequested,
  rejected,
  flags: rows
    .filter(row => row.direction === 'review')
    .map(({ raw, ...decision }) => decision),
};

const pool = new pg.Pool();

try {
  const db = await pool.connect();
  try {
    await db.query('BEGIN');
    await db.query(ddl);

    await db.query(`
      INSERT INTO trade_profile (dataset, profile_path, run_id, raw)
      SELECT dataset, profile_path, $2, raw
      FROM jsonb_to_recordset($1::jsonb)
        AS x(dataset text, profile_path text, raw jsonb)
      ON CONFLICT (dataset, profile_path)
      DO UPDATE SET run_id = EXCLUDED.run_id, raw = EXCLUDED.raw
    `, [JSON.stringify(rows), run.id]);

    await db.query(`
      INSERT INTO trade_profile_direction
        (dataset, profile_path, partner_role, direction, reason)
      SELECT dataset, profile_path, partner_role, direction, reason
      FROM jsonb_to_recordset($1::jsonb)
        AS x(dataset text, profile_path text, partner_role text,
             direction text, reason text)
      ON CONFLICT (dataset, profile_path)
      DO UPDATE SET partner_role = EXCLUDED.partner_role,
                    direction = EXCLUDED.direction,
                    reason = EXCLUDED.reason
    `, [JSON.stringify(rows)]);

    await db.query(`
      INSERT INTO trade_ingestion_run (run_id, status, report)
      VALUES ($1, $2, $3::jsonb)
      ON CONFLICT (run_id)
      DO UPDATE SET status = EXCLUDED.status, report = EXCLUDED.report
    `, [run.id, run.status, JSON.stringify(report)]);

    await db.query('COMMIT');
  } catch (error) {
    await db.query('ROLLBACK');
    throw error;
  } finally {
    db.release();
  }
} finally {
  await pool.end();
}

Rejecting every duplicate key within a batch is conservative. It avoids silently choosing between competing profile observations and prevents a batch upsert from updating the same target row repeatedly.

Serialize jobs that cover the same profile slots. Otherwise an older source run can finish ingesting after a newer one and replace its payload. The upsert prevents duplicates; it doesn't establish observation order.

A finished run can still leave gaps

The worked example is ImportYeti US Importer & Global Supplier Search. Its source can present a login wall after the public request allowance is exhausted.

With proxy rotation configured, the Actor rotates sessions. When the limit is hit, it retains collected records and ends with a status message instead of failing. A profile that can't load is skipped with a log line.

That behavior makes deletion-by-absence unsafe.

The batch above updates accepted observations and leaves absent profiles alone. It also records unmatched requested paths. Redirects may explain a mismatch, but they still need reconciliation.

Don't rebuild the relationship graph from whatever this dataset happens to contain. A missing observation isn't evidence that a commercial relationship ended.

Likewise, reciprocal profile observations should attach provenance to the same eventual supplier-to-buyer edge. They shouldn't create separate commercial relationships merely because different profiles supplied the evidence.

Retry database ingestion from the stored dataset before starting another source run. A fresh run incurs start and emitted-result charges again, plus platform usage.

Give the next on-call shift a queue

Monitor committed ingestion reports. Alert on rejected records, unfamiliar role values, role-context conflicts, unsupported contexts, and unmatched requested paths. Track source-run failures separately from database failures.

Treat absent partners as a coverage question rather than evidence of no trading activity. Omitted fields stay omitted when the source doesn't publish them.

Additive outer fields can wait for routine review. A changed identity field or reversed direction deserves immediate attention because it changes who your dashboard recommends contacting.

Before enabling edge materialization:

Review actual partnerRole values and populate the vocabulary.
Approve nested partner identity mappings without relying on names.
Preserve the supplying profile, dataset, and run as edge evidence.
Prevent overlapping writers from replacing newer observations.
Keep existing edges when coverage is incomplete.
Make review decisions block graph changes.

An upsert can prevent duplicate rows. Only an explicit direction contract can prevent a perfectly unique wrong relationship.

Your comment timeline is using the wrong clock

Nimbryxo-studio — Tue, 10 Nov 2026 10:06:05 +0000

Your release report puts the busiest discussion at the start of the song. Before you rewrite the chart, check which clock it reads.

SoundCloud comments separate posting time (createdAt) from playback position (timestampMs). A comment attached to the opening can have been posted long after release.

Example input for collecting reactions by their position inside a track:

const run_input = {
  mode: "comments",
  trackUrls: [{ url: "https://soundcloud.com/user/track" }],
  commentsSort: "track-timestamp",
  maxItems: 200,
};

That order serves a different question from “What are listeners saying recently?” Once collection reaches its row cap, changing the chart's sort can't recover comments outside the collected batch.

The annoyance looks cosmetic. The repair might require recollection.

Read the symptom before changing the collector

The Actor asks SoundCloud to order comments using commentsSort. Its choices are newest, oldest, and track-timestamp; the default is newest.

The source supports different views of the same discussion. Your dashboard needs to choose which question it answers.

Symptom	Distinguishing evidence	Easy misdiagnosis	Fix and cost
Release discussion appears near the song's opening	The chart reads `timestampMs` instead of `createdAt`	Listeners mainly discussed the release immediately	Change the projection locally
Historical discussion appears on collection day	The chart reads `scrapedAt`	A fresh burst of comments	Bucket by posting time locally
A recent-discussion report uses an unsuitable collection order	Persisted `commentsSort` is `oldest` or `track-timestamp`	Broken dates or shuffled rows	Recollect newest-first; another run incurs charges
Coverage stops at the configured ceiling	Successful run's raw dataset count equals `maxItems`	Proof that every available comment was collected	Flag possible truncation; increasing the cap can increase cost
Comments disappear from a chart	Required clock field is absent, or a row fails validation	A quiet audience	Report exclusions separately; never substitute a zero

These signals can coexist. Fixing the chart field doesn't establish that its underlying sample suits the report.

Keep the submitted order beside the finished rows

A dataset tells you what arrived. It doesn't prove which input selected those rows.

Persist the exact submitted input alongside the run ID and dataset ID. For a historical diagnosis, use that snapshot rather than today's scheduler configuration.

This code waits for collection and loads its dataset. The inline run_input serves as the submitted snapshot throughout the remaining cells.

import { ApifyClient } from "apify-client";
import { z } from "zod";

const client = new ApifyClient({
  token: process.env.APIFY_TOKEN,
});

const run = await client
  .actor("nimbryxo-studio/soundcloud-charts-search-and-artist-catalog")
  .call(run_input);

const { items } = await client
  .dataset(run.defaultDatasetId)
  .listItems();

const finishedSuccessfully = run.status === "SUCCEEDED";
const rawRowCount = items.length;

Keep rawRowCount before filtering or validation. The cap applies to dataset rows, so comparing it with a smaller chart-ready subset answers the wrong question.

A finished run can also have an unsuccessful status. Treat that as an operational issue before interpreting its contents as a completed collection.

Preserve missing values instead of inventing moments

Empty fields are omitted from records. Missing playback position therefore needs its own handling.

Replacing an absent timestampMs with zero manufactures a reaction at the opening. Replacing an absent createdAt with scrapedAt manufactures a posting date. Both repairs make the chart look healthier than its evidence.

This validator accepts missing clocks, checks supplied values, and keeps rejected comment rows separate.

const identifier = z.union([
  z.string().min(1),
  z.number().finite(),
]);

const instant = z.string().refine(
  value => Number.isFinite(Date.parse(value)),
  "Expected a parseable instant",
);

const commentSchema = z.object({
  recordType: z.literal("comment"),
  commentId: identifier,
  trackId: identifier,
  createdAt: instant.optional(),
  timestampMs: z.number().finite().nonnegative().optional(),
  scrapedAt: instant.optional(),
});

type Comment = z.infer<typeof commentSchema>;

const comments: Comment[] = [];
let nonCommentRows = 0;
let rejectedComments = 0;

for (const item of items) {
  if (
    typeof item !== "object" ||
    item === null ||
    !("recordType" in item) ||
    item.recordType !== "comment"
  ) {
    nonCommentRows++;
    continue;
  }

  const parsed = commentSchema.safeParse(item);
  if (parsed.success) comments.push(parsed.data);
  else rejectedComments++;
}

A validation rejection means the row doesn't meet this application's boundary requirements. It doesn't, by itself, identify an Actor defect.

The distinction matters during diagnosis: an absent clock is different from a supplied value your application can't parse. Keep both counts visible beside the chart.

Which clock does the chart actually read?

Rows can't reveal your dashboard's mapping. Pass the field its mapper actually uses.

The following detector assumes the report's purpose is recent discussion. Example configuration passes a playback field to expose that mismatch:

function diagnoseRecentDiscussion(
  timeField: "createdAt" | "timestampMs" | "scrapedAt",
) {
  return {
    unsuccessfulRun: !finishedSuccessfully,
    wrongChartAxis: timeField !== "createdAt",
    collectionTimeUsed: timeField === "scrapedAt",
    orderMismatch: run_input.commentsSort !== "newest",
    possibleCapSaturation:
      finishedSuccessfully && rawRowCount === run_input.maxItems,
    missingPostingTime: comments.filter(
      row => row.createdAt === undefined,
    ).length,
    missingPlaybackPosition: comments.filter(
      row => row.timestampMs === undefined,
    ).length,
    rejectedComments,
    nonCommentRows,
  };
}

const diagnosis = diagnoseRecentDiscussion("timestampMs");

orderMismatch identifies configuration unsuitable for this purpose. It doesn't establish malformed records or prove that recent comments are missing.

Likewise, equality with the cap is evidence of possible truncation. The public source might have had exactly that many available rows. The count alone can't settle it.

A lower count isn't a completeness certificate either. You've observed the returned batch, not independently established the size of the public discussion.

Build a projection for each question

Keep posting activity and musical position as separate projections. Each can be useful without pretending to answer the other's question.

comment
  createdAt   -> posting day       -> release-response timeline
  timestampMs -> playback bucket  -> moments within the track
  scrapedAt   -> collection time  -> ingestion monitoring

Group playback reactions by track. Pooling raw positions across recordings can make unrelated musical moments look equivalent.

Example projection settings use UTC posting days and a playback bucket width defined locally:

const playbackBucketMs = 1000;

function projectComments(field: "createdAt" | "timestampMs") {
  const buckets = new Map<string, number>();

  if (!finishedSuccessfully) return buckets;

  for (const row of comments) {
    const time = field === "createdAt"
      ? row.createdAt === undefined
        ? undefined
        : Date.parse(row.createdAt)
      : row.timestampMs;

    if (time === undefined) continue;

    const bucket = field === "createdAt"
      ? new Date(time).toISOString().slice(0, 10)
      : Math.floor(time / playbackBucketMs) * playbackBucketMs;

    const key = JSON.stringify([String(row.trackId), bucket]);
    buckets.set(key, (buckets.get(key) ?? 0) + 1);
  }

  return buckets;
}

const postingDays = projectComments("createdAt");
const playbackPositions = projectComments("timestampMs");

These maps describe the validated, collected sample. Neither map repairs its selection.

For release-response analysis, compare posting days with your own release calendar. Keep ingestion time separately so a late collection doesn't create a false surge on collection day.

For comparisons across tracks of different lengths, use relative playback position. That requires a separate duration lookup keyed by trackId; comment rows don't provide track duration.

Recollect only when the sample is wrong

If the mapping is wrong, rebuilding the projection is enough. If the collection order is wrong for recent discussion, sorting existing rows newest-first changes presentation without changing membership.

The SoundCloud Charts, Search & Artist Catalog Scraper is the worked example here. The reusable failure is a bounded collection whose ordering doesn't match the downstream question.

This corrected input changes selection order while retaining the original scope and ceiling:

const correctedInput = {
  ...run_input,
  commentsSort: "newest",
};

Submit it through the same collection path when the diagnosis warrants another run. Keeping that submission separate from diagnosis helps avoid paid retries whenever someone opens the report.

The ceiling remains a trade-off. maxItems supports values from 1 to 5000, and raising it can collect more rows. It still can't establish that a release window is fully covered.

Recollection incurs event fees plus platform usage. On the FREE tier, emitted results cost $0.00099 each. The start event costs $0.10 per event, with one event per GB of memory and a minimum of one event.

There's another analytical catch: newest-first serves recent-discussion monitoring, but it can't guarantee recovery of an older release window after collection has been delayed. Store successive observations if that history matters, and deduplicate using your comment identity policy.

For ingestion, retain the earliest collection time separately from any later refresh time. A returning comment is still the same historical event.

Before you publish the release report

Confirm the dashboard reads the clock its question requires.
Inspect the submitted input snapshot.
Check run status and raw row count before filtering.
Keep missing-clock counts separate from validation rejections.
Rebuild projections locally when selection remains suitable.
Recollect deliberately when selection must change.

A tidy timeline can still be an answer to the wrong question.

Your failed email runs can keep charging start fees

Nimbryxo-studio — Mon, 09 Nov 2026 10:04:53 +0000

Your broken SMTP password can keep generating start fees all night.

You rotate a credential. Notifications stop. The scheduler keeps trying, while an empty dataset makes the spending dashboard look reassuring.

A failed email run can incur a start charge without producing a result record. Validation and sending failures leave the reason in the run's status message. The dataset stays empty because a confirmation is written only after the SMTP server accepts the message.

Your budget needs to count authorized attempts before execution. Counting confirmation rows afterward misses the failure you most need to contain.

What does an empty dataset actually save?

Starting execution and writing a result are separate charging events.

The start event is charged when execution begins. Its quantity depends on allocated memory: one event per GB, with a minimum of one event. The result event applies to a record in the default dataset.

Platform usage is additional, including usage consumed by a failed run.

For this worked example, the Actor sends a single email per run. Multiple recipients still belong to that message.

What you change	Budget consequence
Scheduled starts, retries, manual starts	Each execution creates another start-charge opportunity
Allocated memory	Changes the number of start events
`to`, `cc`, or `bcc` recipient lists	Does not create a separate confirmation per recipient
Body or attachment content	Can affect SMTP acceptance, without creating a separate published content event

Put every launch behind the same allowance. First, remove failures you can catch without starting anything.

Reject broken inputs before paying to discover them

The schema has 13 properties, with 5 required fields. The body requirement crosses fields, so checking required keys alone misses it.

This local policy also rejects blank required strings. Attachment entries receive a string-value check here; attachment contents and SMTP acceptance need separate checks.

Example input and validation for a scheduled notification:

import { z } from 'zod';
import { ApifyClient } from 'apify-client';

const nonblank = z.string().refine(value => value.trim().length > 0);

const inputSchema = z.object({
  to: nonblank,
  cc: z.string().optional(),
  bcc: z.string().optional(),
  subject: nonblank,
  text: z.string().optional(),
  html: z.string().optional(),
  replyTo: z.string().optional(),
  attachments: z.array(
    z.record(z.string(), z.string()),
  ).optional(),
  smtpHost: nonblank,
  smtpPort: z.number().int().min(1).max(65535).default(587),
  smtpUser: nonblank,
  smtpPassword: nonblank,
  useTls: z.boolean().default(true),
}).strict().refine(
  value => Boolean(value.text?.trim() || value.html?.trim()),
  { message: 'Provide text or html', path: ['text'] },
);

const run_input = inputSchema.parse({
  to: 'oncall@example.com',
  subject: 'Scheduled export ready',
  text: 'The export is ready for review.',
  smtpHost: 'smtp.example.com',
  smtpUser: 'notifications@example.com',
  smtpPassword: 'example-password-replace-before-use',
});

Keep credentials out of validation logs. Record failing field paths and your validation message, rather than dumping the rejected object.

Authentication failures, connection refusals, and timeouts still require execution to discover. Those belong behind the attempt budget.

Calculate a conditional event ceiling

Use the published event prices directly. Keep currency arithmetic in integer microdollars so rounding never decides whether another attempt fits.

There's a boundary you can't fill with optimism: the pricing describes memory scaling but doesn't specify allocation rounding or an upper allocation limit. A defensible ceiling therefore needs a verified start-event bound for the memory configuration you'll actually launch.

The function below refuses approval until that bound is supplied. It allows at most one result event per attempt, matching this Actor's confirmation behavior.

Example budget policy, with the memory-dependent bound deliberately unset:

const prices = {
  FREE:     { start: 100_000n, result: 990n },
  BRONZE:   { start: 77_000n,  result: 820n },
  SILVER:   { start: 53_000n,  result: 660n },
  GOLD:     { start: 30_000n,  result: 490n },
  PLATINUM: { start: 30_000n,  result: 490n },
  DIAMOND:  { start: 30_000n,  result: 490n },
} as const;

function positiveInteger(value: number): number {
  if (!Number.isSafeInteger(value) || value < 1) {
    throw new Error('Expected a positive safe integer');
  }
  return value;
}

function eventCeiling(
  input: unknown,
  tier: keyof typeof prices,
  authorizedAttempts: number,
  verifiedStartEventBound: number | null,
) {
  const parsedInput = inputSchema.parse(input);
  const attempts = positiveInteger(authorizedAttempts);

  if (verifiedStartEventBound === null) return null;

  const starts = positiveInteger(verifiedStartEventBound);
  const rate = prices[tier];
  const perAttemptMicros =
    BigInt(starts) * rate.start + rate.result;

  return {
    input: parsedInput,
    authorizedAttempts: attempts,
    perAttemptMicros,
    totalMicros: BigInt(attempts) * perAttemptMicros,
  };
}

function usd(micros: bigint): string {
  return `${micros / 1_000_000n}.${
    (micros % 1_000_000n).toString().padStart(6, '0')
  }`;
}

const policy = {
  occurrences: 30,
  retriesPerOccurrence: 2,
  recoveryProbes: 1,
  maxAttemptsPerNotification: 3,
  failureThreshold: 2,
  verifiedStartEventBound: null as number | null,
};

const authorizedAttempts =
  positiveInteger(policy.occurrences) *
  (1 + policy.retriesPerOccurrence) +
  policy.recoveryProbes;

const plan = eventCeiling(
  run_input,
  'FREE',
  authorizedAttempts,
  policy.verifiedStartEventBound,
);

type ApprovedPlan = NonNullable<typeof plan>;

This is an event-fee ceiling, conditional on the selected tier, the verified memory bound, and the Actor's result behavior. Keep platform usage in a separate budget.

Pin the memory configuration associated with that bound. Recheck the plan when the tier, pricing, or configuration changes.

Make every launch spend from the same allowance

A scheduler retry limit does little if someone can bypass it with a manual trigger.

Use a stable notification key for the logical email. Scheduled retries and manual attempts must claim against that same key. Give the SMTP configuration its own circuit key, so a persistent configuration failure can suppress other notifications using it.

Implement the following interface in your persistent job store. Its reservation operation must atomically check the attempt allowance, the period budget, and the circuit before reserving money.

Example integration contract for your existing job ledger:

type ReservationRequest = {
  notificationKey: string;
  circuitKey: string;
  periodKey: string;
  mode: 'regular' | 'probe';
  maxAttemptsPerNotification: number;
  maxAttemptsInPeriod: number;
  failureThreshold: number;
  reserveMicros: bigint;
  periodCeilingMicros: bigint;
};

type AttemptReceipt = {
  runId: string | null;
  outcome: 'confirmed' | 'failed' | 'review';
  runStatus: string | null;
  statusMessage: string | null;
  resultRecords: number | null;
};

interface BudgetLedger {
  reserve(request: ReservationRequest): Promise<string | null>;
  finish(reservationId: string, receipt: AttemptReceipt): Promise<void>;
}

Use a transaction or equivalent atomic operation. A read followed by an independent update lets overlapping workers spend the same balance.

Count a reservation against both money and attempt limits immediately. If the worker disappears, leave it pending for reconciliation. Automatically refunding an uncertain launch makes a network timeout a rather expensive loophole.

The circuit state should survive budget-period changes. Midnight doesn't repair a password.

Put the cap on the start request

The preflight gate controls whether another run may start. The platform-side cap controls what an individual pay-per-event run may charge.

Use both. A run cap can't enforce the shared allowance across separate executions, and a local estimate can't stop an already running execution.

Example launcher using maxTotalChargeUsd on the REST start endpoint:

const confirmationSchema = z.object({
  status: z.string(),
  sentAt: z.string(),
  messageId: z.string(),
});

async function launchNotification(
  approved: ApprovedPlan,
  ledger: BudgetLedger,
  scope: {
    notificationKey: string;
    circuitKey: string;
    periodKey: string;
    mode: 'regular' | 'probe';
  },
) {
  const token = process.env.APIFY_TOKEN;
  if (!token) throw new Error('APIFY_TOKEN is required');

  const reservationId = await ledger.reserve({
    ...scope,
    maxAttemptsPerNotification: policy.maxAttemptsPerNotification,
    maxAttemptsInPeriod: approved.authorizedAttempts,
    failureThreshold: policy.failureThreshold,
    reserveMicros: approved.perAttemptMicros,
    periodCeilingMicros: approved.totalMicros,
  });

  if (reservationId === null) return;

  const receipt: AttemptReceipt = {
    runId: null,
    outcome: 'review',
    runStatus: null,
    statusMessage: null,
    resultRecords: null,
  };

  try {
    const url = new URL(
      'https://api.apify.com/v2/acts/' +
      'nimbryxo-studio~smtp-email-sender-for-workflow-notifications/runs',
    );
    url.searchParams.set(
      'maxTotalChargeUsd',
      usd(approved.perAttemptMicros),
    );

    const response = await fetch(url, {
      method: 'POST',
      headers: {
        Authorization: `Bearer ${token}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify(approved.input),
    });

    if (!response.ok) {
      throw new Error(`Start request returned HTTP ${response.status}`);
    }

    const started = z.object({
      data: z.object({ id: z.string() }),
    }).parse(await response.json());

    receipt.runId = started.data.id;

    const client = new ApifyClient({ token });
    const finished = await client.run(started.data.id).waitForFinish();
    if (!finished) return;

    receipt.runStatus = finished.status;
    receipt.statusMessage = finished.statusMessage ?? null;

    const { items } = await client
      .dataset(finished.defaultDatasetId)
      .listItems();

    receipt.resultRecords = items.length;
    const records = z.array(confirmationSchema).max(1).safeParse(items);

    if (finished.status === 'SUCCEEDED' &&
        records.success &&
        records.data.length === 1 &&
        records.data[0].status === 'sent') {
      receipt.outcome = 'confirmed';
    } else if (finished.status === 'FAILED') {
      receipt.outcome = 'failed';
    }
  } finally {
    await ledger.finish(reservationId, receipt);
  }
}

The launcher is defined without dispatching the illustrative email. Invoke it from your worker only after approving the memory bound and supplying the persistent ledger.

Keep status messages in restricted diagnostic storage. They help repair the connection; they shouldn't become a broadly distributed log of notification inputs.

Reconcile each reservation with charged start and result events afterward. Dataset length can't establish the start-event quantity. Retain uncertain reservations until billing and run state resolve them.

Treat a capped dataset as an unfinished receipt

A cap can leave a run with incomplete output. Downstream ingestion should retain the run's completion state alongside its records, rather than marking a job complete merely because a dataset request succeeded.

For SMTP Email Sender for Workflow Notifications, that distinction is particularly awkward: the output is a confirmation of an external side effect.

SMTP acceptance also doesn't prove inbox delivery. Keep those meanings separate in your application.

The decision belongs after inspecting status, status message, and records together:

reserve allowance
       |
    capped start
       |
       +-- succeeded + valid confirmation --> record SMTP acceptance
       |
       +-- failed ------------------------> retain reason; update circuit
       |
       +-- interrupted or inconsistent ---> hold reservation; review

Don't treat every empty dataset as permission to resend. Sending an email and persisting its confirmation cross system boundaries; an interrupted execution needs reconciliation.

The current messageId is empty because the Actor doesn't assign one before sending. It can't serve as your deduplication key. Keep your notification identity in the job ledger, and avoid claiming exactly-once delivery from this interface.

Open the circuit, then leave a recovery path

Configuration failures usually need a repair. Transient connection failures may justify bounded retries.

Preserve the server's error text for diagnosis, but resist building a supposedly reliable taxonomy from undocumented message fragments. Treat unknown failures conservatively, with an operator-controlled retry decision.

Your ledger's finish operation should update the failure streak atomically. Once the configured threshold is reached, regular reservations stop. Ambiguous attempts should also block automatic resends pending review.

A recovery probe needs explicit authorization in the ledger. Admit it atomically, charge it against the same period budget, and prevent concurrent probes. Close the circuit after a confirmed result; keep it open after another failure.

Changing credentials alone shouldn't silently restart the backlog.

Before the next scheduled attempt

Budget the recurring schedule before enabling it. The policy above includes normal occurrences, their permitted retries, and a recovery allowance. Those are authorized starts, even when you expect most retries to remain unused.

Reset financial allowances at the intended accounting boundary. Keep pending reservations and unresolved notification identities intact across that boundary.

Before restoring the schedule:

Validate the complete input locally.
Verify the start-event bound against the pinned memory configuration.
Route scheduled and manual launches through the shared ledger.
Apply the event cap to every start request.
Preserve terminal status, diagnostic text, and confirmation state together.
Reconcile event charges separately from platform usage.
Admit a controlled recovery probe before releasing queued notifications.

An empty dataset is a poor spending report. Your launch ledger should know what happened before the inbox does.

Your inactive listing survived a successful upsert

Nimbryxo-studio — Sun, 08 Nov 2026 11:02:46 +0000

Your outreach queue still contains a listing that the latest exact-ID lookup marked inactive. The ingestion job finished. Your database is perfectly available. Somehow, that makes it more annoying.

The quiet failure can sit inside your validator: it requires title, rejects the sparse status record, and leaves the previous listing eligible for outreach.

An inactive tombstone must update outreach eligibility without replacing the last listing snapshot. A successful explicit lookup can produce that tombstone when the requested ID is absent. A failed lookup can't establish inactivity.

Keep the lookup narrower than the search

A listing disappearing from a filtered search proves very little. Its price might have changed. It might no longer match your filters.

An explicit lookup gives you a narrower observation: whether that lookup returned the requested advertisement. That distinction belongs in your database design before you schedule anything.

Example input for checking known outreach targets, with imports shared by the remaining code:

import { ApifyClient } from 'apify-client';
import { z } from 'zod';
import pg from 'pg';

const input = {
  mode: 'byListingIds',
  listingIds: ['2501583', '2571748'],
  maxItems: 2,
  enrichDetails: false,
  mirrorImages: false,
};

const client = new ApifyClient({
  token: process.env.APIFY_TOKEN,
});

Keep these checks separate from discovery. You already have identities; fetching richer descriptions and copying images adds work without answering the lifecycle question.

Leave stateKey unset here. This workflow wants each lookup's observations, rather than a stream filtered down to previously unseen or changed records.

Wait for the run you already started

Starting another Actor run is different from retrying a database write. The former performs another lookup and incurs another run's charges. The latter can reuse an existing dataset.

Start the run, then wait on its identity:

const started = await client
  .actor('nimbryxo-studio/bizbuysell-businesses-for-sale-and-financials')
  .start(input);

const finished = await client.run(started.id).waitForFinish();

if (!finished || finished.status !== 'SUCCEEDED') {
  throw new Error(
    `Lookup run ${started.id} unresolved: ${finished?.status ?? 'pending'}`,
  );
}

const datasetId = finished.defaultDatasetId;

Persist started.id in your scheduler before waiting. If your worker disappears, resume waiting on that run instead of starting another.

This recipe processes only successful runs. A failed run might have saved useful items, but accepting those requires a separate partial-run policy. Whatever policy you choose, missing output never authorizes an inactive transition.

Accept the status row without inventing a snapshot

Tombstones preserve identity, status, provenance, and observation timestamps. They don't manufacture a headline or asking price.

Your consumer therefore needs separate paths:

returned listing ──► refresh snapshot + lifecycle
inactive tombstone ► update lifecycle, retain snapshot
invalid record ────► quarantine, retain current state
unresolved lookup ─► alert, retain current state

Use a union that accepts the sparse shape while validating the fields your workflow consumes:

const timestamp = z.string().datetime({ offset: true });

const common = z.object({
  productId: z.union([
    z.string().min(1),
    z.number().int().nonnegative().safe(),
  ]).transform(String),
  scrapedAt: timestamp,
  searchUrl: z.string().min(1),
  truncated: z.boolean().optional(),
});

const listing = common.extend({
  title: z.string().min(1),
  askingPriceNumeric: z.union([
    z.number().finite(),
    z.string(),
  ]).optional(),
  listingStatus: z.never().optional(),
}).passthrough();

const tombstone = common.extend({
  listingStatus: z.literal('inactive'),
  listingStatusSource: z.literal('listing-id-lookup'),
  listingStatusEvidence: z.unknown().optional(),
  statusCheckedAt: timestamp,
}).passthrough();

const observation = z.union([tombstone, listing]);

Requiring a title on ordinary listings is a consumer policy: those are outreach candidates. The tombstone branch has no such requirement. Neither branch requires a price.

The string alternative handles the supplied sample's textual askingPriceNumeric. It isn't permission to guess a monetary value. Below, strings remain in raw JSON and generate a representation warning; only finite numbers enter the numeric column.

.passthrough() retains additive fields without making every new field a paging event. A strict object would reject harmless additions. Here, drift means a consumed field breaks its contract, or a financial representation needs attention.

Give the database separate clocks

Your idempotency key is (source, product_id). It identifies a marketplace advertisement, which is what the explicit lookup checks.

It doesn't resolve the underlying business across relistings. Merging those requires other evidence and a separate decision. A matching headline is an invitation to investigate, not a primary key.

Create a current-state table, a replayable issue store, and the relation your outreach worker reads:

const pool = new pg.Pool();

await pool.query(`
  CREATE TABLE IF NOT EXISTS listing_state (
    source text NOT NULL,
    product_id text NOT NULL,
    title text,
    asking_price_numeric numeric,
    snapshot jsonb,
    snapshot_at timestamptz,
    eligible boolean NOT NULL DEFAULT false,
    lifecycle_at timestamptz,
    last_observation jsonb,
    PRIMARY KEY (source, product_id)
  );

  CREATE TABLE IF NOT EXISTS ingestion_issues (
    dataset_id text NOT NULL,
    item_offset bigint NOT NULL,
    kind text NOT NULL,
    payload jsonb NOT NULL,
    PRIMARY KEY (dataset_id, item_offset, kind)
  );

  CREATE OR REPLACE VIEW outreach_candidates AS
    SELECT source, product_id, title, asking_price_numeric
    FROM listing_state
    WHERE eligible AND snapshot IS NOT NULL;
`);

Typed columns make eligibility and routine filtering straightforward. JSONB preserves the accepted observation and the last useful listing for inspection.

The separate clocks matter. A newer tombstone can retire a target while an older, previously unprocessed listing still supplies its historical snapshot.

Observation timestamps prevent an older retry from reversing a newer decision. They aren't marketplace publication dates, and they can't correct a bad source observation.

Upsert state without erasing the last listing

Write snapshots and lifecycle state separately, inside the same transaction.

Each statement deduplicates its incoming identities before ON CONFLICT. Otherwise, repeated IDs within a batch can make the statement fail instead of quietly behaving like sequential updates.

This function writes the whole batch:

const incomingSql = `
  WITH incoming AS (
    SELECT *
    FROM jsonb_to_recordset($1::jsonb) AS x(
      product_id text,
      observed_at timestamptz,
      inactive boolean,
      raw jsonb,
      title text,
      price numeric
    )
  )
`;

async function applyBatch(
  db: Awaited<ReturnType<typeof pool.connect>>,
  rows: Record<string, unknown>[],
) {
  const values = [JSON.stringify(rows)];

  await db.query(incomingSql + `
    INSERT INTO listing_state AS current (
      source, product_id, title, asking_price_numeric,
      snapshot, snapshot_at
    )
    SELECT DISTINCT ON (product_id)
      'bizbuysell', product_id, title, price, raw, observed_at
    FROM incoming
    WHERE NOT inactive
    ORDER BY product_id, observed_at DESC
    ON CONFLICT (source, product_id) DO UPDATE SET
      title = EXCLUDED.title,
      asking_price_numeric = EXCLUDED.asking_price_numeric,
      snapshot = EXCLUDED.snapshot,
      snapshot_at = EXCLUDED.snapshot_at
    WHERE current.snapshot_at IS NULL
       OR EXCLUDED.snapshot_at > current.snapshot_at
  `, values);

  await db.query(incomingSql + `
    INSERT INTO listing_state AS current (
      source, product_id, eligible, lifecycle_at, last_observation
    )
    SELECT DISTINCT ON (product_id)
      'bizbuysell', product_id, NOT inactive, observed_at, raw
    FROM incoming
    ORDER BY product_id, observed_at DESC, inactive DESC
    ON CONFLICT (source, product_id) DO UPDATE SET
      eligible = EXCLUDED.eligible,
      lifecycle_at = EXCLUDED.lifecycle_at,
      last_observation = EXCLUDED.last_observation
    WHERE current.lifecycle_at IS NULL
       OR EXCLUDED.lifecycle_at > current.lifecycle_at
       OR (
         EXCLUDED.lifecycle_at = current.lifecycle_at
         AND NOT EXCLUDED.eligible
       )
  `, values);
}

The snapshot statement never sees tombstones. The lifecycle statement never replaces financial or descriptive fields.

Equal lifecycle timestamps favor inactivity. That's a conservative outreach policy, rather than a claim about source ordering. Equal snapshot timestamps preserve the existing snapshot.

A later ordinary listing can restore eligibility. Whether your business permits automatic reactivation is another policy boundary; add manual approval there if outreach requires it.

Make retries boring and failures replayable

A page is the transaction boundary. Valid records and their accompanying issue records commit together.

If a later page fails, committed pages remain usable. Reprocess the same dataset from the beginning: keys prevent duplicate listings, and timestamps prevent stale lifecycle writes.

Page through the dataset, quarantine invalid items, and retain unresolved requested IDs separately:

const issueSql = `
  INSERT INTO ingestion_issues (dataset_id, item_offset, kind, payload)
  SELECT $1, item_offset, kind, payload
  FROM jsonb_to_recordset($2::jsonb) AS x(
    item_offset bigint, kind text, payload jsonb
  )
  ON CONFLICT (dataset_id, item_offset, kind) DO NOTHING
`;

const seen = new Set<string>();
const pageSize = input.maxItems;
let offset = 0;

try {
  while (true) {
    const { items } = await client.dataset(datasetId).listItems({
      offset,
      limit: pageSize,
    });
    if (items.length === 0) break;

    const accepted: Record<string, unknown>[] = [];
    const issues: {
      item_offset: number;
      kind: string;
      payload: unknown;
    }[] = [];

    items.forEach((raw, index) => {
      const parsed = observation.safeParse(raw);
      const itemOffset = offset + index;

      if (!parsed.success) {
        issues.push({
          item_offset: itemOffset,
          kind: 'validation',
          payload: { raw, errors: parsed.error.issues },
        });
        return;
      }

      const row = parsed.data;
      const inactive = row.listingStatus === 'inactive';
      seen.add(row.productId);

      accepted.push({
        product_id: row.productId,
        observed_at: row.listingStatus === 'inactive'
          ? row.statusCheckedAt
          : row.scrapedAt,
        inactive,
        raw,
        title: typeof row.title === 'string' ? row.title : null,
        price: typeof row.askingPriceNumeric === 'number'
          ? row.askingPriceNumeric
          : null,
      });

      if (!inactive && typeof row.askingPriceNumeric === 'string') {
        issues.push({
          item_offset: itemOffset,
          kind: 'shape_drift',
          payload: { field: 'askingPriceNumeric', raw },
        });
      }

      if (row.truncated === true) {
        issues.push({
          item_offset: itemOffset,
          kind: 'incomplete_coverage',
          payload: { raw },
        });
      }
    });

    const db = await pool.connect();
    try {
      await db.query('BEGIN');
      await applyBatch(db, accepted);
      await db.query(issueSql, [datasetId, JSON.stringify(issues)]);
      await db.query('COMMIT');
    } catch (error) {
      await db.query('ROLLBACK');
      throw error;
    } finally {
      db.release();
    }

    offset += items.length;
  }

  const unresolved = input.listingIds.filter(id => !seen.has(id));
  if (unresolved.length > 0) {
    await pool.query(issueSql, [
      datasetId,
      JSON.stringify([{
        item_offset: -1,
        kind: 'unresolved_lookup',
        payload: { listingIds: unresolved, runId: started.id },
      }]),
    ]);
  }
} finally {
  await pool.end();
}

Quarantine retains the original item, its dataset position, and validation errors. You can replay it after correcting the consumer without paying for another lookup.

An unresolved requested ID is an operational signal, not a negative listing observation. It can reflect lookup trouble, incomplete coverage, or rejected records. Inspect the run and quarantine before assigning a cause.

What can still leave a target unresolved?

The worked source is BizBuySell Businesses-for-Sale & Financials Scraper. Its inactive marker represents observed absence from a successful explicit lookup. It doesn't distinguish sold from expired.

That limitation should survive every downstream transformation, including the labels your sales team sees.

Signal	Database action	Operational response
Accepted inactive tombstone	Disable eligibility; retain snapshot	Count the transition as normal processing
Validation failure	Preserve current state; quarantine raw item	Alert on broken consumer contracts
Unresolved requested ID	Preserve current state	Investigate lookup coverage
Text in the numeric price field	Retain raw value; leave typed price empty	Review representation drift
Unsuccessful run	Skip this ingestion path	Alert through the scheduler

Detail enrichment is also best-effort. An unavailable detail page can leave a listing with search fields and an enrichment failure status. It doesn't establish listing inactivity.

Alert on unresolved checks persisting beyond your freshness policy, rather than treating every inactive transition as an incident. Also monitor job failures and whether the issue store is actually being reviewed. A quarantine nobody reads is an unusually durable trash can.

Before you leave this on call

Persist the run identity before waiting, and retain its dataset identity for replay.
Keep snapshot and lifecycle timestamps independent.
Route invalid records to a replayable queue.
Treat missing output as unresolved.
Have the outreach worker recheck eligibility when claiming queued work.

Updating a database flag won't recall a message already sent. The final eligibility check belongs where outreach becomes action.

A yearly salary filter changes who enters your benchmark

Nimbryxo-studio — Sat, 07 Nov 2026 11:00:53 +0000

Your salary benchmark can change before you calculate a single average.

You're explaining a dashboard refresh after someone added minSalaryYearly. The query still finishes. The chart still renders. Now you need to establish whether advertised pay changed or different postings entered the calculation.

The Actor's onlyWithSalary accepts disclosed amounts across pay periods; minSalaryYearly requires a disclosed yearly minimum and compares it in each job's own currency.

That changes membership and leaves currency comparability unresolved. A successful ingestion job can deliver an unsuitable comparison group. Quietly, naturally.

Find the gate that moved

Start with observable evidence. A smaller dataset alone can't identify the cause.

Symptom	Distinguishing evidence	Easy misdiagnosis	Fix and its cost
Disclosed postings disappear after adding a yearly floor	Broader saved items lack `salaryYearlyMinimum`, or their minimum falls below the floor	Employers reduced pay	Separate missing normalization from below-floor amounts; recovering excluded records requires broader collection
A salary record cannot enter a currency group	`salaryCurrency` is absent or blank	The amount must use the office country's currency	Hold it for review; comparison coverage shrinks
A combined band mixes monetary units	Distinct nonempty `salaryCurrency` values appear	Annualization made the amounts comparable	Group by currency; any conversion needs a separate policy
Raw amounts use different time units	Distinct `salaryPeriod` values appear	Every `salaryMin` is annual	Use supplied yearly minimums for annual comparisons; unresolved records remain outside
Returned items reach the collection limit	Item count reaches `maxItems`	Salary eligibility explains the entire count	Investigate coverage separately; broader collection adds cost

Collect a disclosure cohort before applying your analytical eligibility rules.

Example diagnostic input keeps the yearly floor disabled and bounds collection:

import { ApifyClient } from 'apify-client';

const client = new ApifyClient({ token: process.env.APIFY_TOKEN });
const input = {
  mode: 'search',
  searchQuery: 'software engineer',
  onlyWithSalary: true,
  minSalaryYearly: 0,
  maxItems: 50,
};

const run = await client
  .actor('nimbryxo-studio/welcome-to-the-jungle-job-listings-and-salary')
  .call(input);

if (run.status !== 'SUCCEEDED') {
  throw new Error(`Resolve run status before salary diagnosis: ${run.status}`);
}

const { items } = await client.dataset(run.defaultDatasetId).listItems();
const atCollectionLimit = items.length >= input.maxItems;

The collection limit makes this a bounded diagnostic pull. Reaching it signals unresolved coverage, not proof that additional matching jobs exist.

For an incident, prefer the broader snapshot you already saved. Each run reads current live listings, so a recovery pull can't reconstruct the earlier population exactly.

Keep omitted fields visible

The Actor leaves unpublished values out. A disclosed amount can therefore arrive without a currency or normalized yearly minimum.

That omission matters analytically. Replacing missing normalization with zero would manufacture a low salary. Filling currency from office location would manufacture a monetary unit.

Validate the salary fields you inspect while preserving the rest of each record:

import { z } from 'zod';

const jobSchema = z.object({
  jobId: z.string().min(1),
  salary: z.string().optional(),
  salaryMin: z.number().finite().optional(),
  salaryMax: z.number().finite().optional(),
  salaryPeriod: z.string().optional(),
  salaryCurrency: z.string().optional(),
  salaryYearlyMinimum: z.number().finite().optional(),
  isSalaryDisclosed: z.boolean().optional(),
}).passthrough();

type Job = z.infer<typeof jobSchema>;

const jobs: Job[] = [];
const validationReview: { index: number; raw: unknown }[] = [];

items.forEach((raw, index) => {
  const parsed = jobSchema.safeParse(raw);
  if (parsed.success) jobs.push(parsed.data);
  else validationReview.push({ index, raw });
});

A validation rejection means the record failed this ingestion contract. It doesn't establish why. Keep the raw item for inspection rather than counting it as an undisclosed salary.

The passthrough records also retain source context. You'll want that context when someone asks why a posting missed the benchmark.

Which rows lost yearly eligibility?

salary represents the source's pay period. salaryYearlyMinimum supplies the yearly comparison value when available.

The Actor never estimates pay.

That explains why a monthly amount can satisfy disclosure eligibility while failing yearly eligibility. Even a yearly amount can lack the normalized field. Parsing the readable salary string would introduce your own interpretation.

Example proposed dashboard input adds a floor; the detector separates missing yearly values from amounts below it:

const dashboardInput = { ...input, minSalaryYearly: 50000 };

const hasUsableYearlyMinimum = (
  job: Job,
): job is Job & { salaryYearlyMinimum: number } =>
  typeof job.salaryYearlyMinimum === 'number'
  && Number.isFinite(job.salaryYearlyMinimum)
  && job.salaryYearlyMinimum >= 0;

const yearlyEligible = jobs.filter(hasUsableYearlyMinimum);
const missingYearly = jobs.filter(
  job => job.salaryYearlyMinimum === undefined,
);
const invalidYearly = jobs.filter(
  job => job.salaryYearlyMinimum !== undefined
    && !hasUsableYearlyMinimum(job),
);
const belowFloor = yearlyEligible.filter(
  job => job.salaryYearlyMinimum < dashboardInput.minSalaryYearly,
);

These buckets answer different operational questions.

missingYearly identifies a normalization gap. belowFloor identifies an amount comparison, still expressed in each posting's currency. invalidYearly flags values outside the analytical rule chosen here.

They describe the collected cohort only. If you first collect with the stricter floor, excluded records can't explain themselves afterward.

For hourly or daily amounts, annualization requires assumptions about paid hours or days. Keep those assumptions in a separate transformation, with their own version. They shouldn't arrive disguised as source values.

Separate currencies before computing bands

Annualization changes the time unit. Currency conversion changes the monetary unit.

Neither operation performs the other.

A numeric floor applied across currencies therefore doesn't represent a common purchasing threshold. Grouping surviving records afterward prevents mixed-currency arithmetic, but can't undo the selection already performed.

Build currency groups from usable yearly records, and inspect original pay periods independently:

const currencyOf = (job: Job) => job.salaryCurrency?.trim() || undefined;
const periodOf = (job: Job) => job.salaryPeriod?.trim() || undefined;

const currencies = new Set(
  jobs.flatMap(job => currencyOf(job) ? [currencyOf(job)!] : []),
);
const periods = new Set(
  jobs.flatMap(job => periodOf(job) ? [periodOf(job)!] : []),
);

const missingCurrency = jobs.filter(job => !currencyOf(job));
const missingPeriod = jobs.filter(job => !periodOf(job));
const comparisonGroups = new Map<string, Job[]>();

for (const job of yearlyEligible) {
  const currency = currencyOf(job);
  if (!currency) continue;
  const group = comparisonGroups.get(currency) ?? [];
  group.push(job);
  comparisonGroups.set(currency, group);
}

const mixedCurrencies = currencies.size > 1;
const mixedPeriods = periods.size > 1;

Mixed original periods are a warning against averaging raw amounts. They don't automatically disqualify records whose supplied yearly minimum and currency are usable.

Missing currency does disqualify a record from these currency groups. An amount without a monetary unit can't support a defensible cross-country comparison.

The admission path should remain visible:

disclosed amount
  -> usable source yearly minimum?
       absent: normalization review
       present -> currency available?
                    absent: currency review
                    present: yearly comparison group by currency

Country, role and contract restrictions still belong in your benchmark definition. Currency groups solve unit compatibility; they don't make unrelated jobs comparable.

Treat input changes as benchmark changes

A dashboard refresh should carry its selection policy alongside its results.

For the Welcome to the Jungle Job Listings & Salary Scraper, keep the saved Actor input separate from your downstream comparison rule. That lets you identify which layer changed admission.

Capture the proposed salary-policy change together with the diagnostic counts:

const salaryEligibilityChanged =
  input.onlyWithSalary !== dashboardInput.onlyWithSalary
  || input.minSalaryYearly !== dashboardInput.minSalaryYearly;

const audit = {
  collectedInput: { ...input },
  proposedInput: { ...dashboardInput },
  comparisonRule: 'source-yearly-minimum-grouped-by-currency',
  salaryEligibilityChanged,
  atCollectionLimit,
  counts: {
    collected: items.length,
    validationReview: validationReview.length,
    missingYearly: missingYearly.length,
    invalidYearly: invalidYearly.length,
    belowProposedFloor: belowFloor.length,
    missingCurrency: missingCurrency.length,
    missingPeriod: missingPeriod.length,
  },
  mixedCurrencies,
  mixedPeriods,
};

These counts overlap. A posting can lack both currency and normalization, so adding the review counts would exaggerate excluded membership.

Persist this evidence with the source snapshot. Compare it with the previous refresh before attributing chart movement to compensation changes.

The detector covers the salary settings under investigation. Changes to search scope, geography, contract filters or collection limits also need review when deciding whether snapshots remain comparable.

What the surviving rows can't tell you

A finished, filtered dataset shows accepted records. It can't reveal the exact composition of the rejected population.

Without a broader saved snapshot, you can't determine how many missing postings lacked normalization versus fell below the floor. A recovery run adds event charges and platform usage, and live listings may have changed.

Broader collection moves filtering downstream. That improves diagnosis because you retain records that fail your comparison gate, but you pay to collect those records too.

Cross-currency conversion adds another policy surface: rate source, effective date and rounding. Keep that transformation outside the Actor's behavior and preserve the original amounts.

Also name the measure accurately. These are advertised salary bands. They don't establish accepted offers or actual employee earnings.

Before publishing the next refresh

Confirm the run completed successfully and inspect validation failures.
Check collection-limit warnings before explaining count changes.
Separate missing yearly values from below-floor yearly values.
Hold missing currencies for review and calculate within explicit currency groups.
Save the input, comparison policy and source snapshot together.
Label eligibility changes before interpreting movement as a pay trend.

The average is downstream. Defend who gets admitted first.

Your revenue ranking needs more than a ticker key

Nimbryxo-studio — Fri, 06 Nov 2026 09:52:10 +0000

Your revenue ranking can change after a join, before any company reports new revenue.

You're preparing the account priorities for Friday. The enrichment job has finished, but the table puts annual revenue beside quarterly revenue. The SQL runs. That's the annoying part.

Financial statements describe different reporting windows, so their separate paths carry meaning. Quarterly collection also requires the annual-financials toggle. Flatten away those distinctions, and your scoring job can quietly promote the wrong accounts.

Request the statements your score actually needs

Start with a bounded watchlist. Disable unrelated collections so you can inspect the financial ingestion without carrying history and news through it.

Install apify-client, zod, and pg. Supply APIFY_TOKEN and the standard PG* connection environment variables.

Example input for the revenue workflow:

import { ApifyClient } from 'apify-client';
import { z } from 'zod';
import pg from 'pg';

const run_input = {
  mode: 'tickers',
  tickers: ['AAPL'],
  includeFinancials: true,
  includeQuarterlyFinancials: true,
  includeHistory: false,
  includeDividends: false,
  includeSplits: false,
  includeNews: false,
  includeRecommendations: false,
  includeAnalystPriceTargets: false,
  includeEarningsHistory: false,
  includeEarningsCalendar: false,
};

This requests both cadences for storage. Your scoring policy will choose which one to use.

Keeping collection separate from scoring matters. You shouldn't need another collection run because someone wants to compare quarterly performance instead.

Wait before you read the dataset

Starting collection and replaying ingestion are different operations.

Set SOURCE_RUN_ID when you're repairing the database side. Leave it unset when you want a fresh source observation.

This cell waits for completion, then downloads the dataset through the REST endpoint:

if (!process.env.APIFY_TOKEN) {
  throw new Error('APIFY_TOKEN is required');
}

const client = new ApifyClient({
  token: process.env.APIFY_TOKEN,
});

const started = process.env.SOURCE_RUN_ID
  ? { id: process.env.SOURCE_RUN_ID }
  : await client.actor(
      'nimbryxo-studio/yahoo-finance-market-data-and-fundamentals-scraper',
    ).start(run_input);

const run = await client.run(started.id).waitForFinish();

if (!run || run.status !== 'SUCCEEDED' || !run.defaultDatasetId) {
  throw new Error(`Source run is not publishable: ${run?.status}`);
}

const response = await fetch(
  `https://api.apify.com/v2/datasets/${run.defaultDatasetId}/items?format=json&clean=true`,
  {
    headers: {
      Authorization: `Bearer ${process.env.APIFY_TOKEN}`,
    },
  },
);

if (!response.ok) {
  throw new Error(`Dataset download failed: ${response.status}`);
}

const items = z.array(z.unknown()).parse(await response.json());

The full download keeps this recipe straightforward. Its cost is memory proportional to the dataset; a large watchlist deserves paginated ingestion with the same database contract.

A successful run is permission to inspect its records. It isn't proof that every requested account has usable financials.

What belongs in the statement key?

A financial statement key needs ticker, cadence, statement kind, and fiscal period.

The period end alone can't distinguish an annual income statement from a quarterly income statement ending on that date.

Carry the source structure into explicit columns:

Source	Stored distinction
`ticker`	Resolved source identity
`financials.income`, `.balance`, `.cashflow`	Annual cadence and statement kind
`financials.quarterly.income`, `.balance`, `.cashflow`	Quarterly cadence and statement kind
Fiscal-period map key	Period end
`financials.currency`	Reporting currency

The flattening operation should preserve this split:

ticker
  annual / income / fiscal period   -> statement
  quarterly / income / fiscal period -> statement

current key:
  ticker + cadence + kind + period

Use the returned ticker, rather than the input spelling. An alias can resolve to a different symbol, with requestedTicker retaining the original request.

That key is stable enough for refreshing source statements. It isn't a permanent company identifier. For account joins across symbol changes, corporate actions, or multiple listings, maintain a separately reviewed issuer mapping.

Currency stays outside the key deliberately. A changed reporting currency should revise the same logical period and raise a review issue, rather than silently create another statement identity.

Accept added fields, quarantine broken meanings

Optional data is omitted when the source doesn't publish it. Absence must stay distinct from zero revenue.

Use passthrough schemas because a new, unrelated field shouldn't stop account enrichment. Validate the structure you consume, then preserve statement labels in JSONB.

This cell quarantines malformed items and flattens usable periods:

const Statement = z.object({
  Total_Revenue: z.unknown().optional(),
}).passthrough();

const PeriodMap = z.record(z.string(), Statement);

const StatementSet = z.object({
  income: PeriodMap.optional(),
  balance: PeriodMap.optional(),
  cashflow: PeriodMap.optional(),
}).passthrough();

const TickerItem = z.object({
  recordType: z.literal('ticker'),
  ticker: z.string().trim().min(1),
  requestedTicker: z.string().optional(),
  scrapedAt: z.string().datetime({ offset: true }),
  financials: StatementSet.extend({
    currency: z.string().trim().min(1).optional(),
    quarterly: StatementSet.optional(),
    periods: z.unknown().optional(),
  }).optional(),
}).passthrough();

const ErrorItem = z.object({
  recordType: z.literal('error'),
}).passthrough();

type StatementRow = {
  item_index: number;
  ticker: string;
  cadence: 'annual' | 'quarterly';
  kind: 'income' | 'balance' | 'cashflow';
  period: string;
  reporting_currency: string;
  scraped_at: string;
  revenue: string | null;
  raw: Record<string, unknown>;
};

type Issue = {
  item_index: number;
  code: string;
  detail: string;
  raw: unknown;
};

const rows: StatementRow[] = [];
const issues: Issue[] = [];
const seen = new Set<string>();

function flag(
  item_index: number,
  code: string,
  detail: string,
  raw: unknown,
) {
  issues.push({ item_index, code, detail, raw });
}

function validPeriod(value: string): boolean {
  return /^\d{4}-\d{2}-\d{2}$/.test(value)
    && Number.isFinite(Date.parse(value))
    && new Date(value).toISOString().slice(0, 10) === value;
}

function decimal(value: unknown): string | null {
  if (typeof value === 'number' && Number.isSafeInteger(value)) {
    return String(value);
  }
  return typeof value === 'string' && /^-?\d+(?:\.\d+)?$/.test(value)
    ? value
    : null;
}

items.forEach((raw, item_index) => {
  const parsed = TickerItem.safeParse(raw);

  if (!parsed.success) {
    flag(
      item_index,
      ErrorItem.safeParse(raw).success ? 'source_error' : 'schema_rejected',
      JSON.stringify(parsed.error.issues),
      raw,
    );
    return;
  }

  const item = parsed.data;
  seen.add(item.ticker);
  if (item.requestedTicker) seen.add(item.requestedTicker);

  const financials = item.financials;
  if (!financials?.currency) {
    flag(item_index, 'missing_financials_or_currency', item.ticker, raw);
    return;
  }

  const cadences = [
    ['annual', financials],
    ['quarterly', financials.quarterly],
  ] as const;

  for (const [cadence, statements] of cadences) {
    for (const kind of ['income', 'balance', 'cashflow'] as const) {
      const periods = statements?.[kind];
      if (!periods || Object.keys(periods).length === 0) {
        flag(item_index, 'missing_statement', `${cadence}/${kind}`, raw);
        continue;
      }

      for (const [period, statement] of Object.entries(periods)) {
        const detail = JSON.stringify([item.ticker, cadence, kind, period]);

        if (!validPeriod(period)) {
          flag(item_index, 'unusable_period', detail, statement);
          continue;
        }

        const revenue = kind === 'income'
          ? decimal(statement.Total_Revenue)
          : null;

        if (kind === 'income' && revenue === null) {
          flag(item_index, 'unusable_revenue', detail, statement);
        }

        rows.push({
          item_index,
          ticker: item.ticker,
          cadence,
          kind,
          period,
          reporting_currency: financials.currency,
          scraped_at: item.scrapedAt,
          revenue,
          raw: statement,
        });
      }
    }
  }
});

for (const ticker of run_input.tickers) {
  if (!seen.has(ticker)) {
    flag(-1, 'missing_ticker', ticker, null);
  }
}

The revenue parser is intentionally conservative. Safe integer numbers and plain decimal strings become database numeric values. Formatted strings, fractional JSON numbers, and unsafe integers require review.

We'd rather leave the score absent than repair a financial amount by guessing.

A broken financial container quarantines its whole item. An unusable period label excludes that period. An unusable revenue value still preserves the statement, but leaves its typed revenue empty.

Invalid or delisted symbols produce error records instead of partial ticker records. Those records enter quarantine too, with their original payload available for diagnosis.

Strict validation would also reject harmless additions. Passthrough accepts those additions while still detecting changes that break this workflow's financial meaning.

Keep observations beside the current row

Typed columns make ranking predictable. Raw JSONB preserves statement labels and gives you somewhere to inspect revisions without adding columns for every accounting line.

They serve different jobs. JSONB alone makes every dashboard its own parser; typed columns alone discard useful evidence.

This schema keeps an immutable observation per source item and a refreshable current statement:

const statementColumns = `
  run_id text NOT NULL,
  item_index integer NOT NULL,
  ticker text NOT NULL,
  cadence text NOT NULL,
  kind text NOT NULL,
  period date NOT NULL,
  reporting_currency text NOT NULL,
  scraped_at timestamptz NOT NULL,
  revenue numeric,
  raw jsonb NOT NULL
`;

const setupSql = `
  CREATE TABLE IF NOT EXISTS financial_observation (
    ${statementColumns},
    PRIMARY KEY (run_id, item_index, ticker, cadence, kind, period)
  );

  CREATE TABLE IF NOT EXISTS financial_current (
    ${statementColumns},
    PRIMARY KEY (ticker, cadence, kind, period)
  );

  CREATE TABLE IF NOT EXISTS ingestion_issue (
    run_id text NOT NULL,
    item_index integer NOT NULL,
    code text NOT NULL,
    detail text NOT NULL,
    raw jsonb NOT NULL,
    PRIMARY KEY (run_id, item_index, code, detail)
  );
`;

Replaying the same completed dataset makes the observation inserts no-ops. A fresh collection run creates a separate observation, even when its values agree with the previous run.

Restatements belong in that history. Updating the current row without keeping observations would erase the distinction between “collected again” and “reported differently.”

The writer below batches inserts and upserts. Its transaction lock serializes this ingestion path so currency comparisons see a consistent predecessor.

type RankingPolicy = {
  cadence: 'annual' | 'quarterly';
  currency: string;
  asOf: string;
  maxAgeDays: number;
};

async function ingest(rankingSql: string, policy: RankingPolicy) {
  const pool = new pg.Pool();

  try {
    const db = await pool.connect();

    try {
      await db.query('BEGIN');
      await db.query('SELECT pg_advisory_xact_lock(4101)');
      await db.query(setupSql);

      await db.query(`
        INSERT INTO ingestion_issue
        SELECT $1::text, x.*
        FROM jsonb_to_recordset($2::jsonb) AS x(
          item_index integer, code text, detail text, raw jsonb
        )
        ON CONFLICT DO NOTHING
      `, [run.id, JSON.stringify(issues)]);

      await db.query(`
        WITH saved AS (
          INSERT INTO financial_observation
          SELECT $1::text, x.*
          FROM jsonb_to_recordset($2::jsonb) AS x(
            item_index integer, ticker text, cadence text, kind text,
            period date, reporting_currency text, scraped_at timestamptz,
            revenue numeric, raw jsonb
          )
          ON CONFLICT DO NOTHING
          RETURNING *
        ),
        candidates AS (
          SELECT DISTINCT ON (ticker, cadence, kind, period) *
          FROM saved
          ORDER BY ticker, cadence, kind, period,
                   scraped_at DESC, run_id DESC, item_index DESC
        ),
        drift AS (
          INSERT INTO ingestion_issue
          SELECT n.run_id, n.item_index,
            CASE
              WHEN n.reporting_currency <> c.reporting_currency
                THEN 'currency_changed'
              ELSE 'timestamp_conflict'
            END,
            jsonb_build_array(n.ticker, n.cadence, n.kind, n.period)::text,
            jsonb_build_object('previous', to_jsonb(c), 'incoming', to_jsonb(n))
          FROM candidates n
          JOIN financial_current c USING (ticker, cadence, kind, period)
          WHERE (
            n.reporting_currency <> c.reporting_currency
            AND (n.scraped_at, n.run_id, n.item_index)
              > (c.scraped_at, c.run_id, c.item_index)
          ) OR (
            n.scraped_at = c.scraped_at AND n.raw IS DISTINCT FROM c.raw
          )
          ON CONFLICT DO NOTHING
        )
        INSERT INTO financial_current AS c
        SELECT * FROM candidates WHERE true
        ON CONFLICT (ticker, cadence, kind, period) DO UPDATE SET
          run_id = EXCLUDED.run_id,
          item_index = EXCLUDED.item_index,
          reporting_currency = EXCLUDED.reporting_currency,
          scraped_at = EXCLUDED.scraped_at,
          revenue = EXCLUDED.revenue,
          raw = EXCLUDED.raw
        WHERE (EXCLUDED.scraped_at, EXCLUDED.run_id, EXCLUDED.item_index)
          > (c.scraped_at, c.run_id, c.item_index)
      `, [run.id, JSON.stringify(rows)]);

      await db.query('COMMIT');
    } catch (error) {
      await db.query('ROLLBACK');
      throw error;
    } finally {
      db.release();
    }

    const ranking = await pool.query(rankingSql, [
      policy.cadence,
      run.id,
      policy.asOf,
      policy.currency,
      policy.maxAgeDays,
    ]);

    const alerts = await pool.query(`
      SELECT code, count(*) AS affected
      FROM ingestion_issue
      WHERE run_id = $1
      GROUP BY code
    `, [run.id]);

    return { ranking: ranking.rows, alerts: alerts.rows };
  } finally {
    await pool.end();
  }
}

An older replay can't overwrite a newer source timestamp. Equal timestamps use run identity and item position as deterministic tie-breakers; conflicting stored statement content also creates a review issue. That ordering is bookkeeping, not evidence of which statement is financially authoritative.

The lock trades throughput for simpler correctness. At higher volume, partition the writer or introduce coordinated locks by statement key.

A database failure rolls back observations, issues, and current rows together. A crash after commit is safe to replay. Quarantined records don't erase previously accepted statements.

The worked source here's Yahoo Finance Market Data & Fundamentals Scraper. The storage contract belongs to your scoring system.

Rank a defined cohort, not every revenue value

Revenue measures activity over a period. A balance sheet measures a position at a date. Neither the statement kind nor cadence belongs in an optional dashboard filter.

Choose the latest period before checking its currency and revenue. Otherwise, an unusable recent statement can silently send the query searching backward for a more convenient answer.

Example scoring policy: annual income statements, a single reporting currency, and an explicit freshness window:

const rankingSql = `
  WITH latest AS (
    SELECT DISTINCT ON (ticker)
      ticker, period, reporting_currency, revenue
    FROM financial_current
    WHERE cadence = $1
      AND kind = 'income'
      AND run_id = $2
      AND period <= $3::date
    ORDER BY ticker, period DESC
  )
  SELECT ticker, period, reporting_currency, revenue,
         dense_rank() OVER (ORDER BY revenue DESC) AS revenue_rank
  FROM latest
  WHERE reporting_currency = $4
    AND revenue IS NOT NULL
    AND period >= $3::date - $5::integer
  ORDER BY revenue DESC, ticker
`;

const ingestion = await ingest(rankingSql, {
  cadence: 'annual',
  currency: 'USD',
  asOf: new Date().toISOString().slice(0, 10),
  maxAgeDays: 500,
});

The provenance filter excludes accounts whose current statements weren't accepted from this source run. That makes partial ingestion visible as missing coverage instead of quietly recycling old enrichment.

This policy excludes other currencies. It doesn't convert them. Cross-currency ranking needs an explicit exchange-rate source, conversion convention, and rate date.

Annual cadence still doesn't prove identical duration or economic conditions. Fiscal year ends differ, and fiscal-calendar changes can create unusual reporting windows. Period-end keys alone can't establish duration; review those issuer-period exceptions before treating the ranking as an accounting comparison.

Before this reaches the account score

Keep ingestion.alerts available to your scheduler, alongside the ranked cohort.

Alert on run failures, database failures, and an unexpectedly empty eligible cohort.
Route source errors and schema rejections to separate review queues.
Review missing statements, unusable periods, and rejected revenue values before filling gaps.
Hold currency changes and conflicting equal-timestamp observations for review.
Replay a retained run ID when repairing ingestion; start fresh collection when you need a new observation.
Compare eligible coverage with the requested account list before replacing a published score.

A smaller ranked cohort is inconvenient. A larger cohort built from incompatible revenue is convincing.

Your width filter does not certify downloaded pixels

Nimbryxo-studio — Thu, 05 Nov 2026 10:13:48 +0000

Your width filter can pass a photo that arrives at thumbnail size.

You're checking why the training job rejects images your ingestion job already approved. Another evening disappears into a filter that looks perfectly reasonable.

minWidth and minHeight filter original media dimensions, while imageSize selects the rendition served through photoUrl.

Example inputs keep the same source thresholds while changing the requested rendition:

const originalInput = {
  mode: "place",
  placeUrls: ["Empire State Building"],
  photoCategory: "photos",
  minWidth: 1200,
  minHeight: 1200,
  imageSize: "original",
};

const run_input = {
  ...originalInput,
  imageSize: "thumbnail",
};

Both inputs are valid. Neither turns source metadata into a measurement of downloaded bytes.

The trap is treating adjacent settings as a shared guarantee: your selection rule approves the original, but your training loader receives a particular representation of it.

Why can a width filter admit a thumbnail?

The gallery Actor separates media selection from URL rendition selection because those decisions serve different stages of a workflow.

You might need a large original eventually, while a small preview is enough to decide whether the photograph contains anything useful. Downloading everything at maximum size makes that decision unnecessarily expensive.

The thumbnail rendition is about 200 px. Selecting original requests the largest version the host serves.

Neither setting changes the meaning of the source dimensions. They describe the original media, even when you request a smaller representation.

That distinction belongs in your data model:

Evidence	What it answers	What it cannot establish
Source `width` and `height`	Does the original have sufficient dimensions?	The geometry of fetched bytes
Requested `imageSize`	Which rendition did you request?	Successful delivery or decoding
Dimensions inspected from the file	Does this representation meet your size rule?	Visual detail, relevance, or training suitability

A column named width becomes expensive when your ingestion job, downloader, and training loader quietly assign it different meanings.

Give each measurement a subject. Your future debugging session deserves nouns.

Unknown dimensions are still candidates

The Actor keeps media whose dimensions are unknown, and it omits fields the source doesn't publish.

That preserves potentially useful candidates. Missing evidence of size doesn't establish that an image is too small, so excluding it would make metadata availability decide which images you can examine.

Your consumer needs a stricter boundary.

Passing the source filter means the record remains eligible for inspection; it doesn't mean the delivered image has earned a place in your training corpus.

This workflow puts that boundary after delivery:

public media
    |
source filters
    |
candidate + selected rendition
    |
bounded download
    |
file geometry check
    |
full decode and training checks

Known dimensions help you avoid unnecessary downloads before transfer. Unknown dimensions reach the same file check, without an exemption or an invented measurement.

The file check can save expensive downstream work. It can't refund the bytes that reached it.

Keep source metadata separate from delivery evidence

Validate the fields your consumer needs without converting missing dimensions into zeros.

That distinction matters when you investigate rejections: an unavailable source measurement and a measured undersized original explain different things, even if neither provides source qualification.

This cell retrieves a dataset page and defines the record shape used below:

import { ApifyClient } from "apify-client";
import { z } from "zod";
import { Buffer } from "node:buffer";

const client = new ApifyClient({ token: process.env.APIFY_TOKEN });
const run = await client
  .actor("nimbryxo-studio/google-maps-photo-gallery-extractor-by-category")
  .call(run_input);

if (run.status !== "SUCCEEDED") {
  throw new Error(`Gallery run ended with ${run.status}`);
}

const { items } = await client.dataset(run.defaultDatasetId).listItems();

const PhotoRow = z.object({
  photoId: z.string().min(1),
  photoUrl: z.string().url(),
  width: z.number().int().positive().optional(),
  height: z.number().int().positive().optional(),
  mediaType: z.enum(["photo", "video", "streetView"]).optional(),
  isVideo: z.boolean().optional(),
  isStreetView: z.boolean().optional(),
});

Requiring an identifier and usable URL is your admission policy. A record that can't satisfy it goes to quarantine, where you can inspect the failure instead of silently repairing the evidence.

Keep media classification explicit too. recordType remains "photo" for backward compatibility, including video rows, and a video's main image URL points to its poster frame.

A poster can pass a dimension gate perfectly. That doesn't make it the kind of photograph you intended to collect.

The consumer below requires mediaType to identify a still photo and also checks the video and Street View flags. Missing classification stays outside this queue.

Dataset traversal is a separate concern. This example handles the retrieved page; consuming a page doesn't establish that you've consumed the dataset or exhausted a place's gallery.

Inspect the representation you fetched

Node has no general image decoder built in. With the package set used here, a small header probe can inspect raster dimensions before your existing training loader performs a full decode.

The probe below recognizes PNG and JPEG. Unsupported encodings remain unqualified rather than acquiring dimensions copied from the source record.

Keep the distinction clear: reading dimensions from a header doesn't prove the remaining file is intact, and it doesn't establish how much useful detail the image contains.

This cell bounds each transfer and inspects the delivered header:

type Dimensions = { width: number; height: number };

function probeDimensions(bytes: Buffer): Dimensions | undefined {
  const dimensions = (width: number, height: number) =>
    width > 0 && height > 0 ? { width, height } : undefined;

  const png = Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]);
  if (
    bytes.length >= 33 && bytes.subarray(0, 8).equals(png) &&
    bytes.readUInt32BE(8) === 13 &&
    bytes.toString("ascii", 12, 16) === "IHDR"
  ) {
    return dimensions(bytes.readUInt32BE(16), bytes.readUInt32BE(20));
  }

  if (bytes.length < 4 || bytes.readUInt16BE(0) !== 0xffd8) return;
  let offset = 2;
  while (offset < bytes.length) {
    if (bytes[offset++] !== 0xff) return;
    while (offset < bytes.length && bytes[offset] === 0xff) offset++;
    if (offset >= bytes.length) return;
    const marker = bytes[offset++];
    if ([0x00, 0xda, 0xd9].includes(marker)) return;
    if (marker === 0x01 || (marker >= 0xd0 && marker <= 0xd8)) continue;
    if (offset + 2 > bytes.length) return;
    const length = bytes.readUInt16BE(offset);
    if (length < 2 || offset + length > bytes.length) return;
    const isFrame = marker >= 0xc0 && marker <= 0xcf &&
      ![0xc4, 0xc8, 0xcc].includes(marker);
    if (isFrame) {
      if (length < 8) return;
      return dimensions(
        bytes.readUInt16BE(offset + 5), bytes.readUInt16BE(offset + 3),
      );
    }
    offset += length;
  }
}

const maxDownloadBytes = 16 * 1024 * 1024;
const downloadTimeoutMs = 15_000;

async function downloadBounded(url: string): Promise<Buffer> {
  const response = await fetch(url, {
    signal: AbortSignal.timeout(downloadTimeoutMs),
  });
  if (!response.ok || !response.body) {
    await response.body?.cancel();
    throw new Error(`Image request failed: HTTP ${response.status}`);
  }
  const reader = response.body.getReader();
  const chunks: Uint8Array[] = [];
  let total = 0;
  try {
    while (true) {
      const part = await reader.read();
      if (part.done) break;
      total += part.value.byteLength;
      if (total > maxDownloadBytes) throw new Error("Transfer budget exceeded");
      chunks.push(part.value);
    }
    return Buffer.concat(chunks, total);
  } finally {
    await reader.cancel().catch(() => undefined);
    reader.releaseLock();
  }
}

These transfer limits are local consumer choices. Set them for your workload, then keep a separate pixel allocation limit in the decoder, since compressed byte size doesn't bound decoded memory.

If your loader applies orientation transforms or crops before admission, measure that resulting representation there too. The dimensions that matter depend on the stage making the decision.

Most importantly, decode the same bytes you inspected. Fetching the URL again for training would introduce another delivery whose properties you haven't checked.

Give delivery failures their own lane

Public image URLs can change or disappear. A failed request doesn't tell you whether the original had adequate resolution.

Keep request failures, unsupported headers, and undersized deliveries separate, or your rejection metric will combine problems that require different fixes.

For this worked example, Google Maps Photo Gallery Extractor by Category supplies the candidates. The reusable pattern is to retain source claims alongside evidence collected from each delivery.

This cell records both measurements and groups outcomes by requested rendition and source qualification:

const quarantined: unknown[] = [];
const skipped: unknown[] = [];
const readyForDecode: Array<{ photoId: string; bytes: Buffer }> = [];
const counts = new Map<string, number>();
const observations: Array<{
  photoId: string;
  rendition: string;
  sourceWidth: number | null;
  sourceHeight: number | null;
  downloadedWidth: number | null;
  downloadedHeight: number | null;
  sourceQualified: boolean;
  outcome: string;
  failure: string | null;
}> = [];

for (const raw of items) {
  const parsed = PhotoRow.safeParse(raw);
  if (!parsed.success) {
    quarantined.push(raw);
    continue;
  }
  const row = parsed.data;
  if (
    row.mediaType !== "photo" ||
    row.isVideo === true || row.isStreetView === true
  ) {
    skipped.push(raw);
    continue;
  }
  const sourceQualified =
    row.width !== undefined && row.height !== undefined &&
    row.width >= run_input.minWidth && row.height >= run_input.minHeight;

  let measured: Dimensions | undefined;
  let outcome: string;
  let failure: string | null = null;
  try {
    const bytes = await downloadBounded(row.photoUrl);
    measured = probeDimensions(bytes);
    if (!measured) {
      outcome = "needs-probe";
    } else if (
      measured.width < run_input.minWidth ||
      measured.height < run_input.minHeight
    ) {
      outcome = "delivery-too-small";
    } else {
      outcome = "size-qualified";
      readyForDecode.push({ photoId: row.photoId, bytes });
    }
  } catch (error) {
    outcome = "delivery-error";
    failure = error instanceof Error ? error.message : "Download failed";
  }
  observations.push({
    photoId: row.photoId,
    rendition: run_input.imageSize,
    sourceWidth: row.width ?? null,
    sourceHeight: row.height ?? null,
    downloadedWidth: measured?.width ?? null,
    downloadedHeight: measured?.height ?? null,
    sourceQualified,
    outcome,
    failure,
  });
  const group = JSON.stringify([
    run_input.imageSize, sourceQualified, outcome,
  ]);
  counts.set(group, (counts.get(group) ?? 0) + 1);
}

The observations preserve uncertainty in separate source and download columns. The grouped counts let you isolate deliveries that failed the size gate despite qualifying source metadata, with the requested rendition attached.

A false source qualification means the available metadata didn't establish eligibility. The retained dimensions let you distinguish missing evidence from an explicit measurement below your threshold.

The queue retains bytes for the decoder. For continuous ingestion, hand each accepted buffer to your loader or object storage and release it; a bounded file size doesn't bound total batch memory.

Persist the delivery observation alongside the stored asset. Keeping dimensions while discarding which representation they describe recreates the original bug with better column names.

What broke was the admission rule

The broken assumption is that a source-qualified record represents a training-qualified file.

The Actor permits a small rendition of a sufficiently large original and retains candidates with unknown source dimensions. Both behaviors support useful workflows, but neither supplies file-level approval.

Upscaling an undersized delivery doesn't recover discarded detail. It can satisfy a tensor shape while giving the model enlarged blur.

For inexpensive visual selection, use previews to choose candidates, then request a larger rendition for accepted media. Join later results using photoId, the identifier intended for deduplication, rather than relying on matching positions across retrievals.

That identity still needs a delivery history. If you upsert every observation solely by media identifier, a later preview can overwrite measurements associated with a previously stored larger asset.

We'd keep media identity stable and attach delivery measurements to the particular asset retained by the consumer. A requested rendition helps explain the transfer, but it isn't a content identity.

Also account for the selection bias introduced by previews. Fine text or subtle visual features may disappear at preview size, so rejecting those previews can discard originals your task would have accepted.

A preview workflow saves transfers only when its selection step has enough information to make the decision. For OCR or detail-sensitive training data, that trade-off deserves checking before you optimize download size.

Before you resume the training job

Separate original dimensions from delivered dimensions.
Record the requested rendition with each retained asset.
Send unknown source dimensions through the file gate.
Separate request failures, unsupported encodings, and undersized deliveries.
Decode the inspected bytes before training admission.
Count source-qualified delivery rejections by rendition.

The pixels your model receives are the ones your admission rule has to judge.

Your sentiment chart can move when extraction changes

Nimbryxo-studio — Wed, 04 Nov 2026 10:16:47 +0000

Your weekly sentiment chart moved. Marketing wants an explanation, and you're checking the model before sending the report.

Start with the articles that reached it. A full-text cap leaves later articles in the dataset with fullTextStatus=skipped_cap; publisher refusals also remain as records. Filtering for usable bodies changes which coverage your chart represents.

A full-text sentiment trend reflects both article sentiment and the selection of articles whose bodies were available.

Example input: this starts a keyword collection with separate record and extraction budgets.

import { ApifyClient } from "apify-client";
import { z } from "zod";

const client = new ApifyClient({ token: process.env.APIFY_TOKEN });
const run_input = {
  mode: "keyword",
  queries: ["artificial intelligence"],
  dateRange: "7d",
  maxItems: 500,
  extractFullText: true,
  maxFullText: 100,
};

const run = await client
  .actor("nimbryxo-studio/google-news-keyword-and-full-text-article-scraper")
  .call(run_input);

if (run.status !== "SUCCEEDED") {
  throw new Error(`Inspect unfinished collection: ${run.status}`);
}

const { items } = await client.dataset(run.defaultDatasetId).listItems();

That success check belongs in the ingestion job. It doesn't certify every publisher response.

Which symptom explains the missing bodies?

The Actor caps publisher requests so a large metadata collection can't quietly become a large page-fetching job. It also preserves articles when extraction fails, keeping the metadata useful.

Your downstream filter decides whether those preserved records remain visible.

Observable symptom	Cause you can establish	Easy misdiagnosis	Fix and its cost
`skipped_cap`	The article was beyond the shared extraction budget	Publisher blocking	Increase the budget or select a separate extraction cohort; more requests
`short`	A small body was extracted	A complete, unusually brief article	Exclude it or maintain a separate scoring series; less coverage or more reporting work
`paywalled`	A subscription wall was detected	A temporary request failure	Retain metadata and exclude unavailable bodies; a persistent coverage gap
`blocked`	The publisher refused the request	A confirmed subscription wall	Inspect access conditions; review work without guaranteed recovery
`rate_limited`, `timeout`, `error`	The request encountered the named condition	Permanent publisher exclusion	Consider bounded retries; additional requests without guaranteed success
`empty`, `not_found`	No body was found, or the page wasn't found	The extraction cap	Inspect the URL and page; manual work or reduced coverage
`unresolved`	The publisher URL couldn't be resolved	Publisher refusal	Supply a verified publisher URL; URL repair work
Missing status or unfamiliar status	The row doesn't establish a recognized outcome	An ordinary extraction failure	Check saved inputs and preserve the raw value; investigation

A small body can be a preview. It doesn't establish why the body is small.

That distinction matters when someone proposes retrying every excluded article. A subscription wall is unlikely to improve through enthusiasm.

Preserve statuses before classifying them

Empty fields are omitted. Validate optional fields accordingly, and retain unfamiliar status strings rather than rejecting the entire dataset.

This schema validates the fields used by the audit.

const ArticleSchema = z.object({
  url: z.string().url(),
  domain: z.string().optional(),
  publishedAt: z.string().optional(),
  scrapedAt: z.string().optional(),
  sourceUrl: z.string().optional(),
  fullText: z.string().optional(),
  fullTextStatus: z.string().optional(),
});

type Article = z.infer<typeof ArticleSchema>;
const rows = z.array(ArticleSchema).parse(items);

const knownStatuses = new Set([
  "ok", "short", "empty", "blocked", "paywalled", "rate_limited",
  "not_found", "timeout", "error", "unresolved", "skipped_cap",
]);

function outcome(row: Article): string {
  if (row.fullTextStatus === undefined) return "missing_status";
  return knownStatuses.has(row.fullTextStatus)
    ? row.fullTextStatus
    : "unknown_status";
}

const byOutcome = new Map<string, Article[]>();
for (const row of rows) {
  const key = outcome(row);
  const bucket = byOutcome.get(key) ?? [];
  bucket.push(row);
  byOutcome.set(key, bucket);
}

Each table symptom now has a corresponding bucket. Missing and unfamiliar values remain separate, with the original string retained on each row.

Keep the submitted input beside the run identifier in your ingestion records. Items alone can't prove that extraction was enabled. A metadata-only run intentionally omits snippet, because the feed description repeats the headline.

For this input, extraction is enabled. Missing statuses therefore deserve investigation rather than a convenient metadata-only label.

Separate budget exhaustion from retry candidates

maxItems limits emitted unique articles. maxFullText limits page fetching across the run, within its allowed range of 1–500.

Increasing the record budget doesn't increase the extraction budget. The resulting dataset can offer broader metadata coverage while your body-based analysis remains constrained.

These selectors distinguish the remediation queues without claiming that every request failure is temporary.

const capSkipped = byOutcome.get("skipped_cap") ?? [];
const shortBodies = byOutcome.get("short") ?? [];
const paywalled = byOutcome.get("paywalled") ?? [];
const blocked = byOutcome.get("blocked") ?? [];
const unresolved = byOutcome.get("unresolved") ?? [];

const retryCandidates = rows.filter((row) =>
  ["rate_limited", "timeout", "error"].includes(row.fullTextStatus ?? "")
);

const pageReview = rows.filter((row) =>
  ["empty", "not_found"].includes(row.fullTextStatus ?? "")
);

const statusReview = [
  ...(byOutcome.get("missing_status") ?? []),
  ...(byOutcome.get("unknown_status") ?? []),
];

const okWithoutBody = rows.filter(
  (row) => row.fullTextStatus === "ok" && !row.fullText?.trim()
);

A nonempty capSkipped queue establishes a budget effect. The other queues establish different extraction outcomes, which can coexist in the same run.

Treat okWithoutBody as an ingestion or contract investigation. Don't silently score the headline to make that queue disappear.

For unresolved records, inspect url alongside sourceUrl. Resolution failure can preserve the original news link, so the presence of a URL doesn't prove that a publisher page was reached.

Keep excluded articles in the denominator

Your sentiment service may only need eligible text. Your coverage audit needs every emitted article.

Otherwise, dropping failed extractions before aggregation makes the remaining dataset look perfectly healthy. The denominator has left the building.

This aggregation retains extraction outcomes by publisher and publication day.

function publicationDay(value: string | undefined): string {
  const timestamp = value === undefined ? NaN : Date.parse(value);
  return Number.isFinite(timestamp)
    ? new Date(timestamp).toISOString().slice(0, 10)
    : "(missing-or-invalid)";
}

function summarize(input: Article[], daily: boolean) {
  const groups = new Map<string, {
    domain: string;
    period: string;
    emitted: number;
    eligible: number;
    outcomes: Record<string, number>;
  }>();

  for (const row of input) {
    const domain = row.domain ?? "(missing)";
    const period = daily ? publicationDay(row.publishedAt) : "(all)";
    const key = JSON.stringify([domain, period]);
    const group = groups.get(key) ?? {
      domain, period, emitted: 0, eligible: 0, outcomes: {},
    };

    group.emitted += 1;
    group.eligible += Number(
      row.fullTextStatus === "ok" && Boolean(row.fullText?.trim())
    );
    const status = outcome(row);
    group.outcomes[status] = (group.outcomes[status] ?? 0) + 1;
    groups.set(key, group);
  }

  return [...groups.values()];
}

const coverageByPublicationDay = summarize(rows, true);

Here, “eligible” is your scoring policy: a substantial body with nonblank text. It isn't an additional Actor status.

Use publication time for the coverage period. Keep scrapedAt for collection provenance. A rerun's collection time shouldn't move an older article into a different editorial period.

This denominator measures emitted articles, not every article published about your topic. Date filtering, deduplication, feed availability, and collection limits operate before your scoring decision.

Make the scoring basis travel with the result

ok indicates a substantial extracted body. It doesn't promise complete publisher coverage, nor does it establish that every available word was extracted.

Choose an explicit policy for short. For a body-based trend, exclusion is easier to interpret than silently mixing previews with substantial articles.

This creates a decision record for every article, including exclusions.

const scoringDecisions = rows.map((row) => {
  const eligible =
    row.fullTextStatus === "ok" && Boolean(row.fullText?.trim());

  return {
    url: row.url,
    domain: row.domain ?? null,
    publishedAt: row.publishedAt ?? null,
    scrapedAt: row.scrapedAt ?? null,
    extractionStatus: row.fullTextStatus ?? null,
    textBasis: eligible ? "body" : null,
    eligible,
    text: eligible ? row.fullText! : null,
  };
});

These are application-owned records. Store them beside your eventual scores in Postgres, carrying the run identifier and model version in your own ingestion envelope.

If you also need headline or preview sentiment, give each its own basis and series. A fallback can keep a workflow moving while quietly changing the question your chart answers.

The branch should remain visible.

emitted article
  ├─ substantial body → body sentiment + provenance
  └─ other outcome    → exclusion reason + coverage audit

Compare publisher weights before interpreting sentiment

A publisher's share of scored articles can change even when its share of collected articles doesn't.

That can move aggregate sentiment without any change in the sentiment model. Check this possibility before explaining the movement as a change in coverage tone.

This optional comparison reads a historical dataset and computes both publisher composition and cap exposure.

function profile(input: Article[]) {
  const groups = summarize(input, false);
  const eligibleTotal = groups.reduce((sum, group) => sum + group.eligible, 0);
  const skipped = input.filter(
    (row) => row.fullTextStatus === "skipped_cap"
  ).length;

  return {
    emitted: input.length,
    capShare: input.length ? skipped / input.length : null,
    bodyCoverage: input.length ? eligibleTotal / input.length : null,
    publishers: groups.map((group) => ({
      ...group,
      scoredShare: eligibleTotal ? group.eligible / eligibleTotal : null,
    })),
  };
}

const previousDatasetId = process.env.PREVIOUS_DATASET_ID;
const previousRows = previousDatasetId
  ? z.array(ArticleSchema).parse(
      (await client.dataset(previousDatasetId).listItems()).items
    )
  : undefined;

const comparison = previousRows
  ? { previous: profile(previousRows), current: profile(rows) }
  : null;

An absent baseline produces no comparison, rather than a reassuring comparison against the current run itself.

Use complete dataset reads for these rates. Keep query scope, edition, publication windows, extraction policy, and model settings available when assessing comparability.

The worked example here's Google News Keyword & Full-Text Article Scraper. The reusable pattern is preserving selection outcomes beside analytical results.

When publisher weights move, calculate a companion trend using publishers represented in both periods and a consistent weighting policy. Label it separately.

That comparison reduces a composition effect. It doesn't prove that the underlying articles are equivalent.

Spend the next request on the right uncertainty

Raise the extraction budget when skipped records explain the coverage loss and broader fetching serves the report. The cost is additional publisher requests and platform usage.

For a deliberate cohort, select URLs and submit them through articleUrls mode. It always attempts extraction, but the run-wide budget still applies. A separate run also brings another start charge and charges for emitted results.

Keep selection rules explicit. Choosing articles after reading their apparent sentiment would introduce another source of bias.

For paywalls, accept the unavailable body. The Actor reads publicly served content and doesn't bypass access walls. A preview or blocked response can still leave a publisher underrepresented.

For request failures, retry selectively with a budget. A timeout is a retry candidate, not evidence that retrying will recover the page.

The uncomfortable catch: the capped subset isn't established as a random sample. Output order isn't a sampling guarantee. Even stable publisher weights can hide changes in topics or article types within each publisher.

Before you send the report

Confirm the finished run and retain its submitted input.
Count every status, including missing and unfamiliar values.
Compare body coverage and publisher composition across periods.
Keep exclusions and text basis beside scoring results.
Retry request failures separately from access refusals.
Label any companion trend built from a comparable cohort.

A higher sentiment score is a finding. A different set of readable articles is a different explanation.

Your Shorts song join is picking the wrong creator

Nimbryxo-studio — Tue, 03 Nov 2026 10:16:24 +0000

Your campaign shortlist changes creators after a re-run. The song hasn't changed. Your join key has.

If you joined videoId straight into your creator table, a linked upload quietly became the campaign target. That's an uncomfortable discovery with a campaign brief waiting.

YouTube's Shorts chart ranks songs used in Shorts, not individual Shorts videos. The Actor labels these entries track. Some have no linked video ID and omit url, while retaining trackId.

Request the song chart without extra video detail

Start with a narrow research slice. Leave video enrichment out until you've a reason to investigate an attached upload.

Example request: this selects the latest daily global chart and keeps source-side deduplication disabled.

import { ApifyClient } from 'apify-client';
import { z } from 'zod';
import pg from 'pg';

const client = new ApifyClient({
  token: process.env.APIFY_TOKEN,
});

const run_input = {
  mode: 'charts',
  charts: ['shortsSongs'],
  countries: ['global'],
  period: 'daily',
  date: '',
  maxItems: 100,
  maxItemsPerChart: 100,
  includeVideoDetails: false,
  onlyNew: false,
};

An empty date asks for the latest chart. For a backfill, supply a supported historical date instead.

Keep onlyNew off for warehouse ingestion. Its monitor history is bounded, so it isn't your durable deduplication ledger. Postgres should own that decision.

Keep the destination separate from the entity

A convenient watch link answers “where can you inspect this?” It doesn't answer “what received this rank?”

Carry the relationship forward explicitly.

Shorts chart entry
        |
        v
trackId ----> track
        |
        v
chart observation ----> optional linked video
                                |
                                v
                       separate creator research

The observation identifies a track in a particular chart, market, date and period. An attached video belongs on that observation, outside the track's identity.

This also limits what your campaign automation can conclude. Song demand can inform creative selection. A linked upload doesn't establish the creator responsible for that demand, rights ownership or commercial permission.

Treat trackId as the source's track identity. If your workflow needs to reconcile recordings across catalogs, this key alone isn't enough. Add a reviewed mapping rather than merging titles that look similar.

Wait for completion and retain the run ID

Starting a fresh run and replaying an existing dataset are different recovery operations. Make that choice explicit.

This function starts the Actor unless APIFY_RUN_ID selects a run to resume.

async function getRun() {
  const runId = process.env.APIFY_RUN_ID ??
    (await client.actor(
      'nimbryxo-studio/youtube-trending-feed-extractor-by-category',
    ).start(run_input)).id;

  const finished = await client.run(runId).waitForFinish();
  const terminal = ['SUCCEEDED', 'FAILED', 'TIMED-OUT', 'ABORTED'];

  if (finished === undefined || !terminal.includes(finished.status)) {
    throw new Error(`Resume this run with APIFY_RUN_ID=${runId}`);
  }

  return finished;
}

Keep the returned ID in your job history. If database ingestion stops halfway through, resume that run instead of asking “latest” for another snapshot.

A failed source run may still have useful dataset rows. The workflow below preserves valid records but withholds campaign export until the run passes its release checks.

Quarantine identity failures without losing neighboring rows

Missing video fields are acceptable. Missing track identity isn't.

Validate the fields your joins and constraints actually require. This schema also checks the requested chart slice, so an unexpected entity or market can't quietly enter the same table.

const StatusRow = z.object({
  recordType: z.literal('status'),
  status: z.enum([
    'no_results',
    'unsupported_country',
    'unsupported_chart_period',
    'retired',
    'partial_source',
  ]),
  message: z.string().min(1),
  mode: z.enum(['charts', 'hubs']),
  scrapedAt: z.string().min(1),
}).passthrough();

const TrackRow = z.object({
  recordType: z.literal('chartEntry'),
  type: z.literal('track'),
  trackId: z.string().min(1),
  chartType: z.literal('shortsSongs'),
  countryCode: z.literal('global'),
  chartDate: z.string().min(1),
  chartPeriod: z.literal('daily'),
  position: z.number().int().positive(),
  title: z.string().min(1).optional(),
  artists: z.array(z.string().min(1)).min(1).optional(),
  videoId: z.string().min(1).optional(),
  url: z.string().url().optional(),
}).passthrough();

type Track = z.infer<typeof TrackRow>;
type Metadata = {
  title: string | null;
  artists: string[] | null;
};

Use passthrough validation deliberately. Rejecting every unmodeled field would quarantine legitimate enrichment that this integration doesn't consume.

The trade-off is narrower drift detection: new fields survive in raw storage without triggering an alert. Changed identity fields, incompatible types and unexpected entities do trigger review.

Each rejected record keeps its raw payload and dataset offset. You can investigate it without discarding unrelated tracks or guessing what the source originally returned.

Store the track and its observations separately

A track can appear across markets and chart dates. Putting rank directly on the track table would turn every import into an argument about which observation wins.

Use a compound observation key:

trackId, chartType, countryCode, chartDate, chartPeriod

Keep the returned date string intact. Don't replace it with your scheduler's date.

Storage choice	What it gives you	What it costs
Typed columns	Constraints, indexes and predictable research queries	Deliberate schema changes
Raw JSONB	Inspection and reprocessing without another source run	Storage and less convenient querying

The raw payload is evidence for your ingestion decisions. It isn't a substitute for a primary key.

This schema also stores release state and review items alongside the research tables.

const ddl = `
CREATE TABLE IF NOT EXISTS shorts_ingest (
  run_id text PRIMARY KEY,
  status text NOT NULL,
  ready boolean NOT NULL DEFAULT false
);

CREATE TABLE IF NOT EXISTS shorts_tracks (
  track_id text PRIMARY KEY,
  title text,
  artists text[],
  raw jsonb NOT NULL
);

CREATE TABLE IF NOT EXISTS shorts_observations (
  track_id text NOT NULL REFERENCES shorts_tracks(track_id),
  chart_type text NOT NULL,
  country_code text NOT NULL,
  chart_date text NOT NULL,
  chart_period text NOT NULL,
  position integer NOT NULL CHECK (position > 0),
  video_id text,
  url text,
  raw jsonb NOT NULL,
  run_id text NOT NULL REFERENCES shorts_ingest(run_id),
  PRIMARY KEY (
    track_id, chart_type, country_code, chart_date, chart_period
  )
);

CREATE TABLE IF NOT EXISTS shorts_issues (
  run_id text NOT NULL REFERENCES shorts_ingest(run_id),
  item_offset integer NOT NULL,
  reason text NOT NULL,
  raw jsonb NOT NULL,
  PRIMARY KEY (run_id, item_offset, reason)
);
`;

video_id stays nullable on the observation. That preserves the linked-video relationship without promoting it into track identity.

This design keeps the latest ingested version of each observation. If you need every revision of a chart, add an append-only revision table.

Make each dataset page safe to replay

Validate records independently, then commit accepted rows and review items in the same transaction.

The function below compares available metadata before writing. Artist names are sorted for comparison, so ordering alone doesn't create a conflict. Contradictory known metadata goes to review rather than silently renaming an existing track.

It also collapses duplicate keys within a batch. Postgres won't let an upsert update the same target row repeatedly within one statement.

async function savePage(
  pool: pg.Pool,
  runId: string,
  items: unknown[],
  offset: number,
) {
  const issues: {
    item_offset: number;
    reason: string;
    raw: unknown;
  }[] = [];
  const candidates: { item: Track; raw: unknown; at: number }[] = [];

  items.forEach((raw, index) => {
    const at = offset + index;
    const status = StatusRow.safeParse(raw);

    if (status.success) {
      issues.push({
        item_offset: at,
        reason: `status:${status.data.status}`,
        raw,
      });
      return;
    }

    const parsed = TrackRow.safeParse(raw);
    if (parsed.success) {
      candidates.push({ item: parsed.data, raw, at });
    } else {
      issues.push({
        item_offset: at,
        reason: `schema:${parsed.error.message}`,
        raw,
      });
    }
  });

  const db = await pool.connect();
  try {
    await db.query('BEGIN');
    await db.query(
      "SELECT pg_advisory_xact_lock(hashtext('shorts-song-ingestion'))",
    );

    const existing = await db.query<Metadata & { track_id: string }>(
      `SELECT track_id, title, artists FROM shorts_tracks
       WHERE track_id = ANY($1::text[])`,
      [[...new Set(candidates.map(({ item }) => item.trackId))]],
    );
    const metadata = new Map<string, Metadata>(
      existing.rows.map(row => [row.track_id, row]),
    );

    const payload = candidates.flatMap(({ item, raw, at }) => {
      const artists = item.artists
        ? [...new Set(item.artists)].sort()
        : null;
      const known = metadata.get(item.trackId);
      const titleConflict = known?.title != null &&
        item.title !== undefined && known.title !== item.title;
      const artistConflict = known?.artists != null &&
        artists !== null &&
        JSON.stringify([...known.artists].sort()) !== JSON.stringify(artists);

      if (titleConflict || artistConflict) {
        issues.push({
          item_offset: at,
          reason: 'metadata_conflict',
          raw,
        });
        return [];
      }

      const merged = {
        title: known?.title ?? item.title ?? null,
        artists: known?.artists ?? artists,
      };
      metadata.set(item.trackId, merged);

      return [{
        track_id: item.trackId,
        ...merged,
        chart_type: item.chartType,
        country_code: item.countryCode,
        chart_date: item.chartDate,
        chart_period: item.chartPeriod,
        position: item.position,
        video_id: item.videoId ?? null,
        url: item.url ?? null,
        raw,
      }];
    });

    const tracks = [...new Map(
      payload.map(row => [row.track_id, row]),
    ).values()];
    const observations = [...new Map(
      payload.map(row => [
        JSON.stringify([
          row.track_id, row.chart_type, row.country_code,
          row.chart_date, row.chart_period,
        ]),
        row,
      ]),
    ).values()];

    await db.query(`
      INSERT INTO shorts_tracks (track_id, title, artists, raw)
      SELECT track_id, title, artists, raw
      FROM jsonb_to_recordset($1::jsonb)
        AS x(track_id text, title text, artists text[], raw jsonb)
      ON CONFLICT (track_id) DO UPDATE SET
        title = EXCLUDED.title,
        artists = EXCLUDED.artists,
        raw = EXCLUDED.raw
    `, [JSON.stringify(tracks)]);

    await db.query(`
      INSERT INTO shorts_observations
        (track_id, chart_type, country_code, chart_date, chart_period,
         position, video_id, url, raw, run_id)
      SELECT track_id, chart_type, country_code, chart_date, chart_period,
             position, video_id, url, raw, $2
      FROM jsonb_to_recordset($1::jsonb) AS x(
        track_id text, chart_type text, country_code text,
        chart_date text, chart_period text, position integer,
        video_id text, url text, raw jsonb
      )
      ON CONFLICT
        (track_id, chart_type, country_code, chart_date, chart_period)
      DO UPDATE SET
        position = EXCLUDED.position,
        video_id = EXCLUDED.video_id,
        url = EXCLUDED.url,
        raw = EXCLUDED.raw,
        run_id = EXCLUDED.run_id
    `, [JSON.stringify(observations), runId]);

    await db.query(`
      INSERT INTO shorts_issues (run_id, item_offset, reason, raw)
      SELECT $1, item_offset, reason, raw
      FROM jsonb_to_recordset($2::jsonb)
        AS x(item_offset integer, reason text, raw jsonb)
      ON CONFLICT DO NOTHING
    `, [runId, JSON.stringify(issues)]);

    await db.query('COMMIT');
    return observations.length;
  } catch (error) {
    await db.query('ROLLBACK');
    throw error;
  } finally {
    db.release();
  }
}

The advisory lock serializes this integration's page writes. That keeps the metadata comparison and subsequent update consistent across concurrent workers.

It's a throughput trade-off. At larger scale, use staging tables and set-based reconciliation instead of widening this critical section.

A metadata conflict isn't automatically corruption. A corrected title can be legitimate. Holding that record is a campaign safety policy, and it needs a review path.

A finished run can still need review

The source can return status records explaining empty, unsupported or partial coverage. Those are operational signals, not tracks.

This implementation holds every status record for review. That's conservative, but appropriate when a quiet coverage gap could change a campaign shortlist.

Source charts are fixed lists. Paginating the dataset retrieves emitted records; it doesn't unlock additional source-chart pages. Never interpret absent tracks as deletions, especially after a capped or failed import.

This final cell reads every dataset page, commits useful records and releases only a clean, successful run.

const pool = new pg.Pool();

try {
  await pool.query(ddl);
  const run = await getRun();

  await pool.query(`
    INSERT INTO shorts_ingest (run_id, status, ready)
    VALUES ($1, $2, false)
    ON CONFLICT (run_id) DO UPDATE
      SET status = EXCLUDED.status, ready = false
  `, [run.id, run.status]);

  let offset = 0;
  let accepted = 0;

  for (;;) {
    const { items } = await client.dataset(run.defaultDatasetId)
      .listItems({ offset, limit: 100 });

    if (items.length === 0) break;
    accepted += await savePage(pool, run.id, items, offset);
    offset += items.length;
  }

  await pool.query(`
    UPDATE shorts_ingest
    SET ready = $2 AND NOT EXISTS (
      SELECT 1 FROM shorts_issues WHERE run_id = $1
    )
    WHERE run_id = $1
  `, [run.id, run.status === 'SUCCEEDED' && accepted > 0]);

  const { rows: campaignResearch } = await pool.query(`
    SELECT o.track_id, t.title, t.artists, o.position,
           o.video_id, o.url
    FROM shorts_observations o
    JOIN shorts_tracks t USING (track_id)
    JOIN shorts_ingest r USING (run_id)
    WHERE o.run_id = $1 AND r.ready
    ORDER BY o.position
  `, [run.id]);

  // Pass campaignResearch to the research step.
} finally {
  await pool.end();
}

Use the standard Postgres connection environment variables alongside APIFY_TOKEN.

A crash leaves the run unreleased. Replaying the same dataset updates existing observation keys and preserves issue identities. Previously committed pages remain available for inspection.

The upsert uses ingestion order, not a claim about source freshness. Schedule older backfills deliberately so they don't overwrite newer corrections accidentally.

For n8n, pass the selected run ID into the export query. Don't export by calendar date alone, and don't bypass ready because the source run says SUCCEEDED.

Alert on failed source runs, ingestion exceptions, schema failures, metadata conflicts and status records. Also alert when a successful import accepts nothing. The issue table stores the reasons; your scheduler or monitoring workflow must deliver the notification.

A missing linked video is intentionally absent from that list.

Before campaign research leaves the database

The worked source is YouTube Trending Feed Extractor by Category & Market. The reusable pattern is entity identity first, source observations second, activation after review.

Before releasing a slice:

Confirm the selected run is ready and has accepted observations.
Review quarantined records without deleting valid neighboring rows.
Resolve metadata conflicts before changing canonical track details.
Require an explicit linked-video relationship before joining creator research.
Keep commercial-use permission outside the chart inference.

A song trend can tell you what deserves research. It hasn't told you whom to contact.

Your location join is copying salaries across cities

Nimbryxo-studio — Mon, 02 Nov 2026 10:20:21 +0000

You expand a job into city rows, and every office suddenly appears to share a salary range. The dashboard looks consistent. That's the uncomfortable part.

A posting's selected salary range isn't a salary observation for every listed location.

The Actor chooses the range matching the primary location when applicable. Otherwise, it uses the posting's typical/base range. Flattening the location list copies that selection without adding evidence about another office.

Why that salary stays with the posting

Employers can publish several geographic pay bands inside a single job description. The Actor exposes a selected range through scalar compensation fields, alongside the sentence it came from.

That gives you something useful for a posting record. It doesn't give you a location-to-salary mapping.

Start by fetching details, because listing-only records don't provide the location and description enrichment this workflow needs.

Example input for a careers site, followed by retrieval:

import { ApifyClient } from 'apify-client';

const client = new ApifyClient({
  token: process.env.APIFY_TOKEN,
});

const run_input = {
  mode: 'tenant',
  tenant: 'workday',
  site: 'Workday',
  scrapeDetails: true,
  maxItems: 50,
};

const run = await client
  .actor('nimbryxo-studio/workday-careers-site-job-listings-extractor')
  .call(run_input);

if (run.status !== 'SUCCEEDED') {
  throw new Error(`Run ended with ${run.status}`);
}

const { items } = await client
  .dataset(run.defaultDatasetId)
  .listItems();

The transformation below processes those retrieved items. Retrieval completeness remains a separate concern.

Underneath, the Actor pages through search results and removes duplicates across pages and URLs. It splits capped searches by job family, then a small secondary facet when needed.

Those strategies address collection coverage. They don't change what the selected salary represents.

Keep availability and compensation apart

Your location bridge answers: “Where can someone take this role?”

Your compensation observation answers: “Which published range was selected for this posting?”

Those questions need different records. Keep the boundary visible:

posting
  ├── location availability → location labels
  └── selected compensation → range, source sentence, primary context

For sales territory planning, the availability rows can help identify hiring activity near an account's offices.

For a city compensation comparison, those same rows are insufficient. Joining them to the selected range creates geographic attribution through SQL syntax. Geography rarely accepts syntax as evidence.

Keep currency and compensation frequency attached, too. A range without its unit can't support a meaningful comparison.

Validate without inventing geography

Validate the scalar fields you intend to calculate with. Preserve unfamiliar location structures rather than guessing their nested keys.

This schema accepts optional compensation fields and retains the rest of the record:

import { z } from 'zod';

const jobSchema = z.object({
  jobPostingId: z.string().optional(),
  requisitionId: z.string().optional(),
  externalPath: z.string().optional(),
  url: z.string().optional(),
  location: z.string().optional(),
  primaryLocation: z.unknown().optional(),
  locations: z.array(z.unknown()).optional(),
  additionalLocations: z.array(z.unknown()).optional(),
  compensationRangeMin: z.number().finite().optional(),
  compensationRangeMax: z.number().finite().optional(),
  compensationCurrency: z.string().optional(),
  compensationFrequency: z.string().optional(),
  compensationText: z.string().optional(),
  compensationType: z.string().optional(),
}).passthrough();

Missing fields are omitted by the Actor. Your consumer should preserve that absence rather than converting it into zero pay.

The location mapper below handles string labels from location and additionalLocations. Structured entries stay in the original payload and trigger a mapping flag.

That flag describes your mapper's limits. It doesn't certify that the source listed every possible workplace.

Split the record before you split the rows

Create location availability independently from the selected compensation.

The attribution policy below is a consumer decision. It deliberately makes no claim that the Actor supplied a verified geographic attribution flag.

This transformation also rejects duplicate consumer keys within the batch:

const seenKeys = new Set<string>();

const postings = items.map((item) => {
  const job = jobSchema.parse(item);

  const identity = [
    ['jobPostingId', job.jobPostingId],
    ['requisitionId', job.requisitionId],
    ['externalPath', job.externalPath],
    ['url', job.url],
  ].find(([, value]) => Boolean(value));

  if (!identity) {
    throw new Error('Posting has no usable identity');
  }

  const posting_key = JSON.stringify([
    run_input.tenant,
    run_input.site,
    ...identity,
  ]);

  if (seenKeys.has(posting_key)) {
    throw new Error(`Duplicate consumer key: ${posting_key}`);
  }
  seenKeys.add(posting_key);

  const primaryLabel = job.location?.trim();
  const additional = job.additionalLocations ?? [];
  const additionalLabels = additional
    .filter((value): value is string => typeof value === 'string')
    .map((value) => value.trim())
    .filter(Boolean);

  const selectedPay = Object.fromEntries(
    Object.entries({
      compensationRangeMin: job.compensationRangeMin,
      compensationRangeMax: job.compensationRangeMax,
      compensationCurrency: job.compensationCurrency,
      compensationFrequency: job.compensationFrequency,
      compensationText: job.compensationText,
      compensationType: job.compensationType,
    }).filter(([, value]) => value !== undefined),
  );

  return {
    posting_key,
    raw: item,
    primary_location_context: job.primaryLocation ?? null,
    location_labels: [...new Set([
      ...(primaryLabel ? [primaryLabel] : []),
      ...additionalLabels,
    ])],
    needs_location_mapping:
      !primaryLabel ||
      additional.some(
        (value) => typeof value !== 'string' || !value.trim(),
      ),
    selected_pay: Object.keys(selectedPay).length ? selectedPay : null,
    attribution_policy: 'posting_selected_unverified_geography',
  };
});

Exact label deduplication prevents repeated labels within a posting. It doesn't resolve spelling variants or prove that matching labels describe the same place across employers.

Identity fallbacks deserve similar care. If the preferred identifier becomes available later, this key can change. Keep the identity policy stable for a source, or reconcile that transition explicitly.

For geographic pay attribution, retain compensationText and inspect its wording against the primary-location context. A populated primary location alone can't distinguish a geographic match from the base-range fallback.

We'd leave the geographic claim unresolved until that evidence is checked.

Replace snapshots in a batch

A small ingestion service can store each transformed snapshot as JSONB and expose separate views for availability and compensation.

Replacing the snapshot also replaces its derived locations and pay. That avoids stale child rows when a later record omits a previously published range.

This creates the storage boundary and writes the whole batch with a single upsert:

import pg from 'pg';

const pool = new pg.Pool();

try {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS postings (
      posting_key text PRIMARY KEY,
      body jsonb NOT NULL
    );

    CREATE OR REPLACE VIEW posting_locations AS
    SELECT p.posting_key, l.location_label
    FROM postings AS p
    CROSS JOIN LATERAL jsonb_array_elements_text(
      p.body->'location_labels'
    ) AS l(location_label);

    CREATE OR REPLACE VIEW compensation_observations AS
    SELECT
      posting_key,
      body->'primary_location_context' AS primary_location_context,
      body->>'attribution_policy' AS attribution_policy,
      body->'selected_pay'->>'compensationText' AS source_text,
      body->'selected_pay'->>'compensationCurrency' AS currency,
      body->'selected_pay'->>'compensationFrequency' AS frequency,
      (body->'selected_pay'->>'compensationRangeMin')::numeric
        AS range_min,
      (body->'selected_pay'->>'compensationRangeMax')::numeric
        AS range_max
    FROM postings
    WHERE body->'selected_pay' <> 'null'::jsonb;
  `);

  await pool.query(`
    INSERT INTO postings (posting_key, body)
    SELECT value->>'posting_key', value
    FROM jsonb_array_elements($1::jsonb)
    ON CONFLICT (posting_key)
    DO UPDATE SET body = EXCLUDED.body
  `, [JSON.stringify(postings)]);
} finally {
  await pool.end();
}

This stores current snapshots, not salary history. Serialize ingestion for each source if an older run could otherwise overwrite a newer snapshot.

It also doesn't delete postings absent from the batch. Absence from a capped or partial collection is insufficient evidence that a role closed.

In n8n, keep these derived structures separate before any item-splitting node. Once compensation travels inside each location item, downstream workflows can mistake copied values for independent observations.

Which average are you actually calculating?

The following queries make the difference visible. Both require complete, ordered bounds and explicit currency and frequency.

The first assigns selected ranges to availability labels. The second preserves posting grain:

-- Availability-weighted selected pay, grouped by location label.
SELECT
  l.location_label,
  c.currency,
  c.frequency,
  count(*) AS joined_rows,
  avg(c.range_min) AS mean_selected_minimum
FROM posting_locations AS l
JOIN compensation_observations AS c USING (posting_key)
WHERE c.range_min IS NOT NULL
  AND c.range_max IS NOT NULL
  AND c.range_min <= c.range_max
  AND c.currency IS NOT NULL
  AND c.frequency IS NOT NULL
GROUP BY l.location_label, c.currency, c.frequency;

-- Posting-weighted selected pay across the ingested portfolio.
SELECT
  currency,
  frequency,
  count(*) AS postings_with_ranges,
  avg(range_min) AS mean_selected_minimum
FROM compensation_observations
WHERE range_min IS NOT NULL
  AND range_max IS NOT NULL
  AND range_min <= range_max
  AND currency IS NOT NULL
  AND frequency IS NOT NULL
GROUP BY currency, frequency;

The first can describe selected pay attached to jobs available in a location. Labeling it “local salary” would overstate the evidence.

The second describes published selected minima across postings. It still isn't average employee pay, an offered salary, or a representative market estimate.

If you aggregate the joined rows across locations, postings with more location labels receive more weight. Even a company-wide chart can therefore change because of location fanout.

What broke was the attribution

The worked example here's Workday Careers Site Job Listings Extractor. The reusable pattern applies whenever a record contains a repeated dimension beside a selected scalar measurement.

The failure is a consumer assumption crossing a boundary:

Consumer assumption	Behavior that breaks it	Design consequence
Every location inherits the selected pay	Compensation selection happens at posting grain	Separate availability from compensation
Source deduplication makes re-runs unique	The Actor keeps no state between runs	Persist stable consumer identities
Missing pay means zero	Empty fields are omitted	Track missingness separately from numeric values
A shared cap gives balanced employer coverage	Results follow URL order; an earlier site can consume the cap	Track collection scope before comparing employers

There's another coverage limit: a search slice can remain capped after splitting. Narrowing the search can help reach those roles, but a successful run status alone doesn't establish completeness.

Your geographic attribution policy and your collection-coverage policy answer different questions. Neither should quietly substitute for the other.

Before the dashboard refreshes

Check unresolved location mappings before using office counts.
Keep the selected range, source sentence, units, and primary context together.
Require a separate evidence check before publishing geographic pay claims.
Name each metric's grain and denominator in the dashboard.
Compare coverage by employer before interpreting differences.

An office can be hiring without having a salary observation.