Multi-Layer Data Access Patterns

This document describes the multi-layer data access architecture that provides Redis caching, database persistence, and ERP system fallback.

Overview

The data access layer implements a sophisticated caching and fallback strategy to optimize performance while ensuring data consistency and availability. The system uses Redis for fast access, PostgreSQL for persistence, and ERP systems as the source of truth.

Architecture Diagram

graph TB
    subgraph "Application Layer"
        APP[Web Application]
        API[API Services]
        WORKER[Background Workers]
    end

    subgraph "Data Access Layer"
        DAL[Data Access Layer]
        CACHE[Cache Manager]
        DB[Database Manager]
        ERP[ERP Manager]
    end

    subgraph "Storage Layer"
        REDIS[(Redis Cache)]
        POSTGRES[(PostgreSQL)]
        ERPSYS[ERP Systems]
    end

    APP --> DAL
    API --> DAL
    WORKER --> DAL

    DAL --> CACHE
    DAL --> DB
    DAL --> ERP

    CACHE --> REDIS
    DB --> POSTGRES
    ERP --> ERPSYS

    style DAL fill:#fff3e0
    style CACHE fill:#e8f5e8
    style DB fill:#e1f5fe
    style ERP fill:#fce4ec

Data Flow Patterns

Read Operations

sequenceDiagram
    participant App as Application
    participant DAL as Data Access Layer
    participant Cache as Redis Cache
    participant DB as PostgreSQL
    participant ERP as ERP System

    App->>DAL: getData(key)

    DAL->>Cache: get(key)
    alt Cache Hit
        Cache-->>DAL: data
        DAL-->>App: data (fast)
    else Cache Miss
        DAL->>DB: query(key)
        alt DB Hit
            DB-->>DAL: data
            DAL->>Cache: set(key, data, ttl)
            DAL-->>App: data (medium)
        else DB Miss
            DAL->>ERP: fetch(key)
            alt ERP Success
                ERP-->>DAL: data
                DAL->>DB: store(data)
                DAL->>Cache: set(key, data, ttl)
                DAL-->>App: data (slow)
            else ERP Failure
                DAL-->>App: error
            end
        end
    end

Write Operations

sequenceDiagram
    participant App as Application
    participant DAL as Data Access Layer
    participant Cache as Redis Cache
    participant DB as PostgreSQL
    participant ERP as ERP System

    App->>DAL: updateData(key, data)

    par Write to ERP
        DAL->>ERP: update(key, data)
        ERP-->>DAL: success/error
    and Write to Database
        DAL->>DB: update(key, data)
        DB-->>DAL: success/error
    and Invalidate Cache
        DAL->>Cache: delete(key)
        Cache-->>DAL: success
    end

    DAL-->>App: result

    Note over DAL,ERP: ERP is source of truth
    Note over DAL,Cache: Cache invalidated on write

Caching Strategies

Cache-Aside Pattern

flowchart TD
    A[Request Data] --> B{Cache Hit?}
    B -->|Yes| C[Return Cached Data]
    B -->|No| D[Query Database]
    D --> E{Data Found?}
    E -->|Yes| F[Store in Cache]
    E -->|No| G[Query ERP]
    F --> H[Return Data]
    G --> I{ERP Success?}
    I -->|Yes| J[Store in DB & Cache]
    I -->|No| K[Return Error]
    J --> H

    style C fill:#c8e6c9
    style H fill:#c8e6c9
    style K fill:#ffcdd2

Write-Through Pattern

flowchart TD
    A[Write Request] --> B[Write to ERP]
    B --> C{ERP Success?}
    C -->|Yes| D[Write to Database]
    C -->|No| E[Return Error]
    D --> F{DB Success?}
    F -->|Yes| G[Update Cache]
    F -->|No| H[Log Error & Continue]
    G --> I[Return Success]
    H --> I

    style I fill:#c8e6c9
    style E fill:#ffcdd2

Cache Management

TTL Strategy

