System Architecture

This section provides a comprehensive overview of how the tracking backend is structured and how its components interact internally.

Project Structure

tracking_backend/
├── app/
│   ├── main.py              # FastAPI application entry point & lifespan
│   ├── config.py            # Environment configuration (Settings class)
│   ├── database.py          # SQLAlchemy engine & session management
│   ├── models.py            # Database ORM models
│   ├── schemas.py           # Pydantic request/response schemas
│   ├── crud.py              # Database CRUD operations
│   ├── routers/             # API endpoint modules
│   │   ├── health.py
│   │   ├── movements.py
│   │   ├── drivers.py
│   │   ├── tractors.py
│   │   ├── trailers.py
│   │   ├── order_stops.py
│   │   └── chat.py
│   ├── services/            # Business logic & external integrations
│   │   ├── terminal_api.py      # ELD provider integration (Motive, Samsara, Geotab)
│   │   ├── mcleod_api.py        # McLeod TMS integration
│   │   ├── cache_service.py     # In-memory caching layer
│   │   ├── eta_service.py       # Mapbox ETA calculations
│   │   └── gemini_service.py    # AI chat functionality
│   └── sse/                 # Server-Sent Events streaming
│       └── location_stream.py
├── docker-compose.yml       # Container orchestration
├── Dockerfile               # Container image definition
└── requirements.txt         # Python dependencies

Core Design Patterns

DB-First Reads with Background Sync

The system follows a DB-first read pattern to ensure immediate API responses:

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   API Request   │────▶│   Database      │────▶│   Response      │
│                 │     │   (Fast Read)   │     │   (Immediate)   │
└─────────────────┘     └─────────────────┘     └─────────────────┘
                               ▲
                               │ Background Sync
                               │
┌─────────────────┐     ┌─────────────────┐
│ External APIs   │────▶│   Cache Layer   │
│ (Terminal/McLeod)│     │   (In-Memory)   │
└─────────────────┘     └─────────────────┘

API requests read directly from PostgreSQL for immediate responses
Background pollers continuously sync data from external APIs
Cache layer reduces redundant external API calls
Staleness tracking via updated_at timestamps

Dynamic Polling Strategy

Polling intervals adapt based on SSE client activity:

State	Interval	Condition
Active	5-10 seconds	SSE clients connected
Idle	60 seconds	No active SSE clients

This is implemented in location_stream.py via a client counter that triggers faster polling when client_count > 0.

Token Bucket Rate Limiting

External API calls use a proactive token bucket algorithm:

# Simplified concept from terminal_api.py
class RateLimiter:
    def __init__(self, calls_per_minute):
        self.tokens = calls_per_minute
        self.last_refill = time.time()
    
    async def acquire(self):
        # Refill tokens based on elapsed time
        # Wait if no tokens available
        # Consume one token per call

This prevents hitting external rate limits (e.g., Samsara’s 1000 calls/min).

Data Flow Diagrams

Location Data Flow

┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│   Samsara    │    │    Motive    │    │    Geotab    │
│   (ELD API)  │    │   (ELD API)  │    │   (ELD API)  │
└──────┬───────┘    └──────┬───────┘    └──────┬───────┘
       │                   │                   │
       └───────────────────┼───────────────────┘
                           ▼
                 ┌──────────────────┐
                 │   Terminal API   │
                 │  (terminal_api.py)│
                 └────────┬─────────┘
                          ▼
                 ┌──────────────────┐
                 │   Cache Layer    │
                 │(cache_service.py)│
                 └────────┬─────────┘
                          ▼
        ┌─────────────────┼─────────────────┐
        ▼                 ▼                 ▼
┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│   tractors   │  │   trailers   │  │   drivers    │
│    table     │  │    table     │  │    table     │
└──────────────┘  └──────────────┘  └──────────────┘

Movement Data Flow

┌──────────────────┐
│   McLeod TMS     │
│   (SOAP/REST)    │
└────────┬─────────┘
         ▼
┌──────────────────┐
│  mcleod_api.py   │
│  (sync_movements)│
└────────┬─────────┘
         ▼
┌──────────────────┐
│   movements      │
│     table        │
└────────┬─────────┘
         │
    ┌────┴────┐
    ▼         ▼
