{"slug":"business/DATA_ACCESS_PATTERNS","title":"Data Access Patterns — When to Use What","section":"business","sectionLabel":"Business","status":"unclassified","type":"document","path":"business/DATA_ACCESS_PATTERNS.mdx","bytes":21102,"description":null,"updated":null,"source":"integratewise/Integratewise-docs-wiki","content":"\n> Definitive guide: Spine DB direct vs Gateway API vs Cloudflare operational layer.\n> This is an architecture decision. Follow it everywhere.\n\n---\n\n## Current Surface-Binding Doctrine\n\n- Operational Workbench = the primary work surface; left sidebar nav; Twin sidebar pops out when needed.\n- Twin Workbench = full AI ecosystem surface (skills, knowledge, agents, prompts, conversational library).\n- Governance is embedded at every layer — not a separate workbench.\n- OpenWebUI is not the customer-facing product shell.\n- Twin / OpenWebUI runtime surfaces use the canonical Forest + Paper system language.\n- Midnight Executive is investor-deck / fundraising language only.\n\nUse older L1/L2 / workspace / knowledge-UI language in this file through that boundary if any historical phrasing remains.\n\n---\n\n## The decoupled architectural layers:\n\n```\n┌─────────────────────────────────────────────────────────────────┐\n│  FRONTEND (React App)                                           │\n│                                                                 │\n│  Reads: Edge Low-Latency Cache (D1, KV)                         │\n│  Writes: Gateway API (validation + pipeline)                    │\n│  Auth: Spine Gateway Auth (JWT)                                 │\n│                                                                 │\n├─────────────────────────────────────────────────────────────────┤\n│  GATEWAY (Cloudflare Worker)                                    │\n│                                                                 │\n│  Writes that need validation, pipeline, or multi-service        │\n│  External API calls (OAuth, connector sync)                     │\n│  Webhook ingestion & Rate limiting                              │\n│                                                                 │\n├─────────────────────────────────────────────────────────────────┤\n│  ACTIVE MEMORY PLANE (D1, KV, Queues, Durable Objects)           │\n│                                                                 │\n│  Sandboxed Edge operational database layer                      │\n│  Low-latency read/write for internal Twin reasoning loops       │\n│                                                                 │\n├─────────────────────────────────────────────────────────────────┤\n│  CORE MEMORY SUBSTRATE (Supabase DB + Filesystem Vault)         │\n│                                                                 │\n│  Fortress Database of Record (relational & pgvector)            │\n│  Mirrored and persistent storage synchronized from Edge         │\n│  Accessed by external clients via the Continuity Bridge MCP     │\n└─────────────────────────────────────────────────────────────────┘\n```\n\n---\n\n## Rule 1: READS — Frontend queries Spine DB directly\n\n**Never route reads through Gateway for workspace views.**\n\nThe Spine DB client is already initialized in the frontend (`utils/spineDb/client.ts`). RLS enforces tenant isolation at the database level. The JWT from Spine DB Auth carries the tenant_id. Every query is automatically scoped.\n\n| What the view needs  | Query                                                                      | Goes through                      |\n| -------------------- | -------------------------------------------------------------------------- | --------------------------------- |\n| Account list         | `spineDb.from(\"entities\").select(\"*\").eq(\"entity_type\", \"account\")`        | Spine DB direct                   |\n| Tasks for user       | `spineDb.from(\"entities\").select(\"*\").eq(\"entity_type\", \"task_item\")`      | Spine DB direct                   |\n| Contacts             | `spineDb.from(\"entities\").select(\"*\").eq(\"entity_type\", \"contact\")`        | Spine DB direct                   |\n| Calendar events      | `spineDb.from(\"entities\").select(\"*\").eq(\"entity_type\", \"calendar_event\")` | Spine DB direct                   |\n| Signals/alerts       | `spineDb.from(\"signals\").select(\"*\").eq(\"status\", \"active\")`               | Spine DB direct                   |\n| Insights             | `spineDb.from(\"entities\").select(\"*\").eq(\"entity_type\", \"insight\")`        | Spine DB direct                   |\n| Metrics/KPIs         | `spineDb.from(\"entities\").select(\"*\").eq(\"entity_type\", \"metric\")`         | Spine DB direct                   |\n| Goals                | `spineDb.from(\"goals\").select(\"*\")`                                        | Spine DB direct                   |\n| Memories             | `spineDb.from(\"consolidated_memories\").select(\"*\")`                        | Spine DB direct                   |\n| Documents            | `spineDb.from(\"context_extractions\").select(\"*\")`                          | Spine DB direct                   |\n| Audit trail          | `spineDb.from(\"governance_audit_log\").select(\"*\")`                         | Spine DB direct                   |\n| Dashboard projection | `spineDb.rpc(\"get_department_projection\", { dept: \"finance\" })`            | Spine DB direct (stored function) |\n\n**Why:** Zero latency overhead. No Worker cold starts. No Gateway routing. RLS handles security. Spine DB JS client handles connection pooling and caching. One round-trip to the database, not three hops through Workers.\n\n**Realtime updates:** Use Spine DB Realtime subscriptions for live data:\n\n```typescript\nspineDb\n  .channel(\"signals\")\n  .on(\"postgres_changes\", { event: \"INSERT\", schema: \"spine\", table: \"signals\" }, handleNewSignal)\n  .subscribe();\n```\n\nThis replaces the need for WebSocket through Workflow Worker for most UI updates.\n\n---\n\n## Rule 2: WRITES — Frontend calls Gateway API\n\n**All writes go through Gateway. Never write to Spine DB directly from frontend.**\n\nWrites need validation, pipeline processing, audit logging, and downstream effects. The Gateway orchestrates this.\n\n| What the user does    | API call                               | Why Gateway                                       |\n| --------------------- | -------------------------------------- | ------------------------------------------------- |\n| Create a task         | `POST /api/v1/workspace/entities`      | Needs validation + Spine write + audit            |\n| Log a note            | `POST /api/v1/workspace/entities`      | Needs entity linking + audit                      |\n| Approve a triage item | `POST /api/v1/triage/approve`          | Needs Govern check + memory write + audit         |\n| Merge duplicates      | `POST /api/v1/identity/merge`          | Needs Govern approval + multi-table write + audit |\n| Connect a tool        | `POST /api/v1/connectors/connect`      | Needs OAuth flow + connector service              |\n| Execute an action     | `POST /api/v1/act/execute`             | Needs Govern gate + multi-service orchestration   |\n| Update entity fields  | `PATCH /api/v1/workspace/entities/:id` | Needs validation + version tracking + audit       |\n| Disconnect connector  | `DELETE /api/v1/connectors/:id`        | Needs Govern approval + cleanup                   |\n\n**Why:** Writes have side effects. A task creation might trigger a Twin evaluation. An approval might write to multiple tables. A merge affects entity relationships. The Gateway ensures all side effects happen correctly, atomically, and with audit trail.\n\n**Exception:** User profile updates and auth operations go through Spine DB Auth directly (already implemented in auth-provider.tsx).\n\n---\n\n## Rule 3: CLOUDFLARE OPERATIONAL — The Active Memory Plane\n\nCloudflare D1 (`spine_vault`), KV, Queues, R2, and AI Search represent the active operational **Memory Plane** of IntegrateWise. The Twin Workspace reads Zod-validated context edge-locally from this plane.\n\nFor **promoted tenants**, Supabase (fortress) and the Filesystem Vault serve as the underlying **Core Memory Substrate** for permanent synchronization, external client access (via MCP), and auditability. Supabase is **usage-allocated, not default**: initial/edge-only tenants run entirely on D1 + KV + R2 + AI Search (D1 is their record of truth, R2 spine-backup is the replay path) and have **no Supabase footprint**. Supabase is added only when a tenant is promoted (data volume / retention / compliance / plan). Internal reasoning loops are always sandboxed from Supabase. All writes go Cloudflare Worker first; for promoted tenants they also update Supabase via the CF MCP Worker and propagate back. See `docs/tech/MCP_ADK_ROUTING_SPINE_MEMORY_ARCHITECTURE.md` §7.7.\n\n### The Three Data Channels:\n\n- **Channel 1 — Via Connector (Pull-Based)**: Source Tool (e.g. HubSpot) $\\rightarrow$ Nango OAuth $\\rightarrow$ Schema Ingress validation $\\rightarrow$ Loader $\\rightarrow$ Normalizer $\\rightarrow$ D1 `spine_vault` (AES-256-GCM encrypted).\n- **Channel 2A — Via MCP (API Wrapper)**: Raw unstructured push data transits loader and normalizer before spine insertion.\n- **Channel 2B — Via MCP (Structured)**: EXTERNAL clients (Claude, ChatGPT, 3rd-party agents) query `mcp.integratewise.ai` directly to read D1 `spine_vault` and write via `memory.propose` (bypassing normalizers). The Twin/Agent Zero does NOT use this path — it reads via the IW Continuity Bridge (`continuity-tool-server → pipeline`); MCP is the external door.\n- **Channel 3 — Knowledge Ingestion**: Document uploads/Slack threads run through the Knowledge service, creating `context_extractions` in R2 and AI Search. This enriches Entity 360 but does not write to Spine.\n\n### Cloudflare Queues — Pipeline message processing\n\n| Queue              | Purpose                        | Producer                | Consumer      |\n| ------------------ | ------------------------------ | ----------------------- | ------------- |\n| PIPELINE_QUEUE     | 8-stage normalization          | Loader, Webhook Ingress | Normalizer    |\n| KNOWLEDGE_QUEUE    | Context extraction + embedding | Normalizer (S8)         | Knowledge     |\n| INTELLIGENCE_QUEUE | Signal processing              | Normalizer (S8)         | Intelligence  |\n| SIGNAL_QUEUE       | Signal broadcasting            | Intelligence            | Workflow      |\n| DLQ_QUEUE          | Dead letter (failed messages)  | Any stage on max retry  | Manual review |\n\n**Integration best practices for Queues:**\n\n- Idempotency: Every message has a content-hash dedup key. Same payload = same hash = skip processing.\n- Retry: 3 retries with exponential backoff (30s, 60s, 90s) before DLQ.\n- Message persistence: Failed messages go to DLQ_QUEUE with full context (stage, error, payload, trace_id, timestamp).\n- Ordering: Not guaranteed. Each message is independent. Use version numbers for conflict resolution.\n- Batch size: Consumer processes up to 10 messages per batch.\n\n### Cloudflare KV — Edge cache (NOT source of truth)\n\n| KV Namespace     | What it caches                            | TTL  | Source of truth              |\n| ---------------- | ----------------------------------------- | ---- | ---------------------------- |\n| RATE_LIMITS      | IP rate limit counters                    | 60s  | Ephemeral (no SSOT)          |\n| SESSIONS         | Active session tokens                     | 24h  | `spine_vault` auth.sessions  |\n| CONNECTOR_STATUS | Connector install status, tenant mappings | 5min | `spine_vault` connectors     |\n| SIGNAL_CACHE     | Recent signals for fast read              | 60s  | `spine_vault` signals        |\n| METRICS          | Service health metrics                    | 30s  | Ephemeral                    |\n| WEBHOOK_LOG      | Webhook dedup keys                        | 24h  | Ephemeral (idempotency only) |\n\n**Rule:** KV is a read-through cache. If KV misses, read from D1 `spine_vault`. Never write to KV as the primary store. KV data can be lost or evicted — that is fine because D1 has the truth.\n\n### Cloudflare D1 — Edge operational DB (`spine_vault`)\n\n- **Reads**: Twin (reads structured Spine context), human (browses metadata at `/memory`), gateway (resolves tenant ID).\n- **Writes**: Pipeline (Spine writes), Tenants service (user/tenant credentials), Triage Bot (memory metadata).\n\n| Usage                       | Purpose                                       | Source of truth                                 |\n| --------------------------- | --------------------------------------------- | ----------------------------------------------- |\n| Spine Storage               | Encrypted tenant entities, signals, and goals | D1 is authoritative for active operational data |\n| Fingerprint dedup           | Content-hash keys for idempotency in pipeline | D1 is authoritative for dedup keys only         |\n| Merge candidates            | Identity resolution candidate pairs           | D1 matches, Supabase mirrors candidate pairs    |\n| AI memories (Flow C buffer) | Temporary storage before triage approval      | D1 is buffer, Supabase org_memory is truth      |\n\n#### Two Operational Modes:\n\n1. **Initial Users (CF Edge Stack only)**: D1 is the sole database. In case of D1 corruption, the system recovers by replaying JSONL backups stored in R2: `spine-backup/{tenant_id}/{date}.jsonl`.\n2. **Promoted Users (Edge + Supabase)**: The Pipeline dual-writes to D1 and Supabase. Supabase acts as the Fortress database of record.\n\n---\n\n### Cloudflare R2 — File storage (Cold Store)\n\n- **Reads**: Knowledge service (extracts content for embedding), human downloads. Twin **never** reads from R2 directly.\n- **Writes**: Triage Bot (approved memory markdown content), Knowledge service (document uploads).\n\n| Usage            | Purpose                                                        |\n| ---------------- | -------------------------------------------------------------- |\n| Document uploads | PDFs, images, exports stored in R2                             |\n| Metadata         | Stored in D1 (context_extractions table) with R2 key reference |\n\n---\n\n### Cloudflare Vectorize — Semantic Index (AI Search)\n\n- **Reads**: Twin (semantic queries for memory retrieval). **This is the only memory retrieval path for the Twin.**\n- **Writes**: Triage Bot (indexes approved R2 entries), Knowledge service (indexes uploaded document embeddings).\n\n#### R2 $\\rightarrow$ Vectorize Reindexing Cron:\n\nTo mitigate non-atomic write failures across R2 and Vectorize, a daily scheduled cron job:\n\n1. Scans R2 `memory/{tenant_id}/` keys.\n2. Compares keys against Vectorize metadata `memory_id` entries.\n3. Chunks, embeds, and indexes any missing R2 records into Vectorize.\n\n---\n\n### Cloudflare Durable Objects — Real-time coordination\n\n| DO             | Purpose                                               |\n| -------------- | ----------------------------------------------------- |\n| SignalStreamDO | WebSocket/SSE broadcasting of live signals per tenant |\n| PresenceDO     | User online/away/busy status                          |\n| RoomDO         | Collaborative editing rooms                           |\n| HITLGateDO     | Approval workflow state machine                       |\n\n**Rule:** DOs coordinate real-time state. They do NOT store canonical data. Signal content is in Spine DB. Presence is ephemeral. HITL decisions are written to Spine DB governance_audit_log.\n\n---\n\n## Rule 4: PIPELINE PROCESSING — Full integration best practices\n\nWhen data flows through the pipeline (Loader → Normalizer → Spine), these practices apply:\n\n### Idempotency\n\n- Every record gets a content-hash fingerprint at S1 (Analyze)\n- Hash = SHA-256 of (tenant_id + entity_type + source + normalized_content)\n- Before processing, check D1 for existing fingerprint\n- If exists and version is not newer → skip (dedup)\n- If exists and version is newer → process (update)\n- If not exists → process (insert)\n\n### Retry with backoff\n\n- Each pipeline stage retries 3 times on failure\n- Backoff: 30s → 60s → 90s (exponential)\n- After 3 retries → message sent to DLQ_QUEUE\n- DLQ message includes: stage name, error message, full payload, trace_id, timestamp, attempt count\n\n### Dead Letter Queue (DLQ)\n\n- DLQ_QUEUE receives all messages that fail after max retries\n- DLQ messages are also written to `normalization_errors` table in Spine DB for dashboard visibility\n- DLQ messages can be replayed manually via admin API\n- DLQ retention: indefinite (until manually resolved)\n\n### Version tracking\n\n- Every entity gets a monotonic version number\n- On conflict (two sources update same entity): highest version wins\n- Version is incremented on every successful write to Spine\n- Old versions are preserved in audit trail\n\n### Trace correlation\n\n- Every pipeline message carries a `correlation_id` (UUID)\n- Same correlation_id flows through all stages: Loader → Normalizer S1-S8 → Spine write → Knowledge enqueue → Signal enqueue\n- All logs, errors, and audit entries reference the correlation_id\n- Enables end-to-end tracing of any record from ingestion to workspace view\n\n### Schema-driven extraction\n\n- S5 (Extract) only extracts fields defined in the tenant's schema (tenant_spine_config)\n- Unknown fields are dropped — no data leakage\n- Schema is resolved at S1 and flows through all stages\n- If schema changes mid-pipeline, the message uses the schema version it started with\n\n---\n\n## Summary: Decision Matrix\n\n| Operation                                      | Layer                           | Why                                            |\n| ---------------------------------------------- | ------------------------------- | ---------------------------------------------- |\n| Read entities for workspace views              | Spine DB direct                 | RLS handles security, zero Worker overhead     |\n| Read signals, goals, memories                  | Spine DB direct                 | Same reason                                    |\n| Realtime updates (new signals, entity changes) | Spine DB Realtime               | Built-in, no custom WebSocket needed           |\n| Dashboard projections                          | Spine DB RPC (stored functions) | Complex aggregations run in DB, not in Workers |\n| Create/update entities                         | Gateway API                     | Needs validation, pipeline, audit              |\n| Approve/reject actions                         | Gateway API                     | Needs Govern check, multi-table write          |\n| Connect/disconnect tools                       | Gateway API                     | Needs OAuth, connector service                 |\n| Pipeline processing                            | Cloudflare Queues               | Async, retryable, with DLQ                     |\n| Dedup/fingerprinting                           | Cloudflare D1                   | Edge-local, fast lookup                        |\n| Rate limiting, session cache                   | Cloudflare KV                   | Ephemeral, edge-local                          |\n| File storage                                   | Cloudflare R2                   | Blob storage, metadata in Spine DB             |\n| Live signal broadcasting                       | Durable Objects                 | Real-time coordination                         |\n| Auth (login, signup, OAuth)                    | Spine DB Auth direct            | Already implemented                            |\n\n---\n\n## What Changes in the Frontend\n\nThe current `useWorkspaceEntities` hook calls Gateway API for reads. This should be changed to query Spine DB directly:\n\n**Current (wrong — 3 hops):**\n\n```\nuseWorkspaceEntities(\"task_item\")\n  → apiFetch(\"/api/v1/workspace/entities?entity_type=task_item\")\n    → Gateway Worker\n      → Spine-v2 Worker\n        → Spine DB\n```\n\n**Correct (direct — 1 hop):**\n\n```\nuseWorkspaceEntities(\"task_item\")\n  → spineDb.from(\"entities\").select(\"*\").eq(\"entity_type\", \"task_item\")\n    → Spine DB (RLS scoped)\n```\n\nThis eliminates 18 API calls per workspace load and replaces them with direct DB queries that are faster, cheaper, and more reliable.\n\n**Writes stay through Gateway:**\n\n```\ncreateEntity(data)\n  → apiFetch(\"POST /api/v1/workspace/entities\", data)\n    → Gateway → validation → Spine write → audit → downstream\n```\n\n---\n\n_This document is the architectural law for data access. If you are reading data for a view, go to Spine DB. If you are writing data that has side effects, go through Gateway. If you are processing pipeline messages, use Cloudflare Queues with idempotency, retry, and DLQ. No exceptions._\n"}