interface CacheConfig {
  // Product data - changes infrequently
  products: {
    ttl: 3600; // 1 hour
    refreshThreshold: 0.8; // Refresh at 80% TTL
  };

  // Customer data - moderate changes
  customers: {
    ttl: 1800; // 30 minutes
    refreshThreshold: 0.7;
  };

  // Inventory - changes frequently
  inventory: {
    ttl: 300; // 5 minutes
    refreshThreshold: 0.5;
  };

  // Search results - short-lived
  search: {
    ttl: 60; // 1 minute
    refreshThreshold: 0.9;
  };
}

Cache Invalidation

graph TD
    A[Data Update Event] --> B{Update Type}
    B -->|Product Update| C[Invalidate Product Cache]
    B -->|Customer Update| D[Invalidate Customer Cache]
    B -->|Inventory Update| E[Invalidate Inventory Cache]
    B -->|Bulk Sync| F[Invalidate All Related]

    C --> G[Clear Search Cache]
    D --> G
    E --> G
    F --> H[Clear All Cache]

    G --> I[Notify Cache Refresh]
    H --> I

    style A fill:#fff3e0
    style I fill:#e8f5e8

Database Patterns

Connection Pooling

interface DatabaseConfig {
  pool: {
    min: 5; // Minimum connections
    max: 20; // Maximum connections
    acquireTimeout: 30000; // 30 seconds
    idleTimeout: 600000; // 10 minutes
    reapInterval: 1000; // 1 second
  };

  readReplicas: {
    enabled: true;
    hosts: string[];
    loadBalancing: 'round-robin' | 'least-connections';
  };

  transactions: {
    isolationLevel: 'READ_COMMITTED';
    timeout: 30000;
  };
}

Query Optimization

graph LR
    A[Query Request] --> B{Query Type}
    B -->|Read| C[Read Replica]
    B -->|Write| D[Primary DB]
    B -->|Search| E[Full-Text Index]
    B -->|Analytics| F[Materialized View]

    C --> G[Connection Pool]
    D --> G
    E --> H[GIN Index]
    F --> I[Scheduled Refresh]

    style E fill:#e8f5e8
    style F fill:#fff3e0

ERP Integration Patterns

Circuit Breaker Pattern

stateDiagram-v2
    [*] --> Closed: Normal Operation

    Closed --> Open: Failure Threshold Reached
    Open --> HalfOpen: Timeout Elapsed
    HalfOpen --> Closed: Success
    HalfOpen --> Open: Failure

    note right of Closed
        Requests pass through
        Failures counted
    end note

    note right of Open
        Requests fail fast
        No ERP calls made
    end note

    note right of HalfOpen
        Limited requests allowed
        Testing ERP health
    end note

Retry Strategy

flowchart TD
    A[ERP Request] --> B{Success?}
    B -->|Yes| C[Return Data]
    B -->|No| D{Retryable Error?}
    D -->|No| E[Return Error]
    D -->|Yes| F{Retry Count < Max?}
    F -->|No| E
    F -->|Yes| G[Exponential Backoff]
    G --> H[Wait]
    H --> A

    style C fill:#c8e6c9
    style E fill:#ffcdd2
    style G fill:#fff3e0

Performance Optimization

Data Prefetching

sequenceDiagram
    participant App as Application
    participant Prefetch as Prefetch Service
    participant Cache as Redis Cache
    participant DB as Database
    participant ERP as ERP System

    App->>Prefetch: requestData(productId)

    par Immediate Response
        Prefetch->>Cache: get(productId)
        Cache-->>Prefetch: data (if available)
        Prefetch-->>App: immediate response
    and Background Prefetch
        Prefetch->>DB: getRelatedProducts(productId)
        DB-->>Prefetch: relatedIds[]
        loop For each related product
            Prefetch->>Cache: get(relatedId)
            alt Cache Miss
                Prefetch->>ERP: fetch(relatedId)
                ERP-->>Prefetch: data
                Prefetch->>Cache: set(relatedId, data)
            end
        end
    end

Batch Operations

