DEV Community: Nimbryxo-studio

Your upsert can preserve an offer that never existed

Nimbryxo-studio — Fri, 23 Oct 2026 13:23:00 +0000

Your retry stops creating duplicate rows. Now you need to decide whether a missing price should erase the previous one.

Keeping it feels responsible. But this source drops empty and null fields before publishing each record. A missing field carries no instruction to preserve an earlier value.

Merging sparse observations can assemble an offer you never observed. A retained price, a different seller, and fresh availability can look perfectly respectable together. Your database has become creative.

Define the destination before defining the key

You're building a product monitor for your own business. Its current row should answer: what did the latest observation actually contain?

Start with explicit product URLs. They make the requested marketplace visible before collection, which helps when you reconcile missing products later.

Example input and application scope:

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 = {
  productUrls: ['https://www.amazon.com/dp/B09X7MPX8L'],
  postalCode: '10001',
  proxyCountry: 'AUTO',
  includeRelatedProducts: false,
};

const destinationScope = JSON.stringify({
  postalCode: run_input.postalCode,
  proxyCountry: run_input.proxyCountry,
});

function productKey(url: string, asin: string): string {
  const host = new URL(url).hostname.toLowerCase().replace(/^www\./, '');
  return `${host}|${asin.toUpperCase()}`;
}

const manifest = new Set(run_input.productUrls.map((url) => {
  const match = new URL(url).pathname.match(/^\/dp\/([a-z0-9]{10})(?:\/|$)/i);
  if (!match) throw new Error(`Expected a direct product URL: ${url}`);
  return productKey(url, match[1]);
}));

The database key combines marketplace, ASIN, and this application-owned scope.

ASIN alone can't distinguish marketplace offers. Delivery targeting also affects buy-box price and availability, so a changed destination should create a separate observation stream.

Keep scope construction consistent across producers. If an equivalent request produces a differently serialized scope, your application creates another identity rather than updating the existing one.

The scope records your request. It doesn't certify that the page honored it. Postal-code targeting is best effort; retain locationText and check it before destination-sensitive repricing.

If the rendered destination doesn't match your requirements, keep the observation for diagnosis and withhold it from automated pricing decisions. A successfully stored record isn't permission to act on every field.

Also separate runs when requesting the same ASIN across marketplaces: repeated ASINs are scraped once. A wider manifest can't recover observations that were never emitted.

Replay the dataset without collecting again

A failed collection and a failed ingestion are different problems.

Keep the run ID in your scheduler's execution metadata immediately after starting collection. A database repair should replay its dataset, avoiding another collection run.

This cell starts a run unless you supply APIFY_RUN_ID, then waits for a terminal status:

const started = process.env.APIFY_RUN_ID
  ? { id: process.env.APIFY_RUN_ID }
  : await client
      .actor('nimbryxo-studio/amazon-multi-domain-product-price-and-detail')
      .start(run_input);

const terminalStatuses = new Set([
  'SUCCEEDED', 'FAILED', 'TIMED-OUT', 'ABORTED',
]);
let finished = await client.run(started.id).waitForFinish();
while (!finished || !terminalStatuses.has(finished.status)) {
  finished = await client.run(started.id).waitForFinish();
}
const sourceRun = finished.id;
const runStatus = finished.status;
const datasetId = finished.defaultDatasetId;
if (!datasetId) throw new Error(`Run ${sourceRun} has no dataset`);

The ingestion below accepts valid observations from a terminal run while flagging its unsuccessful status. That preserves useful records without presenting partial coverage as a clean refresh.

This policy suits a monitor that can display freshness beside each product. A workflow requiring a complete comparison set should hold publication until its coverage checks pass.

A replay must use the original input manifest and destination scope. Store those alongside the run ID. A dataset doesn't become a different destination because your environment changed.

Waiting also gives pagination a settled source. Reading while collection continues would require a separate policy for when coverage becomes final.

What should validation reject?

Validate the fields your application depends on. Preserve everything else.

This consumer requires usable identity and observation time. It accepts boolean availability and structured prices; differently shaped values go to quarantine rather than through a hopeful coercion.

The schema validates a narrow contract while retaining additional fields:

const moneySchema = z.object({
  value: z.number().finite().optional(),
  currency: z.string().optional(),
  display: z.string().optional(),
}).passthrough();

const observationSchema = z.object({
  url: z.string().url(),
  asin: z.string().trim().toUpperCase().regex(/^[A-Z0-9]{10}$/),
  scrapedAt: z.string().datetime(),
  price: moneySchema.optional(),
  shippingPrice: z.unknown().optional(),
  inStock: z.boolean().optional(),
  seller: z.object({
    name: z.string().optional(),
  }).passthrough().optional(),
  locationText: z.string().optional(),
}).passthrough();

type Observation = z.infer<typeof observationSchema>;

Strict validation of the entire record would turn harmless additions into ingestion failures. Passthrough validation keeps those additions available for later inspection.

It still rejects changes to fields this consumer understands. A price arriving in an unexpected representation deserves investigation, especially before a repricer touches it.

Quarantining the entire record is conservative. You lose a potentially useful observation when an offer field fails validation, but you avoid silently accepting a partially interpreted offer.

Missing optional fields take a different path. They're valid observations with less information, so they proceed to storage.

shippingPrice remains untyped because its representation isn't specified here. Its absence can't universally become zero: FREE shipping omits the field, but unsuccessful extraction can also leave fields absent.

That distinction matters for landed-cost comparisons. Unknown shipping can't safely participate in the same calculation as an explicitly known shipping charge.

Give current state and last-known values separate jobs

Store the original item in JSONB. Derive typed columns for filters and dashboards.

That duplicates some information, but it preserves the observation when your extraction logic changes. Typed columns alone make later diagnosis depend on the parser that happened to accept the record.

Keep the original item rather than serializing the parsed object. Validation normalizes identity for matching; your evidence should still show what arrived.

The storage responsibilities should look like this:

incoming observation
        |
        +--> current snapshot: replace the field set
        |
        +--> last-known price: retain with its own timestamp

no incoming observation
        |
        +--> existing snapshot ages; availability stays unchanged

This schema keeps the latest observation, a separately dated price fallback, and durable ingestion diagnostics:

const pool = new pg.Pool();

await pool.query(`
  CREATE TABLE IF NOT EXISTS product_current (
    marketplace text NOT NULL,
    asin text NOT NULL,
    destination_scope text NOT NULL,
    observed_at timestamptz NOT NULL,
    ingested_at timestamptz NOT NULL DEFAULT now(),
    source_run text NOT NULL,
    observation jsonb NOT NULL,
    price_display text,
    price_currency text,
    shipping jsonb,
    seller_name text,
    in_stock boolean,
    last_known_price jsonb,
    last_known_price_at timestamptz,
    PRIMARY KEY (marketplace, asin, destination_scope)
  );
  CREATE TABLE IF NOT EXISTS product_quarantine (
    source_run text NOT NULL,
    item_offset bigint NOT NULL,
    reason jsonb NOT NULL,
    observation jsonb NOT NULL,
    PRIMARY KEY (source_run, item_offset)
  );
  CREATE TABLE IF NOT EXISTS product_ingestion_report (
    source_run text NOT NULL,
    destination_scope text NOT NULL,
    report jsonb NOT NULL,
    checked_at timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (source_run, destination_scope)
  );
`);

Expose observed_at with current-state data. Expose last_known_price_at wherever you show the fallback.

ingested_at answers when your database accepted the record. It can't make an old scrape fresh.

The price projection keeps display text and currency without pretending that formatted text is a calculation-ready amount. Add a numeric projection when your downstream calculation has an explicit currency policy.

JSONB also has a limit: replacing it preserves the latest payload, not every previous observation. If you need price history or an audit trail beyond retained source datasets, add an append-only observation ledger.

