Data Sync Jobs skill for Codex

Done well means: running the job twice changes nothing the second time, a crash halfway loses nothing on the next run, every record in the target can be traced to its source ID, and a reconciliation report proves the two sides agree. Duplicates and silent drift are the two failure modes; design against both from the first line.

Decide the shape first

Answer these before writing code; put the answers in the job’s README.

Question	Default answer
Which side is the source of truth, per field?	One owner per field. Two-way sync only when a field genuinely has two owners
How do records match across systems?	Store the source’s immutable ID in the target (external ID field or mapping table). Never match on name or email alone
How are changes found?	Incremental cursor on updated_at (or the API’s change feed); full scan only for small sets or reconciliation
How are deletes found?	Change feed with tombstones if the API has one; otherwise periodic full ID-set comparison
What happens on conflict?	Field owner wins; for true two-way fields, last-writer-wins by source timestamp, logged
How often?	As slow as the business tolerates. Every 15 min is usually enough; real-time needs webhooks

If the user needs real-time and the source offers webhooks, receive them ($webhook-receivers) and keep this job as the nightly reconciler that catches what the webhooks missed.

Workflow

- [ ] 1. Read AGENTS.md; find where jobs, secrets and schedules live in this repo
- [ ] 2. Read both APIs' docs: pagination, filters on updated time, rate limits, upsert support
- [ ] 3. Write the field mapping table (source field, target field, owner, transform)
- [ ] 4. Implement fetch -> transform -> upsert with a persisted cursor
- [ ] 5. Add --dry-run that prints planned creates/updates/deletes and writes nothing
- [ ] 6. Run dry-run against real data (read-only credentials if possible)
- [ ] 7. First real run on a small slice; then full backfill; then schedule
- [ ] 8. Add the reconciliation check and alert on mismatch

Fetching needs network access; a sandboxed Codex session may need approval for it. Use read-only API keys while developing, and get explicit confirmation before the first run that writes to, or deletes from, a production system.

Incremental cursor done right

OVERLAP = timedelta(minutes=5)

def run(state: State, dry_run: bool) -> Stats:
    since = state.cursor - OVERLAP if state.cursor else None   # overlap catches clock skew
    high_water = state.cursor
    stats = Stats()
    for page in source.list_changed(updated_since=since, page_size=200):
        for rec in page:
            row = transform(rec)                     # pure function, unit-tested
            action = target.plan_upsert(external_id=rec["id"], row=row)
            stats.count(action)                      # create | update | unchanged
            if not dry_run and action != "unchanged":
                target.upsert(external_id=rec["id"], row=row)
            high_water = max(high_water or rec["updated_at"], rec["updated_at"])
        if not dry_run:
            state.save_cursor(high_water)            # checkpoint per page, after writes
    return stats

• Advance the cursor only after the page’s writes succeed. Saving it first loses records on a crash.
• Use the source’s updated_at, never your clock. Overlap the window by a few minutes; idempotent upserts make the re-read harmless.
• Sort by (updated_at, id) when the API allows it; records sharing a timestamp across a page boundary are otherwise skipped.
• Offset pagination over a changing dataset skips and repeats rows. Prefer cursor or keyset pagination; if offset is all you have, keep pages small and rely on overlap plus reconciliation.
• Store state (cursor, last run, counts) in a durable place: a DB table, not a local file on an ephemeral runner.

Idempotent writes

Target	Upsert mechanism
Postgres	INSERT ... ON CONFLICT (source_system, source_id) DO UPDATE SET ... WHERE crm_contacts.row_hash IS DISTINCT FROM excluded.row_hash
Salesforce	Upsert on an External ID field
HubSpot	Batch upsert by a unique property (idProperty)
Airtable	performUpsert with fieldsToMergeOn
API without upsert	Look up via mapping table (source_id -> target_id), then create or update

• Compare a hash of mapped fields before updating; skipping unchanged rows avoids rate limit burn and false “modified” timestamps that cause two-way sync ping-pong.
• For APIs without upsert, write the mapping row in the same step as the create; a crash between “created” and “mapped” is how duplicates appear. Recover by searching the target for the external ID before creating.

Deletes and two-way sync

• Never hard-delete in the target on the first sign of absence. Mark deleted_at (soft delete), and hard-delete only after the user approves the policy.
• Full ID comparison: fetch all source IDs (IDs only, cheap), diff against the mapping table, and refuse to act if more than a threshold (e.g. 5 %) would be deleted. A broken filter or an expired token often looks like “everything was deleted”.
• Two-way sync: tag each write with its origin (last_synced_from) and skip echo updates caused by your own previous write. Without this, two jobs bounce the same change forever.

Rate limits and failures

• Respect Retry-After on 429; back off exponentially with jitter on 5xx and timeouts. Do not retry 400/422; log the record and continue (dead-letter list).
• Use bulk endpoints when they exist; batch by the API’s documented maximum.
• One poison record must not stop the run. Collect failures with the source ID and error, report them, and keep the cursor moving only past records that succeeded or were dead-lettered deliberately.
• A single instance at a time: take a lock (DB advisory lock, or the scheduler’s concurrency setting) so overlapping runs do not race.

Reconciliation

Run after each full sync and on a schedule (daily or weekly):

-- Counts and a fingerprint per day bucket; compare with the same query on the source export
SELECT date_trunc('day', source_updated_at) AS day,
       count(*) AS n,
       md5(string_agg(source_id || ':' || row_hash, ',' ORDER BY source_id)) AS fp
FROM crm_contacts
WHERE deleted_at IS NULL
GROUP BY 1 ORDER BY 1;

Mismatched buckets narrow the search to one day of records; resync just that window.

Gotchas

• Time zones: some APIs filter updated_since in the account’s local zone or round to the minute. Test the filter with a record you just edited.
• Field renames or picklist changes in the SaaS tool break mappings silently; validate the source schema at the start of each run and fail loudly on unknown required fields.
• API search endpoints are often eventually consistent; do not use them to confirm a write you just made.
• Email and phone normalisation (case, whitespace, E.164) before matching, but keep the external ID as the real key.
• PII in logs: log IDs and counts, not full records.

Verification

• Run twice in a row: the second run reports zero creates and zero updates.
• Kill the job mid-run, restart, and confirm no duplicates (count by external ID with GROUP BY source_id HAVING count(*) > 1).
• Edit one record in the source and confirm exactly one update in the target on the next run.
• Reconciliation reports zero mismatched buckets.

Run report

**Sync:** HubSpot contacts -> Postgres crm_contacts (one-way, HubSpot owns all fields)
**Window:** updated since 2026-10-27T06:00Z (cursor - 5 min overlap)
**Result:** 1,204 fetched, 37 created, 112 updated, 1,055 unchanged, 2 failed (dead-lettered)
**Deletes:** 3 soft-deleted (0.02 % of 14,880, under 5 % guard)
**Reconciliation:** 30/30 day buckets match
**Failures:** 50123 (422 invalid email), 50991 (422 invalid email)