graph TD
    A[Multiple Requests] --> B[Request Aggregator]
    B --> C{Batch Size Reached?}
    C -->|No| D[Wait for More Requests]
    C -->|Yes| E[Execute Batch Query]
    D --> C
    E --> F[Distribute Results]
    F --> G[Return to Clients]

    style B fill:#fff3e0
    style E fill:#e8f5e8

Error Handling

Fallback Strategies

graph TD
    A[Data Request] --> B[Try Cache]
    B --> C{Cache Available?}
    C -->|Yes| D[Return Cached Data]
    C -->|No| E[Try Database]
    E --> F{DB Available?}
    F -->|Yes| G[Return DB Data]
    F -->|No| H[Try ERP]
    H --> I{ERP Available?}
    I -->|Yes| J[Return ERP Data]
    I -->|No| K[Return Stale Cache]
    K --> L{Stale Data Available?}
    L -->|Yes| M[Return with Warning]
    L -->|No| N[Return Error]

    style D fill:#c8e6c9
    style G fill:#c8e6c9
    style J fill:#c8e6c9
    style M fill:#fff3e0
    style N fill:#ffcdd2

Health Monitoring

interface HealthMetrics {
  cache: {
    hitRate: number;
    missRate: number;
    evictionRate: number;
    connectionCount: number;
  };

  database: {
    connectionPoolUsage: number;
    queryLatency: number;
    errorRate: number;
    replicationLag: number;
  };

  erp: {
    responseTime: number;
    errorRate: number;
    circuitBreakerState: 'CLOSED' | 'OPEN' | 'HALF_OPEN';
    lastSuccessfulCall: Date;
  };
}

Configuration

Environment-Specific Settings

interface DataAccessConfig {
  cache: {
    redis: {
      host: string;
      port: number;
      password?: string;
      db: number;
      keyPrefix: string;
    };
    defaultTTL: number;
    maxMemoryPolicy: 'allkeys-lru' | 'volatile-lru';
  };

  database: {
    primary: DatabaseConnection;
    replicas: DatabaseConnection[];
    poolConfig: PoolConfig;
  };

  erp: {
    timeout: number;
    retryConfig: RetryConfig;
    circuitBreaker: CircuitBreakerConfig;
  };
}

Monitoring and Alerting

graph TD
    A[Metrics Collection] --> B[Cache Metrics]
    A --> C[DB Metrics]
    A --> D[ERP Metrics]

    B --> E[Cache Hit Rate Alert]
    C --> F[DB Connection Alert]
    D --> G[ERP Circuit Breaker Alert]

    E --> H[Notification System]
    F --> H
    G --> H

    H --> I[Slack/Email/PagerDuty]

    style A fill:#fff3e0
    style H fill:#e8f5e8

Best Practices

Cache Design

• Key Naming: Consistent, hierarchical key structure
• Data Serialization: Efficient JSON or MessagePack
• Memory Management: Monitor and tune memory usage
• Expiration: Appropriate TTL for data volatility

Database Optimization

• Index Strategy: Cover frequently queried columns
• Query Planning: Analyze and optimize slow queries
• Connection Management: Proper pool sizing
• Transaction Scope: Minimize transaction duration

ERP Integration

• Rate Limiting: Respect API limits
• Error Classification: Distinguish retryable vs permanent errors
• Data Validation: Validate before ERP submission
• Audit Logging: Track all ERP interactions

Worker & Pipeline Data Access

1. TypeScript Temporal Worker Activities (apps/ts-temporal-worker/src/activities/)

Event-driven background workflows (such as ERP order/quote writes, work-item backfills, and unsnooze schedules) access PostgreSQL through the shared @repo/db connection pool and ERP connectors via @repo/integrations.

2. Python Temporal Worker & Lakehouse Reader (apps/temporal-worker/ and packages/erp-data-py)

Python extraction and validation activities in apps/temporal-worker/ query transactional state in PostgreSQL and read analytical/catalog datasets from Cloudflare R2 Iceberg/Gold tables via DuckDB (packages/erp-data-py) and Typesense.