Replace the snapshot behind the stable key

Avoid JSONB concatenation and COALESCE for current offer fields. Both can quietly retain values from another observation.

A seller change makes this particularly misleading. Carrying forward the previous price can attach it to the newly observed seller, even though that combination never appeared in the incoming record.

The upsert replaces the raw observation, clears absent projections, and advances only when scrapedAt is newer:

async function upsertObservation(raw: unknown, item: Observation) {
  const [marketplace, asin] = productKey(item.url, item.asin).split('|');
  const knownPrice = item.price &&
    (item.price.value !== undefined || item.price.display !== undefined)
    ? JSON.stringify(item.price) : null;

  await pool.query(`
    INSERT INTO product_current AS stored (
      marketplace, asin, destination_scope, observed_at, source_run,
      observation, price_display, price_currency, shipping,
      seller_name, in_stock, last_known_price, last_known_price_at
    ) VALUES (
      $1, $2, $3, $4, $5, $6::jsonb, $7, $8, $9::jsonb, $10, $11,
      $12::jsonb,
      CASE WHEN $12::jsonb IS NOT NULL THEN $4::timestamptz END
    )
    ON CONFLICT (marketplace, asin, destination_scope) DO UPDATE SET
      observed_at = EXCLUDED.observed_at,
      ingested_at = now(),
      source_run = EXCLUDED.source_run,
      observation = EXCLUDED.observation,
      price_display = EXCLUDED.price_display,
      price_currency = EXCLUDED.price_currency,
      shipping = EXCLUDED.shipping,
      seller_name = EXCLUDED.seller_name,
      in_stock = EXCLUDED.in_stock,
      last_known_price = COALESCE(
        EXCLUDED.last_known_price, stored.last_known_price),
      last_known_price_at = COALESCE(
        EXCLUDED.last_known_price_at, stored.last_known_price_at)
    WHERE EXCLUDED.observed_at > stored.observed_at
  `, [
    marketplace, asin, destinationScope, item.scrapedAt, sourceRun,
    JSON.stringify(raw), item.price?.display ?? null,
    item.price?.currency ?? null,
    item.shippingPrice === undefined ? null : JSON.stringify(item.shippingPrice),
    item.seller?.name ?? null, item.inStock ?? null, knownPrice,
  ]);
}

The deliberate COALESCE belongs only to the explicitly historical fallback.

Storage	Missing field in a newer observation
Original JSONB	Previous payload is replaced
Current typed columns	Previous value becomes SQL `NULL`
Last-known price	Previous price keeps its original timestamp
No incoming product row	Existing observation continues aging

The composite key identifies a product's requested context. It doesn't identify an individual seller offer. Seller belongs inside the snapshot because the buy box can change sellers.

If your business needs seller-specific inventory, this key isn't enough. That requires a separate offer model and evidence identifying each offer.

Equal timestamps retain the existing row, making dataset replay harmless. The catch is that conflicting observations with identical timestamps also retain the existing row. An observation ledger and an explicit tie-breaking policy become necessary when that distinction matters.

The fallback follows accepted, advancing snapshots. Backfilling older price history needs separate handling too; this upsert intentionally prevents older observations from rewriting current state.

Commit useful rows and account for every request

Read until the dataset returns an empty page. Validate each item independently, and quarantine failures with their dataset offset.

Each write commits independently. If the database connection fails midway, replaying the same dataset safely revisits earlier writes.

That gives you partial progress without a batch-sized transaction. The trade-off is that consumers may see the refresh advance product by product. Use the report as a publication gate when your dashboard needs a consistent batch boundary.

These functions preserve rejected items and compare accepted identities against the manifest:

async function quarantine(raw: unknown, itemOffset: number, reason: unknown) {
  await pool.query(`
    INSERT INTO product_quarantine
      (source_run, item_offset, reason, observation)
    VALUES ($1, $2, $3::jsonb, $4::jsonb)
    ON CONFLICT (source_run, item_offset) DO NOTHING
  `, [sourceRun, itemOffset, JSON.stringify(reason), JSON.stringify(raw ?? null)]);
}

