Your inactive listing survived a successful upsert

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.