┌───────┐  ┌───────┐
│orders │  │ stops │
│ table │  │ table │
└───────┘  └───────┘

Database Models

Key Tables and Relationships

Table	Purpose	Key Fields
`movements`	Load/shipment records	`id`, `order_id`, `status`, `driver_id`
`drivers`	Driver profiles & HOS	`id`, `name`, `hos_status`, `location`
`tractors`	Vehicle data & locations	`id`, `unit_number`, `latitude`, `longitude`
`trailers`	Trailer tracking	`id`, `unit_number`, `latitude`, `longitude`
`orders`	Order headers	`id`, `customer`, `movement_id`
`stops`	Pickup/delivery stops	`id`, `order_id`, `type`, `eta`

Staleness Tracking

Every table includes:

updated_at - Last database update timestamp
synced_at - Last external API sync timestamp (where applicable)

This enables clients to determine data freshness.

Service Layer Details

Terminal API Integration (`terminal_api.py`)

Handles communication with ELD providers:

Motive: OAuth2 authentication, REST API
Samsara: API key authentication, REST API
Geotab: Session-based authentication, MyGeotab SDK

Each provider has:

Connection pooling
Automatic retry with exponential backoff
Error normalization to common schema

Cache Service (`cache_service.py`)

In-memory caching with TTL:

# Cache structure
{
    "locations:tractor:{id}": {
        "data": {...},
        "expires_at": timestamp
    },
    "hos:driver:{id}": {
        "data": {...},
        "expires_at": timestamp
    }
}

Default TTLs:

Location data: 30 seconds
HOS data: 60 seconds
Driver profiles: 300 seconds

Gemini Service (`gemini_service.py`)

AI chat uses function calling with these tools:

Tool	Purpose
`get_driver_location`	Fetch driver’s current position
`get_movement_status`	Check load status
`get_eta`	Calculate arrival time
`search_drivers`	Find drivers by name/ID

Lifespan Management

FastAPI’s lifespan context manages background services:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await start_background_pollers()
    await initialize_cache()
    
    yield  # Application runs
    
    # Shutdown
    await stop_background_pollers()
    await cleanup_connections()

Environment Configuration

Key settings from config.py:

Variable	Purpose
`DATABASE_URL`	PostgreSQL connection string
`TERMINAL_API_KEY`	ELD provider API key
`MCLEOD_API_URL`	TMS integration endpoint
`MAPBOX_TOKEN`	ETA calculation service
`GEMINI_API_KEY`	AI chat functionality
`POLLING_INTERVAL`	Background sync frequency

Overview

Developer Guide

API Reference

System Architecture

System Architecture

Project Structure

Core Design Patterns

DB-First Reads with Background Sync

Dynamic Polling Strategy

Token Bucket Rate Limiting

Data Flow Diagrams

Location Data Flow

Movement Data Flow

Database Models

Key Tables and Relationships

Staleness Tracking

Service Layer Details

Terminal API Integration (`terminal_api.py`)

Cache Service (`cache_service.py`)

Gemini Service (`gemini_service.py`)

Lifespan Management

Environment Configuration

Next Steps

API Reference

To-Do

Overview

Developer Guide

API Reference

​System Architecture

​Project Structure

​Core Design Patterns

​DB-First Reads with Background Sync

​Dynamic Polling Strategy

​Token Bucket Rate Limiting

​Data Flow Diagrams

​Location Data Flow

​Movement Data Flow

​Database Models

​Key Tables and Relationships

​Staleness Tracking

​Service Layer Details

​Terminal API Integration (terminal_api.py)

​Cache Service (cache_service.py)

​Gemini Service (gemini_service.py)

​Lifespan Management

​Environment Configuration

​Next Steps

API Reference

To-Do

System Architecture

Project Structure

Core Design Patterns

DB-First Reads with Background Sync

Dynamic Polling Strategy

Token Bucket Rate Limiting

Data Flow Diagrams

Location Data Flow

Movement Data Flow

Database Models

Key Tables and Relationships

Staleness Tracking

Service Layer Details

Terminal API Integration (`terminal_api.py`)

Cache Service (`cache_service.py`)

Gemini Service (`gemini_service.py`)

Lifespan Management

Environment Configuration

Next Steps