async function ingestDataset() {
  const seen = new Set<string>();
  let quarantined = 0;
  let offset = 0;
  const pageSize = 100;

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

    for (const [index, raw] of items.entries()) {
      const parsed = observationSchema.safeParse(raw);
      if (!parsed.success) {
        await quarantine(raw, offset + index, parsed.error.issues);
        quarantined++;
        continue;
      }
      const key = productKey(parsed.data.url, parsed.data.asin);
      if (!manifest.has(key)) {
        await quarantine(raw, offset + index, {
          reason: 'Identity outside the requested manifest',
        });
        quarantined++;
        continue;
      }
      await upsertObservation(raw, parsed.data);
      seen.add(key);
    }
    offset += items.length;
  }
  return { seen, quarantined };
}

Don't catch database failures and continue as though coverage were complete. A failed write should fail ingestion.

Quarantine is different: it intentionally isolates a bad record while preserving its evidence. The report flags that exclusion, and a later replay can accept the item after you deliberately revise the consumer contract.

An older valid observation still counts as seen. Coverage answers whether the dataset supplied a usable record; the timestamp guard separately decides whether that record advances current state.

Keeping those questions separate prevents an ingestion retry from looking like a new collection failure.

Missing products and missing fields need different alerts

The worked source is Amazon Multi-Domain Product Price & Detail Scraper.

Its omission behavior keeps records free of empty placeholders. That makes the payload sparse; it doesn't make it a partial-update instruction.

An unreachable product is skipped without a dataset row, while processing continues. Unsupported URLs are skipped too. Neither behavior means the product is unavailable.

A missing identity therefore belongs in a coverage alert. An explicit inStock value belongs in the observation. Turning the former into the latter would manufacture another fact.

Keep these signals separate:

Missing requested identity: collection or validation coverage needs attention.
Aging observation: current state has exceeded your freshness budget.
Quarantined record: the consumer contract may have drifted.
Unsuccessful run status: accepted rows belong to an incomplete collection attempt.

This final cell runs ingestion, calculates freshness, and persists a flag your scheduler can alert on:

const freshnessBudgetMs = 3_600_000;

try {
  const { seen, quarantined } = await ingestDataset();
  const currentRows = await pool.query<{
    marketplace: string; asin: string; observed_at: Date;
  }>(`
    SELECT marketplace, asin, observed_at
    FROM product_current WHERE destination_scope = $1
  `, [destinationScope]);

  const observedTimes = new Map(currentRows.rows.map((row) => [
    `${row.marketplace}|${row.asin}`,
    new Date(row.observed_at).getTime(),
  ]));
  const cutoff = Date.now() - freshnessBudgetMs;
  const missing = [...manifest].filter((key) => !seen.has(key));
  const stale = [...manifest].filter((key) => {
    const observed = observedTimes.get(key);
    return observed === undefined || observed < cutoff;
  });
  const report = {
    runStatus, quarantined, missing, stale,
    needsAttention: runStatus !== 'SUCCEEDED' || quarantined > 0 ||
      missing.length > 0 || stale.length > 0,
  };
  await pool.query(`
    INSERT INTO product_ingestion_report
      (source_run, destination_scope, report)
    VALUES ($1, $2, $3::jsonb)
    ON CONFLICT (source_run, destination_scope) DO UPDATE SET
      report = EXCLUDED.report, checked_at = now()
  `, [sourceRun, destinationScope, JSON.stringify(report)]);
} finally {
  await pool.end();
}

Choose the freshness budget for your decisions, then have the scheduler inspect needsAttention. Replaying an old dataset can repair ingestion while correctly leaving freshness red.

The scheduler should also alert when ingestion itself throws. A missing report mustn't look like a clean report.

Unknown field additions pass validation intentionally. Quarantine detects incompatible changes to the fields this consumer checks; it isn't a detector for every possible source change.

