Navan — Data Sync
Overview
This skill provides production-grade sync strategies for Navan data. The two primary tables have fundamentally different sync models: BOOKING requires weekly full-refresh with merge-upsert logic (every record is re-imported, keyed by UUID), while TRANSACTION is incremental and append-only. Real-time use cases require webhook callbacks for event-driven processing. This skill covers all three tiers — scheduled full-refresh, incremental watermark-based sync, and real-time webhooks — along with Airbyte connector configuration and idempotent SQL upsert patterns.
Prerequisites
- Navan account with OAuth 2.0 API credentials (see
navan-install-auth)
- Destination warehouse (Snowflake, BigQuery, PostgreSQL, or Redshift)
- For managed sync: Airbyte instance (Cloud or OSS) with source-navan v0.0.42+
- For webhooks: publicly accessible HTTPS endpoint for callbacks
- Node.js 18+ or Python 3.8+
- Environment variables:
NAVAN_CLIENT_ID, NAVAN_CLIENT_SECRET, NAVAN_BASE_URL
Instructions
Step 1: Full-Refresh Sync for BOOKING Table
The BOOKING table is re-imported weekly by Navan. Every record is refreshed, so your sync must use merge-upsert logic to avoid duplicates while capturing updates.
const tokenRes = await fetch(`${process.env.NAVAN_BASE_URL}/ta-auth/oauth/token`, {
method: 'POST',
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
body: new URLSearchParams({
grant_type: 'client_credentials',
client_id: process.env.NAVAN_CLIENT_ID!,
client_secret: process.env.NAVAN_CLIENT_SECRET!,
}),
});
const { access_token } = await tokenRes.json();
const headers = { Authorization: `Bearer ${access_token}` };
let allBookings: any[] = [];
let page = 0;
const size = 50;
while (true) {
const res = await fetch(
`${process.env.NAVAN_BASE_URL}/v1/bookings?page=${page}&size=${size}`,
{ headers }
);
const { data } = await res.json();
if (!data || !data.length) break;
allBookings.push(...data);
if (data.length < size) break;
page++;
}
console.log(`Extracted ${allBookings.length} bookings for full refresh`);
SQL merge-upsert pattern (PostgreSQL):
CREATE TABLE IF NOT EXISTS navan_booking_staging (
uuid TEXT PRIMARY KEY,
traveler_email TEXT,
origin TEXT,
destination TEXT,
start_date DATE,
end_date DATE,
total_cost NUMERIC(12,2),
currency TEXT DEFAULT 'USD',
department TEXT,
cost_center TEXT,
status TEXT,
in_policy BOOLEAN,
created_at TIMESTAMPTZ,
updated_at TIMESTAMPTZ,
synced_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO navan_booking AS b
SELECT * FROM navan_booking_staging s
ON CONFLICT (uuid) DO UPDATE SET
traveler_email = EXCLUDED.traveler_email,
origin = EXCLUDED.origin,
destination = EXCLUDED.destination,
start_date = EXCLUDED.start_date,
end_date = EXCLUDED.end_date,
total_cost = EXCLUDED.total_cost,
status = EXCLUDED.status,
in_policy = EXCLUDED.in_policy,
updated_at = EXCLUDED.updated_at,
synced_at = NOW()
WHERE b.updated_at < EXCLUDED.updated_at;
Step 2: Incremental Sync for TRANSACTION Table
TRANSACTION data is append-only. Use watermark-based sync to pull only new records.
interface SyncState {
lastSyncDate: string;
lastTransactionId: string;
}
async function loadSyncState(): Promise<SyncState> {
const fs = await import('fs');
try {
return JSON.parse(fs.readFileSync('.navan-sync-state.json', 'utf-8'));
} catch {
return { lastSyncDate: '2025-01-01', lastTransactionId: '' };
}
}
async function saveSyncState(state: SyncState) {
const fs = await import('fs');
fs.writeFileSync('.navan-sync-state.json', JSON.stringify(state, null, 2));
}
const state = await loadSyncState();
today = ().().()[];
txnRes = (
+
,
{ headers }
);
{ : transactions } = txnRes.();
newTxns = transactions.(
t. > state.
);
.();
(newTxns. > ) {
({
: today,
: newTxns[newTxns. - ].,
});
}
Step 3: Webhook Endpoint for Real-Time Events
import { createServer } from 'http';
import { createHmac } from 'crypto';
const server = createServer(async (req, res) => {
if (req.method !== 'POST' || req.url !== '/navan/webhook') {
res.writeHead(404);
res.end();
return;
}
const chunks: Buffer[] = [];
for await (const chunk of req) chunks.push(chunk as Buffer);
const body = Buffer.concat(chunks).toString();
const signature = req.headers['x-navan-signature'] as string;
const expected = createHmac('sha256', process.env.NAVAN_WEBHOOK_SECRET!)
.update(body)
.digest('hex');
if (signature !== expected) {
console.();
res.();
res.();
;
}
event = .(body);
.();
(event.) {
:
.();
;
:
.();
;
:
.();
;
:
.();
;
:
.();
;
:
.();
}
res.();
res.();
});
server.(, .());
Step 4: Airbyte Connector Sync Configuration
source:
sourceDefinitionId: source-navan
connectionConfiguration:
client_id: "${NAVAN_CLIENT_ID}"
client_secret: "${NAVAN_CLIENT_SECRET}"
syncCatalog:
streams:
- stream:
name: bookings
jsonSchema: {}
config:
syncMode: full_refresh
destinationSyncMode: overwrite
schedule:
scheduleType: cron
cronExpression: "0 2 * * 0"
Step 5: Sync Monitoring and Alerting
interface SyncMetrics {
tableName: string;
lastSyncTime: Date;
recordCount: number;
expectedFrequency: string;
isStale: boolean;
}
async function checkSyncHealth(): Promise<SyncMetrics[]> {
const state = await loadSyncState();
const lastSync = new Date(state.lastSyncDate);
const hoursSinceSync = (Date.now() - lastSync.getTime()) / (1000 * 60 * 60);
return [
{
tableName: 'BOOKING',
lastSyncTime: lastSync,
recordCount: 0,
expectedFrequency: 'weekly',
isStale: hoursSinceSync > 7 * 24 + 6,
},
{
tableName: 'TRANSACTION',
lastSyncTime: lastSync,
recordCount: ,
: ,
: hoursSinceSync > ,
},
];
}
health = ();
health.( {
status = m. ? : ;
.();
});
Step 6: Idempotent Load Pattern
async function idempotentLoad(records: any[], tableName: string) {
const batchId = `${tableName}-${new Date().toISOString()}`;
console.log(`Loading ${records.length} records to ${tableName}_staging (batch: ${batchId})`);
console.log(`Merging ${tableName}_staging -> ${tableName}`);
console.log(`Batch ${batchId} complete: ${records.length} records processed`);
return { batchId, recordCount: records.length, status: 'complete' };
}
Output
Successful execution produces:
- Full-refresh BOOKING sync with merge-upsert deduplication
- Incremental TRANSACTION sync with watermark state tracking
- Webhook endpoint for real-time event processing
- Configured Airbyte connector with production-ready sync schedule
- Sync health monitoring with staleness alerting
Error Handling
| Error | HTTP Code | Cause | Solution |
|---|
| Unauthorized | 401 | Expired or invalid bearer token | Re-authenticate via POST /ta-auth/oauth/token |
| Rate Limited | 429 | Too many API requests | Use exponential backoff; increase sync interval |
| Timeout | 504 | Full refresh too large | Chunk by date range (30-day windows) |
| Webhook Sig Invalid | 401 | Tampered or replayed event | Verify NAVAN_WEBHOOK_SECRET; check clock skew |
| Duplicate Records | N/A | Missing UUID dedup in BOOKING sync | Apply merge-upsert with ON CONFLICT (uuid) |
| Sync Drift | N/A | Missed incremental window | Fall back to full refresh; reset watermark |
Examples
Python — Incremental TRANSACTION sync with watermark:
import requests
import json
import os
from datetime import datetime
base_url = os.environ.get('NAVAN_BASE_URL', 'https://api.navan.com')
auth = requests.post(f'{base_url}/ta-auth/oauth/token', data={
'grant_type': 'client_credentials',
'client_id': os.environ['NAVAN_CLIENT_ID'],
'client_secret': os.environ['NAVAN_CLIENT_SECRET'],
})
headers = {'Authorization': f'Bearer {auth.json()["access_token"]}'}
try:
with open('.navan-sync-state.json') as f:
state = json.load(f)
except FileNotFoundError:
state = {'last_sync_date': '2025-01-01'}
today = datetime.now().strftime('%Y-%m-%d')
resp = requests.get(
f'{base_url}/v1/bookings',
params={'createdFrom': state['last_sync_date'], 'createdTo': today, 'page': 0, 'size': 50},
headers=headers,
).json()
txns = resp['data']
print(f'Fetched {len(txns)} records since {state["last_sync_date"]}')
with open(, ) f:
json.dump({: today}, f)
Resources
Next Steps
After configuring data sync, proceed to navan-observability for pipeline monitoring or navan-performance-tuning for optimizing large-volume syncs.
