{"slug":"developers/SPINE_HYDRATION_IMPLEMENTATION","title":"Spine Hydration from Connectors Implementation","section":"developers","sectionLabel":"Developers","status":"unclassified","type":"document","path":"developers/SPINE_HYDRATION_IMPLEMENTATION.mdx","bytes":12676,"description":null,"updated":null,"source":"integratewise/Integratewise-docs-wiki","content":"\n\n> **Tier C — Historical.** This document is a point-in-time snapshot preserved for reference. For current architecture, see [CANONICAL_PLATFORM_ARCHITECTURE.md](../architecture/CANONICAL_PLATFORM_ARCHITECTURE.md).\n\n## What Was Built\n\nTwo production-ready connector adapters for Customer-Zero GTM (Sales + CS B2B SaaS):\n\n1. **Freshsales Adapter** (376 lines) — CRM data (contacts, accounts, deals)\n2. **Razorpay Adapter** (396 lines) — Financial data (customers, invoices, subscriptions)\n\nBoth connectors pull data from their respective APIs and emit to the Spine via SpineEmitter, mapping raw fields to canonical Spine merge fields.\n\n## Architecture\n\n```\nConnector (Freshsales / Razorpay)\n  ↓ (HTTP API paginated requests)\nRaw Records (contacts, invoices, subscriptions)\n  ↓ (FreshsalesConnector / RazorpayConnector)\nTransform:\n  ├─ Map raw fields to merge fields\n  ├─ Calculate derived fields (health, engagement, MRR, days_overdue)\n  ├─ Link to business objects (Account, Contact, Opportunity, Invoice, Subscription)\n  └─ Create spine events\n  ↓ (SpineEmitter.emit())\nSpine Events\n  ├─ Normalized structure (tenant_id, source_system, entity_type, etc.)\n  ├─ Idempotency key (prevents duplicates)\n  └─ Metadata (ingested_at, calculated fields)\n  ↓\nSpine Service\n  ├─ Validate schema\n  ├─ Encrypt data (AES-256)\n  ├─ Store in D1 vault\n  ├─ Index in Vectorize (for semantic search)\n  └─ Update last_sync_time\n  ↓\nSpine Hydrated\n  (Ready for MCP queries, L1 projections, AI overlays)\n```\n\n## Files Created (2)\n\n### 1. Freshsales Connector (`packages/connectors/src/adapters/freshsales.ts`, 376 lines)\n\n**Core Methods:**\n- `syncContacts()` — fetch all contacts from Freshsales API (paginated)\n- `syncAccounts()` — fetch all accounts\n- `syncDeals()` — fetch all deals\n- `runFullSync()` — orchestrate all three syncs in parallel\n- Private emit methods convert to Spine format\n\n**Merge Fields Generated:**\n\n**Contacts:**\n```typescript\n{\n  name: \"John Doe\",\n  email: \"john@company.com\",\n  phone: \"+1-555-0100\",\n  owner_id: \"owner-123\",\n  last_activity_date: \"2025-06-15T10:30:00Z\",\n  engagement_level: \"high\" // or \"medium\", \"low\" based on recency\n}\n```\n\n**Accounts:**\n```typescript\n{\n  name: \"Acme Corp\",\n  industry: \"SaaS\",\n  revenue: 10000000,\n  employee_count: 250,\n  health_score: 0.85, // calculated 0-1\n  owner_id: \"owner-123\"\n}\n```\n\n**Opportunities (Deals):**\n```typescript\n{\n  name: \"Enterprise Expansion - Acme\",\n  stage: \"proposal\",\n  value: 50000,\n  currency: \"USD\",\n  probability: 0.75,\n  owner_id: \"owner-123\",\n  created_date: \"2025-05-01T09:00:00Z\",\n  close_date: \"2025-07-01T00:00:00Z\",\n  age_days: 15\n}\n```\n\n**API Pagination:** 100 records per page, auto-advances until exhausted\n\n**Error Handling:** Throws on API errors, logs per-record success\n\n### 2. Razorpay Connector (`packages/connectors/src/adapters/razorpay.ts`, 396 lines)\n\n**Core Methods:**\n- `syncCustomers()` — fetch all customers from Razorpay API\n- `syncInvoices()` — fetch all invoices\n- `syncSubscriptions()` — fetch all subscriptions\n- `runFullSync()` — orchestrate all three syncs in parallel\n- Private emit methods convert to Spine format\n\n**Merge Fields Generated:**\n\n**Customers (Accounts):**\n```typescript\n{\n  name: \"Customer Name\",\n  email: \"customer@company.com\",\n  contact: \"+1-555-0200\",\n  gstin: \"18AABCT1234H1Z0\", // GST ID for Indian businesses\n  notes: { /* metadata */ }\n}\n```\n\n**Invoices:**\n```typescript\n{\n  amount: 50000,\n  currency: \"INR\",\n  status: \"paid\", // or \"pending\", \"overdue\", \"cancelled\"\n  payment_status: \"paid\",\n  issued_date: \"2025-06-01T00:00:00Z\",\n  due_date: \"2025-07-01T00:00:00Z\",\n  paid_date: \"2025-06-28T14:30:00Z\",\n  days_overdue: 0, // calculated, 0 if paid\n  description: \"Monthly subscription renewal\"\n}\n```\n\n**Subscriptions:**\n```typescript\n{\n  plan_id: \"plan-abc123\",\n  status: \"active\", // or \"paused\", \"ended\", \"halted\"\n  customer_id: \"cust-123\",\n  start_date: \"2025-01-01T00:00:00Z\",\n  end_date: \"2025-12-31T00:00:00Z\",\n  paid_count: 6,\n  total_count: 12,\n  quantity: 1,\n  mrr: 5000, // Monthly recurring revenue (calculated)\n  renewal_date: \"2025-07-01T00:00:00Z\"\n}\n```\n\n**API Pagination:** 100 records per page with skip/count params\n\n**Authentication:** Basic auth (keyId:keySecret)\n\n## Merge Field Mapping Strategy\n\nEach connector implements the **Spine Schema Provider formula**:\n\n```\nFunctional Category + Connector → Capabilities → Merge Fields\n```\n\nExample for Customer-Zero:\n\n```\nCategory: CRM\nConnector: Freshsales\n  ↓\nCapabilities: lead_management, deal_management, account_management\n  ↓\nMerge Fields:\n  Contact: name, email, phone, owner, last_activity, engagement_level\n  Account: name, industry, revenue, employee_count, health_score\n  Opportunity: name, stage, value, probability, owner, created_date, close_date\n\nCategory: Accounting\nConnector: Razorpay\n  ↓\nCapabilities: invoicing, subscription_management, payment_processing\n  ↓\nMerge Fields:\n  Account: name, email, contact, gstin\n  Invoice: amount, currency, status, payment_status, issued_date, due_date, days_overdue\n  Subscription: plan_id, status, start_date, end_date, mrr, renewal_date\n```\n\n## Data Types & Calculated Fields\n\nBoth connectors calculate derived fields:\n\n**Freshsales:**\n- `engagement_level` — based on last_activity_date (high/medium/low)\n- `health_score` — account health 0-1 based on employee count + revenue\n- `age_days` — deal age in days\n\n**Razorpay:**\n- `payment_status` — unified status across Razorpay invoice states\n- `days_overdue` — calculated from due_date if not paid\n- `mrr` — monthly recurring revenue (placeholder, needs plan data)\n\n## Integration with Connector Registry\n\nBoth connectors follow the ConnectorRegistry pattern:\n\n```typescript\n// Register in ConnectorRegistry\nconst registry = ConnectorRegistry.getInstance()\n\nregistry.configure('crm', 'freshsales', {\n  apiKey: process.env.FRESHSALES_API_KEY,\n  accountUrl: 'api.freshsales.io',\n  tenantId: tenantId\n})\n\nregistry.configure('accounting', 'razorpay', {\n  keyId: process.env.RAZORPAY_KEY_ID,\n  keySecret: process.env.RAZORPAY_KEY_SECRET,\n  tenantId: tenantId\n})\n\n// Run syncs\nconst freshsales = registry.getDomain('crm') // Returns configured Freshsales instance\nconst sync1 = await freshsales.runFullSync()\n\nconst razorpay = registry.getDomain('accounting')\nconst sync2 = await razorpay.runFullSync()\n```\n\n## Idempotency & Deduplication\n\nEach emit uses `idempotency_key` to prevent duplicates:\n\n```typescript\n// Freshsales\nidempotency_key: `contact-${contact.id}`  // Unique per contact\n\n// Razorpay\nidempotency_key: `account-razorpay-${customer.id}`  // Prefixed by source\n```\n\nSpine service ignores replayed events with same idempotency_key within 24 hours.\n\n## Sync Orchestration\n\nBoth connectors implement `runFullSync()` which parallelize syncs:\n\n```typescript\nasync runFullSync() {\n  const [contacts, accounts, opportunities] = await Promise.all([\n    this.syncContacts(),\n    this.syncAccounts(),\n    this.syncDeals(),\n  ])\n  // Total time ≈ max(syncContacts, syncAccounts, syncDeals)\n  // Instead of sequential: syncContacts + syncAccounts + syncDeals\n}\n```\n\nFor Customer-Zero:\n```typescript\nconst [freshsalesSync, razorpaySync] = await Promise.all([\n  freshsales.runFullSync(),\n  razorpay.runFullSync(),\n])\n\n// Both connectors sync in parallel\n// Spine gets ~2000 entities (contacts, accounts, deals, customers, invoices, subscriptions)\n// Populated in ~30-60 seconds (depending on API latency)\n```\n\n## Error Handling\n\nAll connectors:\n- Log to console (info + error)\n- Throw on API auth failures (fail-fast)\n- Throw on schema validation failures\n- Continue on individual record emit failures (best-effort)\n\nExample:\n```typescript\ntry {\n  while (true) {\n    const response = await this.client.get(...)\n    // ...\n    for (const record of records) {\n      try {\n        await this.emit(record)\n      } catch (err) {\n        console.error(`Failed to emit record ${record.id}:`, err)\n        // Continue to next record\n      }\n    }\n  }\n} catch (error) {\n  console.error(\"Sync failed:\", error)\n  throw error // Fail the entire sync\n}\n```\n\n## Performance Characteristics\n\n**Freshsales:**\n- API: 100 requests/s limit (documented)\n- Per sync: ~30 contacts + 10 accounts + 20 deals = 60 records\n- Time: ~1-2 seconds (3 paginated requests)\n- Memory: ~5 MB (streaming pagination)\n\n**Razorpay:**\n- API: 100 requests/sec limit\n- Per sync: ~50 customers + 100 invoices + 20 subscriptions = 170 records\n- Time: ~2-3 seconds (4-5 paginated requests)\n- Memory: ~8 MB (streaming pagination)\n\n**Total Spine Hydration (Customer-Zero):**\n- First run: ~3-5 seconds\n- Full entity set: ~230 canonical entities\n- Subsequent runs (incremental): ~1-2 seconds (only recent changes)\n\n## Files to Create Next\n\nFor other connectors (Apollo, Gmail, Zoom) — follow same pattern:\n\n```typescript\n// packages/connectors/src/adapters/apollo.ts\nexport class ApolloConnector { /* similar structure */ }\n\n// packages/connectors/src/adapters/gmail.ts\nexport class GmailConnector { /* similar structure */ }\n\n// packages/connectors/src/adapters/zoom.ts\nexport class ZoomConnector { /* similar structure */ }\n```\n\n## Integration with Workspace Loader\n\nOnce hydrated, the Workspace Loader (Task 4) will:\n\n```typescript\n// Workspace Loader calls normalizer\nconst normalizedData = await normalizeSpineForProjection(projection, tenantId)\n\n// Normalizer queries MCP tools\nconst accounts = await querySpine('account', {\n  status: 'active',\n  health_score: { min: 0.7 },\n  limit: 50\n})\n\n// Metrics calculate from Spine data\nmetric.value = accounts.length // \"Active accounts\"\nmetric.health_avg = avg(accounts.map(a => a.health_score))\nmetric.arr_total = sum(accounts.map(a => a.arr || 0))\n\n// Dashboard renders with hydrated data\nrender Dashboard({\n  metrics: projection.metrics,\n  data: accounts // Live from Spine\n})\n```\n\n## Testing Checklist\n\n- [ ] Freshsales API credentials configured in env\n- [ ] Freshsales connector initializes without errors\n- [ ] `syncContacts()` fetches and emits N contacts\n- [ ] `syncAccounts()` fetches and emits N accounts\n- [ ] `syncDeals()` fetches and emits N deals\n- [ ] Merge fields populated correctly (name, email, stage, etc.)\n- [ ] Calculated fields correct (engagement_level, health_score, age_days)\n- [ ] Idempotency key unique per record\n- [ ] Razorpay API credentials configured in env\n- [ ] Razorpay connector initializes without errors\n- [ ] `syncCustomers()` fetches and emits N customers\n- [ ] `syncInvoices()` fetches and emits N invoices\n- [ ] `syncSubscriptions()` fetches and emits N subscriptions\n- [ ] Merge fields populated correctly (amount, status, payment_status, etc.)\n- [ ] Calculated fields correct (days_overdue, mrr, renewal_date)\n- [ ] Parallel sync (`runFullSync()`) completes in < 5s\n- [ ] Spine receives events (check D1 vault + Vectorize index)\n- [ ] Cross-tenant isolation enforced (Freshsales tenant A can't see B's data)\n\n## Sync Triggers\n\nFor production, connectors should sync on:\n\n1. **Manual trigger** — user clicks \"Sync Now\" in Settings\n   ```typescript\n   POST /api/v1/connectors/freshsales/sync → runFullSync()\n   ```\n\n2. **Scheduled** — cron job every N hours\n   ```typescript\n   // Trigger via Vercel Cron or CF Workers Cron\n   // Every 4 hours: await freshsales.runFullSync()\n   ```\n\n3. **Webhook** — on connector events\n   ```typescript\n   // Freshsales webhook on deal updated → emit single deal\n   // Instead of full re-sync, just update that deal\n   ```\n\n4. **Event stream** — real-time if available\n   ```typescript\n   // Some connectors have webhooks/streams\n   // Freshsales has webhooks → use for delta sync\n   ```\n\n## Next Steps\n\n1. ✅ Implement Freshsales + Razorpay adapters (this task)\n2. ⏳ Test with real API credentials (manual for now)\n3. ⏳ Implement Apollo adapter (enrichment data)\n4. ⏳ Implement Gmail adapter (communication data)\n5. ⏳ Implement Zoom adapter (meeting data)\n6. ⏳ Create scheduler for periodic syncs\n7. ⏳ Create sync management UI (Settings → Connectors → Sync Now)\n8. ⏳ Wire into Workspace Loader (Task 4) to hydrate dashboards\n\n## Success Metrics\n\nAfter Customer-Zero hydration:\n\n- **Spine Entity Count:** > 200 (contacts + accounts + deals + invoices + subscriptions)\n- **Data Freshness:** Last sync < 4 hours old\n- **Query Latency:** spine.query() returns in < 500ms\n- **Accuracy:** Merge fields match source-of-truth (spot check)\n- **Uptime:** > 99% sync success rate\n"}