At larger catalog sizes, restrict the freshness query to the manifest. This version favors readable plumbing over fetching efficiency.

Before the scheduler owns it

Save the run ID, input manifest, and destination scope together.
Replay the stored dataset using that original context.
Check that absent offer fields clear their typed columns.
Verify that older observations can't advance current state.
Make ingestion failures, quarantine, and coverage reports visible.
Keep fallback prices visibly dated.

A missing price is inconvenient. An invented current offer is worse.

Your generated work email is still a pattern guess

Nimbryxo-studio — Thu, 22 Oct 2026 13:14:26 +0000

A domain can receive mail while your contact's guessed inbox doesn't exist.

You're about to connect email to the CRM's verified-email field. The mapping looks harmless. The uncomfortable part is what that checkbox will tell everyone downstream.

The Actor constructs an address from a name and company domain, then checks whether that domain can receive mail. A domain check does not verify a person's inbox. Your ingestion needs to preserve that distinction.

Start with a finished dataset

Separate execution status from contact evidence before inspecting individual rows. A successful run gets you to the diagnosis stage; it doesn't qualify the contacts for outreach.

Example: this input searches for marketing leaders and waits for the run to finish.

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

const run_input = {
  jobTitles: ['VP of Marketing'],
  maxLeads: 25,
};

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

const run = await client
  .actor('nimbryxo-studio/b2b-lead-discovery-by-job-title-location')
  .call(run_input);

if (run.status !== 'SUCCEEDED') {
  throw new Error(`Inspect run ${run.id}: ${run.status}`);
}

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

This guard is an ingestion policy: stop automatic routing when execution hasn't succeeded. It doesn't attempt to diagnose runtime failures from contact fields.

Once execution passes that gate, ask which stage each person reached.

Which email symptom are you looking at?

Read email alongside company_domain and last_name. A blank address alone can't distinguish the failure paths.

Symptom	Cause supported by the fields	Easy misdiagnosis	Fix and its cost
Empty `email`, empty `company_domain`	Company-domain discovery couldn't resolve a domain	The person has no work inbox	Research the company website; additional research work
Empty `email`, populated `company_domain`	Domain discovery succeeded, but the address was withheld; the check result isn't exposed	The person's mailbox was rejected	Investigate mail capability independently; another check, with uncertainty remaining
Populated `email`	A pattern guess passed the domain check	The address belongs to this person	Send it to separate address verification; extra processing and possible unresolved results
Populated `email`, empty `last_name`	The single-name fallback applies	The short address is malformed	Flag the naming assumption for review; additional review work

Validate the source shape without converting empty strings into missing properties.

const Lead = z.object({
  full_name: z.string(),
  first_name: z.string(),
  last_name: z.string(),
  title: z.string(),
  company_name: z.string(),
  company_domain: z.string(),
  location: z.string(),
  linkedin_url: z.string(),
  email: z.string(),
  scraped_at: z.string(),
});

const people = z.array(Lead).parse(items);

Keep email as a string here. An email-format validator would reject the intentional blanks before you could diagnose them.

This schema checks the record contract. It doesn't authenticate a person, company, or address.

A missing domain sends you back to company research

Address construction depends on finding the company's domain. When that dependency is unresolved, retrying a downstream mailbox check has nothing useful to inspect.

Select that branch while retaining the entire person record.

const missingDomain = people.filter(
  person =>
    person.email === '' &&
    person.company_domain === '',
);

Use linkedin_url to carry the record through research. The missing address shouldn't erase the person you discovered.

Start with the company association, then resolve the website. If company_name is also blank, its absence can reflect search-result text that didn't identify the company clearly.

Don't fill the gap from the person's name alone. A confident company guess can produce a plausible address for the wrong employer. That's a particularly tidy way to make the CRM worse.

The repair belongs upstream: establish the employer and website before attempting contact enrichment.

A present domain narrows the failure, not the proof

A populated domain moves your investigation past company-domain discovery. It still doesn't give you the check's underlying evidence.

