Your partners array can reverse the buyer relationship

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: