Your upsert can preserve an offer that never existed

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.