Select records whose domain is available but whose address is withheld.

const domainWithoutEmail = people.filter(
  person =>
    person.email === '' &&
    person.company_domain !== '',
);

This combination is consistent with a failed mail-capability check. The output contains no per-record check result that establishes the precise reason.

Keep your downstream diagnosis tentative. Labeling the person's mailbox invalid would claim evidence these fields don't provide.

If you investigate further, attach the new evidence to the record. Don't infer mail capability from the company website loading successfully: serving a website and receiving mail are different capabilities.

A blind rerun is a weak fix. These fields don't establish a transient failure, so repeating the same operation gives you no guaranteed explanation.

A generated address belongs in the verification queue

The construction uses first.last@domain, with first@domain when no last name is known. That fallback describes how the guess was formed; it doesn't make the address invalid.

Route populated addresses for verification, and flag the naming assumption separately.

const guesses = people
  .filter(person => person.email !== '')
  .map(person => ({
    ...person,
    emailEvidence: 'pattern_guess' as const,
    namingPattern:
      person.last_name === ''
        ? 'single_name' as const
        : 'first_last' as const,
    nextStage: 'address_verification' as const,
  }));

These added labels are your pipeline metadata. They aren't source fields or an independent verification result.

The fallback flag helps a reviewer understand the assumption without reverse-engineering the address. Avoid rebuilding it from the name fields just to check whether it “looks right.” Matching the construction pattern would only confirm that the guess resembles a guess.

Keep the proposed address separate from the CRM's confirmed address field. Promote it only when your separate verification process supplies the evidence your team requires.

That process may leave a contact unresolved. Preserve that outcome too. A queue isn't more useful because every row eventually turns green.

Also avoid using the guessed address as the person key. Naming conventions, employers, and addresses can change. Carry the existing profile reference into your ingestion matching rather than replacing it with an enrichment result.

Count people before filtering for contact details

Discovery and contact readiness answer different questions.

Use the diagnostic branches to route work:

Discovered person
  |-- email empty
  |     |-- domain empty: company research
  |     `-- domain present: investigate mail capability
  `-- email populated: separate address verification
        `-- last name empty: flag naming assumption

Calculate stage counts before any CRM filter removes records.

const metrics = {
  discoveredPeople: people.length,
  discoveredDomains: people.filter(
    person => person.company_domain !== '',
  ).length,
  generatedEmailGuesses: guesses.length,
  unresolvedCompanyDomains: missingDomain.length,
  domainsWithoutEmail: domainWithoutEmail.length,
  singleNameGuesses: guesses.filter(
    person => person.namingPattern === 'single_name',
  ).length,
};

const queues = {
  addressVerification: guesses,
  companyResearch: missingDomain,
  mailCapabilityResearch: domainWithoutEmail,
};

The stage totals overlap. A person with a generated address also contributes to discovery counts, so don't sum them into a contact total.

Likewise, compare equivalent stages across runs. A change in generated-address coverage is a reason to inspect the diagnostic branches, not proof that your search found worse people.

The worked example is B2B Lead Discovery by Job Title, Location & Industry. The reusable pattern is to retain what each stage established without borrowing certainty from the next stage.

The catch is operational. Preserving incomplete contacts creates research queues someone must own. Verification also adds work before an address becomes usable.

Dropping every blank address avoids that queue, but changes the population you retain. Your CRM then favors records whose company domains were discoverable and passed the domain check.

That might suit an immediate contact workflow. It's a poor silent default for account research or market mapping, where the person remains relevant without an address.

Before the CRM mapping goes live

Keep this checklist beside the ingestion job:

Gate automatic routing on run status.
Validate records while preserving empty strings.
Retain linkedin_url through every queue.
Separate company research from mail-capability investigation.
Store generated addresses with their evidence level.
Require independent evidence before setting the verified-email field.
Measure discovery before applying contact-readiness filters.

Your CRM's verified checkbox should describe evidence you've collected, not confidence the field name gave you.