ORSANCO Flows - Technical Specifications Document

1. Executive Summary

This document provides comprehensive technical specifications for upgrading the ORSANCO River Flows system. The upgrade focuses on:

Database Optimization: Properly implementing TimescaleDB hypertables for efficient time-series queries
API Enhancements: New endpoints for bulk data downloads with aggregation support
Authentication: Token-based API authentication
Query Management: Asynchronous query processing with Azure storage integration
Scalability: Foundation for future migration to Azure App Service architecture

Note for Developers: The database currently has TimescaleDB extension enabled but the tables are NOT hypertables. This must be addressed as part of the upgrade.

2. System Architecture

2.1 Current Architecture

Components:

Frontend: React app on Azure Static Web App
Backend: Ubuntu VM hosting Django + PostgreSQL
Database: PostgreSQL with TimescaleDB extension (but NOT using hypertables yet)
Web Server: NGINX reverse proxy
Data Ingestion: Cron jobs download HDF5 files via SFTP

Current Limitations:

RiverFlow and DailyRiverFlow tables use TimescaleModel but are NOT hypertables
ORM queries are slow for large date ranges
No query queue management
Single VM limits scalability

2.2 Enhanced Architecture (This Upgrade)

Enhancements:

✅ Proper TimescaleDB hypertables for RiverFlow table
✅ Raw SQL queries instead of ORM for timeseries data
✅ Azure Blob Storage for completed query CSVs
✅ Azure Table Storage for query status tracking
✅ Token-based authentication for API
✅ Background query worker for large requests

2.3 Future Architecture (Not in Current Scope)

Future Improvements (for reference, not implemented now):

Azure App Service for better scalability
Azure Queue Storage for robust query queueing
Azure Functions for serverless query processing
Managed PostgreSQL service

3. Database Schema & TimescaleDB Implementation

3.1 Current Schema Issues

Problem: Tables created with Django ORM inherit from TimescaleModel but migrations didn't create actual hypertables.

# riverflows/models.py (CURRENT - INCORRECT)
class RiverFlow(TimescaleModel):
  id = models.BigAutoField(primary_key=True)
  time = models.BigIntegerField(...)
  station = models.ForeignKey(Station, ...)
  flow = models.FloatField(...)
  velocity = models.FloatField(...)
  stage = models.FloatField(...)

Result: The table exists in PostgreSQL but is NOT a hypertable, so queries are slow.

3.2 Proper TimescaleDB Implementation

Following the template pattern from template-api-drf-timescaledb, we need:

Migration 0002: Convert to Hypertables

# riverflows/migrations/0002_convert_to_hypertables.py

from django.db import migrations

class Migration(migrations.Migration):
  dependencies = [
    ('riverflows', '0001_initial'),
  ]

  operations = [
    # Convert RiverFlow table to hypertable
    migrations.RunSQL(
      sql="""
      -- Create hypertable on time column
      SELECT create_hypertable(
        'riverflows_riverflow',
        'time',
        chunk_time_interval => 86400000000000,  -- 1 day in microseconds
        if_not_exists => TRUE
      );

      -- Add index on station + time for efficient queries
      CREATE INDEX IF NOT EXISTS idx_riverflow_station_time
        ON riverflows_riverflow (station_id, time DESC);

      -- Add index on time only (for time-range queries)
      CREATE INDEX IF NOT EXISTS idx_riverflow_time
        ON riverflows_riverflow (time DESC);

      -- Consider compression policy (optional)
      -- ALTER TABLE riverflows_riverflow SET (
      --   timescaledb.compress,
      --   timescaledb.compress_segmentby = 'station_id'
      -- );
      """,
      reverse_sql="""
      DROP INDEX IF EXISTS idx_riverflow_station_time;
      DROP INDEX IF EXISTS idx_riverflow_time;
      -- Note: Cannot easily reverse hypertable conversion
      """
    ),

    # Convert DailyRiverFlow table to hypertable
    migrations.RunSQL(
      sql="""
      SELECT create_hypertable(
        'riverflows_dailyriverflow',
        'time',
        chunk_time_interval => 2592000000000000,  -- 30 days in microseconds
        if_not_exists => TRUE
      );

      CREATE INDEX IF NOT EXISTS idx_dailyflow_station_time
        ON riverflows_dailyriverflow (station_id, time DESC);
      """,
      reverse_sql="""
      DROP INDEX IF EXISTS idx_dailyflow_station_time;
      """
    ),

    # Add comments for documentation
    migrations.RunSQL(
      sql="""
      COMMENT ON TABLE riverflows_riverflow IS
        'TimescaleDB hypertable storing raw 15-minute river flow measurements';

      COMMENT ON TABLE riverflows_dailyriverflow IS
        'TimescaleDB hypertable storing pre-aggregated daily flow statistics';
      """,
      reverse_sql=""
    ),
  ]

Key Points:

chunk_time_interval: Controls how data is partitioned (1 day for raw, 30 days for daily)
Indexes on (station_id, time DESC) optimize station-specific time-range queries
Compression can be enabled later for older data

3.3 Updated Models

Models remain largely the same but we add helper methods:

# riverflows/models.py (UPDATED)

from django.db import models
from timescale.db.models.models import TimescaleModel

class RiverFlow(TimescaleModel):
  id = models.BigAutoField(primary_key=True)
  time = models.BigIntegerField(
    default=0,
    null=False,
    blank=False,
    db_index=True,
    help_text="Unix epoch timestamp in seconds"
  )
  station = models.ForeignKey(
    Station,
    blank=True,
    null=True,
    on_delete=models.CASCADE,
    db_index=True
  )
  flow = models.FloatField(default=0, null=True, blank=False)
  velocity = models.FloatField(default=0, null=True, blank=False)
  stage = models.FloatField(default=0, null=True, blank=False)

  class Meta:
    unique_together = (('time', 'station'), )
    db_table = 'riverflows_riverflow'
    indexes = [
      models.Index(fields=['station', '-time'], name='idx_station_time'),
      models.Index(fields=['-time'], name='idx_time_desc'),
    ]

  def __str__(self):
    return f"Flow @ {self.station} - {self.time}"

3.4 Raw SQL Query Module

CRITICAL: Do NOT use Django ORM for timeseries queries. Use raw SQL.

Create riverflows/sql_queries.py:

"""
Centralized SQL queries for ORSANCO river flow data operations.
Uses TimescaleDB features for efficient time-series querying.
"""

from django.db import connection
from typing import List, Dict, Any, Optional
from datetime import datetime

class FlowQueries:
  """Centralized class for all flow-related SQL queries."""

  @staticmethod
  def get_raw_flow_data(
    station_id: int,
    start_time: int,
    end_time: int,
    parameters: List[str] = ['flow', 'velocity', 'stage']
  ) -> List[Dict[str, Any]]:
    """
    Get raw 15-minute data for a station within a time range.

    Args:
      station_id: Station ID
      start_time: Start timestamp (Unix epoch seconds)
      end_time: End timestamp (Unix epoch seconds)
      parameters: List of parameters to retrieve

    Returns:
      List of dicts with time and requested parameters
    """
    # Build SELECT clause dynamically based on requested parameters
    param_columns = ', '.join(parameters)

    query = f"""
      SELECT time, {param_columns}
      FROM riverflows_riverflow
      WHERE station_id = %s
        AND time >= %s
        AND time <= %s
      ORDER BY time ASC
    """

    with connection.cursor() as cursor:
      cursor.execute(query, [station_id, start_time, end_time])
      columns = [col[0] for col in cursor.description]
      return [dict(zip(columns, row)) for row in cursor.fetchall()]

  @staticmethod
  def get_aggregated_flow_data(
    station_id: int,
    start_time: int,
    end_time: int,
    interval: str,  # 'hourly' or 'daily'
    parameters: List[str] = ['flow', 'velocity', 'stage'],
    aggregators: List[str] = ['avg', 'min', 'max']
  ) -> List[Dict[str, Any]]:
    """
    Get aggregated flow data using TimescaleDB time_bucket function.

    Args:
      station_id: Station ID
      start_time: Start timestamp (Unix epoch seconds)
      end_time: End timestamp (Unix epoch seconds)
      interval: 'hourly' or 'daily'
      parameters: List of parameters to aggregate
      aggregators: List of aggregation functions

    Returns:
      List of dicts with bucketed time and aggregated values
    """
    # Determine bucket size
    if interval == 'hourly':
      bucket_size = 3600  # 1 hour in seconds
    elif interval == 'daily':
      bucket_size = 86400  # 1 day in seconds
    else:
      raise ValueError(f"Invalid interval: {interval}")

    # Build aggregation columns dynamically
    agg_cols = []
    for param in parameters:
      for agg in aggregators:
        agg_cols.append(f"{agg.upper()}({param}) as {param}_{agg}")

    agg_clause = ',\n    '.join(agg_cols)

    query = f"""
      SELECT
        time_bucket(%s, time) AS bucket_time,
        {agg_clause}
      FROM riverflows_riverflow
      WHERE station_id = %s
        AND time >= %s
        AND time <= %s
      GROUP BY bucket_time
      ORDER BY bucket_time ASC
    """

    with connection.cursor() as cursor:
      cursor.execute(query, [bucket_size, station_id, start_time, end_time])
      columns = [col[0] for col in cursor.description]
      return [dict(zip(columns, row)) for row in cursor.fetchall()]

  @staticmethod
  def get_daily_flow_data_precomputed(
    station_id: int,
    start_time: int,
    end_time: int
  ) -> List[Dict[str, Any]]:
    """
    Get pre-computed daily aggregates from DailyRiverFlow table.
    This is faster for daily queries since data is pre-aggregated.

    Args:
      station_id: Station ID
      start_time: Start timestamp (Unix epoch seconds)
      end_time: End timestamp (Unix epoch seconds)

    Returns:
      List of dicts with daily aggregated data
    """
    query = """
      SELECT
        time,
        avg_flow, max_flow, min_flow, stddev_flow,
        avg_velocity, max_velocity, min_velocity, stddev_velocity,
        avg_stage, max_stage, min_stage, stddev_stage
      FROM riverflows_dailyriverflow
      WHERE station_id = %s
        AND time >= %s
        AND time <= %s
      ORDER BY time ASC
    """

    with connection.cursor() as cursor:
      cursor.execute(query, [station_id, start_time, end_time])
      columns = [col[0] for col in cursor.description]
      return [dict(zip(columns, row)) for row in cursor.fetchall()]

  @staticmethod
  def get_multiple_stations_aggregated(
    station_ids: List[int],
    start_time: int,
    end_time: int,
    interval: str,
    parameters: List[str],
    aggregators: List[str]
  ) -> Dict[int, List[Dict[str, Any]]]:
    """
    Get aggregated data for multiple stations.
    Returns a dict keyed by station_id.

    Args:
      station_ids: List of station IDs
      start_time: Start timestamp
      end_time: End timestamp
      interval: 'hourly' or 'daily'
      parameters: List of parameters to aggregate
      aggregators: List of aggregation functions

    Returns:
      Dict mapping station_id to list of aggregated data points
    """
    results = {}
    for station_id in station_ids:
      results[station_id] = FlowQueries.get_aggregated_flow_data(
        station_id, start_time, end_time, interval, parameters, aggregators
      )
    return results

Benefits:

Uses TimescaleDB time_bucket() function for efficient aggregation
Avoids Django ORM overhead
Can easily add more TimescaleDB features (continuous aggregates, compression)
Type hints and documentation

4. API Endpoints

4.1 Authentication

4.1.1 Token Authentication Endpoint

Endpoint: POST /api/auth/token/

Purpose: Authenticate frontend and receive JWT token.

Request Body:

{
  "username": "orsanco_flows",
  "password": "secretpassword"
}

Response (Success):

{
  "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "expires_at": "2024-11-12T10:30:00Z"
}

Response (Failure):

{
  "error": "Invalid credentials"
}

Implementation:

# riverflows/views.py

from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework import status
from django.contrib.auth import authenticate
from rest_framework.authtoken.models import Token
from datetime import datetime, timedelta

@api_view(['POST'])
@permission_classes([AllowAny])
def obtain_auth_token(request):
  """
  Obtain authentication token for API access.
  Frontend stores this token and sends it in Authorization header.
  """
  username = request.data.get('username')
  password = request.data.get('password')

  user = authenticate(username=username, password=password)

  if user:
    token, created = Token.objects.get_or_create(user=user)

    # Token expiration (set to 24 hours)
    expires_at = datetime.now() + timedelta(hours=24)

    return Response({
      'token': token.key,
      'expires_at': expires_at.isoformat()
    })
  else:
    return Response(
      {'error': 'Invalid credentials'},
      status=status.HTTP_401_UNAUTHORIZED
    )

@api_view(['POST'])
def refresh_auth_token(request):
  """
  Refresh an existing token.
  Frontend calls this before token expires.
  """
  # Delete old token and create new one
  request.user.auth_token.delete()
  token = Token.objects.create(user=request.user)

  expires_at = datetime.now() + timedelta(hours=24)

  return Response({
    'token': token.key,
    'expires_at': expires_at.isoformat()
  })

Frontend Storage:

// Store token in localStorage (or secure cookie)
const response = await fetch('/api/auth/token/', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({username: 'orsanco_flows', password: 'xxx'})
});

const {token, expires_at} = await response.json();
localStorage.setItem('api_token', token);
localStorage.setItem('token_expires', expires_at);

// Use token in subsequent requests
fetch('/api/stations/', {
  headers: {
    'Authorization': `Token ${token}`
  }
});

4.2 Existing Endpoints (Enhanced)

4.2.1 Get Rivers

Endpoint: GET /api/rivers/

Authentication: Required (Token)

Response:

[
  {"id": 1, "name": "Ohio River"},
  {"id": 2, "name": "Licking River"},
  ...
]

No changes needed - existing endpoint works.

4.2.2 Get Reaches

Endpoint: GET /api/reaches/?river={river_id}

Authentication: Required

Query Parameters:

river (optional): Filter by river ID

Response:

[
  {"id": 1, "name": "606", "river": 1},
  {"id": 2, "name": "607", "river": 1},
  ...
]

Enhancement: Add filtering by river ID (update view).

4.2.3 Get Stations

Endpoint: GET /api/stations/?reach={reach_id}&search={query}

Authentication: Required

Query Parameters:

reach (optional): Filter by reach ID
search (optional): Search by station name or river mile

Response:

[
  {
    "id": 1,
    "hdf_index": 45,
    "name": "Station 699.5",
    "river": 1,
    "reach": 1,
    "x_coord": -84.5,
    "y_coord": 39.1
  },
  ...
]

Enhancement: Add search functionality.

class GetStation(generics.ListCreateAPIView):
  serializer_class = StationSerializer
  permission_classes = [IsAuthenticated]

  def get_queryset(self):
    queryset = Station.objects.all()

    # Filter by reach
    reach_id = self.request.query_params.get('reach')
    if reach_id:
      queryset = queryset.filter(reach_id=reach_id)

    # Search by name or river mile (partial match)
    search = self.request.query_params.get('search')
    if search:
      queryset = queryset.filter(
        models.Q(name__icontains=search) |
        models.Q(hdf_index__icontains=search)
      )

    return queryset

4.2.4 Get Daily Flows (Existing)

Endpoint: GET /api/daily-flows/?river={name}&reach={name}&station={name}&startdate={timestamp}&enddate={timestamp}

No changes needed - uses pre-computed daily table, works well for short ranges.

4.3 New Bulk Download Endpoints

4.3.1 Submit Bulk Download Query

Endpoint: POST /api/bulk-download/

Authentication: Required

Purpose: Submit a query for bulk data download. Query is processed asynchronously.

Request Body:

{
  "station_ids": [1, 5, 23],
  "start_time": 1230768000,
  "end_time": 1699660800,
  "parameters": ["flow", "velocity", "stage"],
  "aggregation": {
    "type": "daily",
    "functions": ["min", "max", "avg"]
  }
}

OR for raw data:

{
  "station_ids": [1],
  "start_time": 1230768000,
  "end_time": 1699660800,
  "parameters": ["flow"],
  "aggregation": {
    "type": "none"
  }
}

Response:

{
  "query_id": "abc123-def456-ghi789",
  "status": "queued",
  "estimated_completion": "2024-11-11T15:30:00Z",
  "message": "Query submitted successfully"
}

Implementation:

# riverflows/views.py

from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework import status as http_status
import uuid
from datetime import datetime, timedelta
from .models import BulkDownloadQuery

@api_view(['POST'])
@permission_classes([IsAuthenticated])
def submit_bulk_download(request):
  """
  Submit a bulk download query.
  Query is added to Azure Table Storage and processed asynchronously.
  """
  data = request.data

  # Validate required fields
  required_fields = ['station_ids', 'start_time', 'end_time', 'parameters', 'aggregation']
  for field in required_fields:
    if field not in data:
      return Response(
        {'error': f'Missing required field: {field}'},
        status=http_status.HTTP_400_BAD_REQUEST
      )

  # Validate station_ids
  station_ids = data['station_ids']
  if not isinstance(station_ids, list) or len(station_ids) == 0:
    return Response(
      {'error': 'station_ids must be a non-empty list'},
      status=http_status.HTTP_400_BAD_REQUEST
    )

  # Validate time range
  start_time = int(data['start_time'])
  end_time = int(data['end_time'])
  if start_time >= end_time:
    return Response(
      {'error': 'start_time must be before end_time'},
      status=http_status.HTTP_400_BAD_REQUEST
    )

  # Validate parameters
  valid_params = ['flow', 'velocity', 'stage']
  parameters = data['parameters']
  if not all(p in valid_params for p in parameters):
    return Response(
      {'error': f'Invalid parameter. Must be one of: {valid_params}'},
      status=http_status.HTTP_400_BAD_REQUEST
    )

  # Validate aggregation
  aggregation = data['aggregation']
  agg_type = aggregation.get('type')
  if agg_type not in ['none', 'hourly', 'daily']:
    return Response(
      {'error': 'aggregation.type must be one of: none, hourly, daily'},
      status=http_status.HTTP_400_BAD_REQUEST
    )

  if agg_type != 'none':
    agg_funcs = aggregation.get('functions', [])
    valid_funcs = ['min', 'max', 'avg']
    if not all(f in valid_funcs for f in agg_funcs):
      return Response(
        {'error': f'Invalid aggregation function. Must be one of: {valid_funcs}'},
        status=http_status.HTTP_400_BAD_REQUEST
      )

  # Generate unique query ID
  query_id = str(uuid.uuid4())

  # Estimate completion time (rough estimate based on data size)
  num_stations = len(station_ids)
  time_range_days = (end_time - start_time) / 86400
  estimated_minutes = (num_stations * time_range_days) / 365  # Rough heuristic
  estimated_completion = datetime.now() + timedelta(minutes=max(1, estimated_minutes))

  # Create query record in Azure Table Storage
  from .azure_storage import create_query_record
  create_query_record(
    query_id=query_id,
    station_ids=station_ids,
    start_time=start_time,
    end_time=end_time,
    parameters=parameters,
    aggregation=aggregation,
    status='queued',
    submitted_at=datetime.now().isoformat(),
    estimated_completion=estimated_completion.isoformat()
  )

  # TODO: If implementing Azure Queue Storage in future, add to queue here
  # For now, a background worker thread will poll Azure Table Storage

  return Response({
    'query_id': query_id,
    'status': 'queued',
    'estimated_completion': estimated_completion.isoformat(),
    'message': 'Query submitted successfully'
  })

4.3.2 Get Query Status

Endpoint: GET /api/bulk-download/{query_id}/status/

Authentication: Required

Response (Queued):

{
  "query_id": "abc123-def456-ghi789",
  "status": "queued",
  "position_in_queue": 3,
  "estimated_completion": "2024-11-11T15:30:00Z"
}

Response (Processing):

{
  "query_id": "abc123-def456-ghi789",
  "status": "processing",
  "progress_percent": 45,
  "estimated_completion": "2024-11-11T15:25:00Z"
}

Response (Complete):

{
  "query_id": "abc123-def456-ghi789",
  "status": "complete",
  "download_url": "https://orsancoflows.blob.core.windows.net/query-results/abc123-def456-ghi789.zip",
  "file_size_mb": 234.5,
  "expires_at": "2024-12-11T15:30:00Z",
  "completed_at": "2024-11-11T15:20:00Z"
}

Response (Failed):

{
  "query_id": "abc123-def456-ghi789",
  "status": "failed",
  "error_message": "Database connection timeout",
  "failed_at": "2024-11-11T15:15:00Z"
}

Implementation:

@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_query_status(request, query_id):
  """
  Get the status of a bulk download query from Azure Table Storage.
  """
  from .azure_storage import get_query_record

  query = get_query_record(query_id)

  if not query:
    return Response(
      {'error': 'Query not found'},
      status=http_status.HTTP_404_NOT_FOUND
    )

  response_data = {
    'query_id': query_id,
    'status': query['status'],
  }

  if query['status'] == 'queued':
    response_data['position_in_queue'] = query.get('position_in_queue', 0)
    response_data['estimated_completion'] = query.get('estimated_completion')

  elif query['status'] == 'processing':
    response_data['progress_percent'] = query.get('progress_percent', 0)
    response_data['estimated_completion'] = query.get('estimated_completion')

  elif query['status'] == 'complete':
    response_data['download_url'] = query.get('download_url')
    response_data['file_size_mb'] = query.get('file_size_mb')
    response_data['expires_at'] = query.get('expires_at')
    response_data['completed_at'] = query.get('completed_at')

  elif query['status'] == 'failed':
    response_data['error_message'] = query.get('error_message')
    response_data['failed_at'] = query.get('failed_at')

  return Response(response_data)

4.3.3 List User Queries

Endpoint: GET /api/bulk-download/queries/

Authentication: Required

Query Parameters:

status (optional): Filter by status (queued, processing, complete, failed)
limit (optional): Max results (default: 50)

Response:

{
  "queries": [
    {
      "query_id": "abc123",
      "status": "complete",
      "station_count": 3,
      "date_range": "2009-01-01 to 2024-11-11",
      "submitted_at": "2024-11-11T14:00:00Z",
      "completed_at": "2024-11-11T15:20:00Z",
      "file_size_mb": 234.5,
      "download_url": "https://..."
    },
    ...
  ],
  "total_storage_mb": 1234.5,
  "storage_limit_mb": 10240
}

4.3.4 Delete Query

Endpoint: DELETE /api/bulk-download/{query_id}/

Authentication: Required

Purpose: Delete a query record and its associated file from Blob Storage.

Response:

{
  "message": "Query deleted successfully",
  "freed_storage_mb": 234.5
}

5. Azure Storage Integration

5.1 Azure Blob Storage (CSV Results)

Purpose: Store generated CSV/ZIP files for download.

Configuration:

# riverflows/settings.py

# Azure Storage settings
AZURE_STORAGE_CONNECTION_STRING = os.environ.get('AZURE_STORAGE_CONNECTION_STRING')
AZURE_BLOB_CONTAINER_NAME = 'query-results'
AZURE_BLOB_SAS_EXPIRY_HOURS = 720  # 30 days

Implementation (riverflows/azure_storage.py):

from azure.storage.blob import BlobServiceClient, generate_blob_sas, BlobSasPermissions
from django.conf import settings
from datetime import datetime, timedelta
import os

def get_blob_service_client():
  """Get Azure Blob Service client."""
  return BlobServiceClient.from_connection_string(
    settings.AZURE_STORAGE_CONNECTION_STRING
  )

def upload_query_result(query_id: str, file_path: str) -> str:
  """
  Upload a query result file to Azure Blob Storage.

  Args:
    query_id: Unique query identifier
    file_path: Local path to file to upload

  Returns:
    Public URL to the uploaded blob
  """
  blob_service_client = get_blob_service_client()
  container_client = blob_service_client.get_container_client(
    settings.AZURE_BLOB_CONTAINER_NAME
  )

  # Create container if it doesn't exist
  try:
    container_client.create_container()
  except:
    pass  # Container already exists

  # Upload file
  blob_name = f"{query_id}.zip"
  blob_client = container_client.get_blob_client(blob_name)

  with open(file_path, 'rb') as data:
    blob_client.upload_blob(data, overwrite=True)

  # Generate SAS token for secure download
  sas_token = generate_blob_sas(
    account_name=blob_service_client.account_name,
    container_name=settings.AZURE_BLOB_CONTAINER_NAME,
    blob_name=blob_name,
    account_key=blob_service_client.credential.account_key,
    permission=BlobSasPermissions(read=True),
    expiry=datetime.utcnow() + timedelta(hours=settings.AZURE_BLOB_SAS_EXPIRY_HOURS)
  )

  # Return full URL with SAS token
  return f"{blob_client.url}?{sas_token}"

def delete_query_result(query_id: str) -> bool:
  """
  Delete a query result file from Azure Blob Storage.

  Args:
    query_id: Unique query identifier

  Returns:
    True if deleted successfully
  """
  blob_service_client = get_blob_service_client()
  container_client = blob_service_client.get_container_client(
    settings.AZURE_BLOB_CONTAINER_NAME
  )

  blob_name = f"{query_id}.zip"
  blob_client = container_client.get_blob_client(blob_name)

  try:
    blob_client.delete_blob()
    return True
  except:
    return False

def get_blob_size_mb(query_id: str) -> float:
  """Get size of a blob in MB."""
  blob_service_client = get_blob_service_client()
  container_client = blob_service_client.get_container_client(
    settings.AZURE_BLOB_CONTAINER_NAME
  )

  blob_name = f"{query_id}.zip"
  blob_client = container_client.get_blob_client(blob_name)

  properties = blob_client.get_blob_properties()
  return properties.size / (1024 * 1024)

5.2 Azure Table Storage (Query Metadata)

Purpose: Track query status, metadata, and history.

Table Schema:

Table Name: bulkdownloadqueries

Columns:
- PartitionKey: "query" (constant for all rows)
- RowKey: query_id (UUID string)
- status: "queued" | "processing" | "complete" | "failed"
- station_ids: JSON string list
- start_time: Unix timestamp
- end_time: Unix timestamp
- parameters: JSON string list
- aggregation: JSON string object
- submitted_at: ISO timestamp
- started_at: ISO timestamp (when processing began)
- completed_at: ISO timestamp (when finished)
- failed_at: ISO timestamp (if failed)
- error_message: Error string (if failed)
- download_url: Blob URL with SAS token
- file_size_mb: Float
- expires_at: ISO timestamp (30 days from completion)
- progress_percent: Integer (0-100)

Implementation:

# riverflows/azure_storage.py (continued)

from azure.data.tables import TableServiceClient, TableEntity
from typing import List, Dict, Any, Optional
import json

def get_table_service_client():
  """Get Azure Table Service client."""
  return TableServiceClient.from_connection_string(
    settings.AZURE_STORAGE_CONNECTION_STRING
  )

def create_query_record(
  query_id: str,
  station_ids: List[int],
  start_time: int,
  end_time: int,
  parameters: List[str],
  aggregation: Dict[str, Any],
  status: str,
  submitted_at: str,
  estimated_completion: str
) -> None:
  """
  Create a query record in Azure Table Storage.
  """
  table_service = get_table_service_client()
  table_client = table_service.get_table_client('bulkdownloadqueries')

  # Create table if it doesn't exist
  try:
    table_service.create_table('bulkdownloadqueries')
  except:
    pass

  entity = TableEntity()
  entity['PartitionKey'] = 'query'
  entity['RowKey'] = query_id
  entity['status'] = status
  entity['station_ids'] = json.dumps(station_ids)
  entity['start_time'] = start_time
  entity['end_time'] = end_time
  entity['parameters'] = json.dumps(parameters)
  entity['aggregation'] = json.dumps(aggregation)
  entity['submitted_at'] = submitted_at
  entity['estimated_completion'] = estimated_completion
  entity['progress_percent'] = 0

  table_client.create_entity(entity)

def get_query_record(query_id: str) -> Optional[Dict[str, Any]]:
  """
  Get a query record from Azure Table Storage.
  """
  table_service = get_table_service_client()
  table_client = table_service.get_table_client('bulkdownloadqueries')

  try:
    entity = table_client.get_entity('query', query_id)

    # Convert to dict and parse JSON fields
    result = dict(entity)
    result['station_ids'] = json.loads(result['station_ids'])
    result['parameters'] = json.loads(result['parameters'])
    result['aggregation'] = json.loads(result['aggregation'])

    return result
  except:
    return None

def update_query_status(
  query_id: str,
  status: str,
  **kwargs
) -> None:
  """
  Update a query's status and other fields.

  Args:
    query_id: Query identifier
    status: New status
    **kwargs: Additional fields to update (e.g., progress_percent, error_message)
  """
  table_service = get_table_service_client()
  table_client = table_service.get_table_client('bulkdownloadqueries')

  entity = table_client.get_entity('query', query_id)
  entity['status'] = status

  for key, value in kwargs.items():
    entity[key] = value

  table_client.update_entity(entity)

def list_queries(status: Optional[str] = None, limit: int = 50) -> List[Dict[str, Any]]:
  """
  List queries, optionally filtered by status.
  """
  table_service = get_table_service_client()
  table_client = table_service.get_table_client('bulkdownloadqueries')

  if status:
    filter_query = f"status eq '{status}'"
    entities = table_client.query_entities(filter_query, results_per_page=limit)
  else:
    entities = table_client.list_entities(results_per_page=limit)

  results = []
  for entity in entities:
    result = dict(entity)
    result['station_ids'] = json.loads(result['station_ids'])
    result['parameters'] = json.loads(result['parameters'])
    result['aggregation'] = json.loads(result['aggregation'])
    results.append(result)

  return results

def delete_query_record(query_id: str) -> bool:
  """
  Delete a query record from Azure Table Storage.
  """
  table_service = get_table_service_client()
  table_client = table_service.get_table_client('bulkdownloadqueries')

  try:
    table_client.delete_entity('query', query_id)
    return True
  except:
    return False

6. Query Processing Worker

6.1 Worker Architecture

The query worker runs as a background thread/process on the Django app server. It:

Polls Azure Table Storage for queries with status='queued'
Processes one query at a time (to avoid overloading DB)
Updates status to 'processing' and tracks progress
Generates CSV files and uploads to Blob Storage
Updates status to 'complete' with download URL

Future Enhancement: Replace with Azure Queue Storage + Azure Functions for better scalability.

6.2 Worker Implementation

Create riverflows/management/commands/process_bulk_queries.py:

from django.core.management.base import BaseCommand
from riverflows.azure_storage import list_queries, update_query_status, upload_query_result, get_blob_size_mb
from riverflows.sql_queries import FlowQueries
from riverflows.models import Station
import csv
import tempfile
import zipfile
import os
from datetime import datetime, timedelta
import time

class Command(BaseCommand):
  help = 'Background worker to process bulk download queries'

  def handle(self, *args, **options):
    self.stdout.write('Starting bulk query worker...')

    while True:
      try:
        # Get queued queries
        queued = list_queries(status='queued', limit=1)

        if queued:
          query = queued[0]
          query_id = query['RowKey']

          self.stdout.write(f'Processing query {query_id}')

          # Update status to processing
          update_query_status(
            query_id,
            'processing',
            started_at=datetime.now().isoformat(),
            progress_percent=0
          )

          try:
            # Process the query
            self._process_query(query)

          except Exception as e:
            # Mark as failed
            self.stdout.write(self.style.ERROR(f'Query {query_id} failed: {str(e)}'))
            update_query_status(
              query_id,
              'failed',
              error_message=str(e),
              failed_at=datetime.now().isoformat()
            )

        else:
          # No queued queries, sleep for a bit
          time.sleep(10)

      except Exception as e:
        self.stdout.write(self.style.ERROR(f'Worker error: {str(e)}'))
        time.sleep(10)

  def _process_query(self, query):
    """Process a single query."""
    query_id = query['RowKey']
    station_ids = query['station_ids']
    start_time = query['start_time']
    end_time = query['end_time']
    parameters = query['parameters']
    aggregation = query['aggregation']

    # Create temp directory for CSV files
    with tempfile.TemporaryDirectory() as tmpdir:
      csv_files = []

      # Process each station
      total_stations = len(station_ids)
      for idx, station_id in enumerate(station_ids):
        # Get station info
        station = Station.objects.get(id=station_id)

        # Generate data based on aggregation type
        if aggregation['type'] == 'none':
          # Raw data
          data = FlowQueries.get_raw_flow_data(
            station_id, start_time, end_time, parameters
          )
        else:
          # Aggregated data
          data = FlowQueries.get_aggregated_flow_data(
            station_id,
            start_time,
            end_time,
            aggregation['type'],
            parameters,
            aggregation.get('functions', ['avg'])
          )

        # Write to CSV
        csv_filename = self._generate_csv_filename(
          station.name, start_time, end_time
        )
        csv_path = os.path.join(tmpdir, csv_filename)

        self._write_csv(csv_path, data)
        csv_files.append(csv_path)

        # Update progress
        progress = int(((idx + 1) / total_stations) * 90)  # 0-90%
        update_query_status(query_id, 'processing', progress_percent=progress)

      # Create ZIP file
      zip_path = os.path.join(tmpdir, f'{query_id}.zip')
      with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for csv_file in csv_files:
          zipf.write(csv_file, os.path.basename(csv_file))

      # Update progress (95%)
      update_query_status(query_id, 'processing', progress_percent=95)

      # Upload to Blob Storage
      download_url = upload_query_result(query_id, zip_path)
      file_size_mb = get_blob_size_mb(query_id)

      # Mark as complete
      expires_at = datetime.now() + timedelta(days=30)
      update_query_status(
        query_id,
        'complete',
        completed_at=datetime.now().isoformat(),
        download_url=download_url,
        file_size_mb=file_size_mb,
        expires_at=expires_at.isoformat(),
        progress_percent=100
      )

      self.stdout.write(self.style.SUCCESS(f'Query {query_id} completed'))

  def _generate_csv_filename(self, station_name, start_time, end_time):
    """Generate CSV filename based on conventions."""
    start_date = datetime.fromtimestamp(start_time).strftime('%Y%m%d')
    end_date = datetime.fromtimestamp(end_time).strftime('%Y%m%d')
    # Sanitize station name (remove spaces, special chars)
    safe_name = station_name.replace(' ', '-').replace('/', '-')
    return f"{start_date}_{end_date}_{safe_name}.csv"

  def _write_csv(self, filepath, data):
    """Write data to CSV file in wide format."""
    if not data:
      return

    with open(filepath, 'w', newline='') as csvfile:
      # Get column names from first row
      fieldnames = data[0].keys()
      writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

      writer.writeheader()
      for row in data:
        writer.writerow(row)

6.3 Running the Worker

Development:

python manage.py process_bulk_queries

Production (systemd service):

# /etc/systemd/system/flows-query-worker.service

[Unit]
Description=ORSANCO Flows Bulk Query Worker
After=network.target postgresql.service

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/riverflows-api-drf
Environment="PATH=/opt/riverflows-api-drf/venv/bin"
ExecStart=/opt/riverflows-api-drf/venv/bin/python manage.py process_bulk_queries
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

sudo systemctl enable flows-query-worker
sudo systemctl start flows-query-worker
sudo systemctl status flows-query-worker

7. Performance Optimization

7.1 Database Query Optimization

TimescaleDB Best Practices:

Use Time Buckets: Always use time_bucket() for aggregations
Add Indexes: Create composite indexes on (station_id, time DESC)
Compression: Enable compression on older chunks
Continuous Aggregates: Consider for frequently-queried aggregations

Example - Enable Compression:

-- Enable compression on RiverFlow table for chunks older than 30 days
ALTER TABLE riverflows_riverflow SET (
  timescaledb.compress,
  timescaledb.compress_segmentby = 'station_id'
);

SELECT add_compression_policy('riverflows_riverflow', INTERVAL '30 days');

Example - Continuous Aggregate (future enhancement):

-- Create a continuous aggregate for daily data (if not using DailyRiverFlow)
CREATE MATERIALIZED VIEW daily_river_flow_cagg
WITH (timescaledb.continuous) AS
SELECT
  time_bucket(86400, time) AS day,
  station_id,
  AVG(flow) as avg_flow,
  MAX(flow) as max_flow,
  MIN(flow) as min_flow,
  AVG(velocity) as avg_velocity,
  MAX(velocity) as max_velocity,
  MIN(velocity) as min_velocity,
  AVG(stage) as avg_stage,
  MAX(stage) as max_stage,
  MIN(stage) as min_stage
FROM riverflows_riverflow
GROUP BY day, station_id;

-- Add refresh policy
SELECT add_continuous_aggregate_policy('daily_river_flow_cagg',
  start_offset => INTERVAL '3 days',
  end_offset => INTERVAL '1 hour',
  schedule_interval => INTERVAL '1 hour');

7.2 Query Worker Optimization

Concurrent Processing (future):

Currently processes one query at a time
Future: Use multiprocessing to handle multiple queries in parallel
Limit concurrent queries to avoid DB overload (e.g., max 3)

Chunked CSV Writing:

For very large queries, write CSV in chunks
Stream data from DB instead of loading all into memory

7.3 Frontend Optimization

Lazy Loading:

Only load Query History as needed
Paginate history (50 at a time)

Caching:

Cache river/reach/station lists in frontend
Only refresh when needed

8. Security Considerations

8.1 Authentication

Token Expiry: Tokens expire after 24 hours
Token Rotation: Frontend auto-refreshes before expiry
HTTPS Only: All API calls over HTTPS

8.2 Authorization

Single password for now (stored in frontend env)
Future: User accounts with role-based access

8.3 Input Validation

Validate all user inputs on backend
Prevent SQL injection (use parameterized queries)
Limit query size to prevent DoS

8.4 Azure Storage Security

SAS Tokens: Blobs use SAS tokens for secure download
Expiry: SAS tokens expire after 30 days
Private Container: Blob container is private (no public access)

8.5 Rate Limiting

Add rate limiting to prevent abuse:

# Install: pip install django-ratelimit

from django_ratelimit.decorators import ratelimit

@ratelimit(key='user', rate='5/m')  # 5 queries per minute
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def submit_bulk_download(request):
  # ... existing code

9. Monitoring & Logging

9.1 Logging Configuration

# riverflows/settings.py

LOGGING = {
  'version': 1,
  'disable_existing_loggers': False,
  'formatters': {
    'verbose': {
      'format': '{levelname} {asctime} {module} {message}',
      'style': '{',
    },
  },
  'handlers': {
    'file': {
      'level': 'INFO',
      'class': 'logging.handlers.RotatingFileHandler',
      'filename': '/var/log/riverflows/django.log',
      'maxBytes': 1024*1024*50,  # 50 MB
      'backupCount': 5,
      'formatter': 'verbose',
    },
    'query_worker': {
      'level': 'INFO',
      'class': 'logging.handlers.RotatingFileHandler',
      'filename': '/var/log/riverflows/query_worker.log',
      'maxBytes': 1024*1024*50,
      'backupCount': 5,
      'formatter': 'verbose',
    },
  },
  'loggers': {
    'django': {
      'handlers': ['file'],
      'level': 'INFO',
    },
    'riverflows': {
      'handlers': ['file'],
      'level': 'DEBUG',
    },
    'query_worker': {
      'handlers': ['query_worker'],
      'level': 'INFO',
    },
  },
}

9.2 Metrics to Monitor

Query processing time (track in Azure Table Storage)
Database query performance (log slow queries)
Worker uptime and restarts
Storage usage (blob + table)
API response times
Failed queries (investigate patterns)

10. Deployment Plan

10.1 Prerequisites

Azure Resources:
- Existing: VM, Static Web App
- New: Storage Account (Blob + Table)
Dependencies:

# requirements.txt (add to existing)
azure-storage-blob==12.19.0
azure-data-tables==12.4.4
djangorestframework-simplejwt==5.3.0  # or django-rest-auth
django-ratelimit==4.1.0

Environment Variables:

# Add to .env or Azure VM environment
AZURE_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;EndpointSuffix=core.windows.net"

10.2 Database Migration Steps

IMPORTANT: Run migrations during maintenance window (low traffic).

# 1. Backup database
pg_dump riverflows_db > backup_$(date +%Y%m%d).sql

# 2. Run migrations
python manage.py migrate riverflows 0002_convert_to_hypertables

# 3. Verify hypertables were created
psql -U postgres -d riverflows_db -c "SELECT * FROM timescaledb_information.hypertables;"

# Expected output:
# hypertable_name         | owner | ...
# riverflows_riverflow    | ...   | ...
# riverflows_dailyriverflow| ...   | ...

# 4. Test a query
python manage.py shell
>>> from riverflows.sql_queries import FlowQueries
>>> data = FlowQueries.get_raw_flow_data(1, 1230768000, 1230854400, ['flow'])
>>> len(data)

10.3 Deployment Steps

Update Backend:

# SSH to VM
cd /opt/riverflows-api-drf
git pull origin main

# Install new dependencies
source venv/bin/activate
pip install -r requirements.txt

# Run migrations
python manage.py migrate

# Restart Django/Gunicorn
sudo systemctl restart gunicorn

# Start query worker
sudo systemctl enable flows-query-worker
sudo systemctl start flows-query-worker

Update Frontend:

# Local development
cd flows-frontend-react
# Update API calls to include authentication
# Build and deploy
npm run build
# Deploy to Azure Static Web App (via GitHub Actions or manual)

Create Azure Storage Resources:

# Create storage account (if doesn't exist)
az storage account create \
  --name orsancoflows \
  --resource-group orsanco-rg \
  --location eastus \
  --sku Standard_LRS

# Get connection string
az storage account show-connection-string \
  --name orsancoflows \
  --resource-group orsanco-rg

# Create blob container
az storage container create \
  --name query-results \
  --account-name orsancoflows \
  --public-access off

# Table is created automatically by code

Testing:

Test authentication endpoint
Submit small test query
Verify query appears in Table Storage
Verify worker processes query
Verify CSV uploads to Blob Storage
Verify download URL works

10.4 Rollback Plan

If issues arise:

# 1. Stop query worker
sudo systemctl stop flows-query-worker

# 2. Restore database backup
psql -U postgres -d riverflows_db < backup_YYYYMMDD.sql

# 3. Revert code
git revert <commit-hash>

# 4. Restart services
sudo systemctl restart gunicorn

11. Testing Strategy

11.1 Unit Tests

# riverflows/tests/test_sql_queries.py

from django.test import TestCase
from riverflows.sql_queries import FlowQueries
from riverflows.models import Station, River, Reach, RiverFlow

class FlowQueriesTestCase(TestCase):
  def setUp(self):
    # Create test data
    river = River.objects.create(name="Test River")
    reach = Reach.objects.create(name="Test Reach", river=river)
    self.station = Station.objects.create(
      name="Test Station",
      river=river,
      reach=reach,
      hdf_index=1,
      x_coord=0, y_coord=0
    )

    # Insert test flow data
    for i in range(100):
      RiverFlow.objects.create(
        time=1230768000 + (i * 900),  # 15-min increments
        station=self.station,
        flow=1000 + i,
        velocity=2.5,
        stage=450.0
      )

  def test_get_raw_flow_data(self):
    data = FlowQueries.get_raw_flow_data(
      self.station.id,
      1230768000,
      1230771600,  # 1 hour
      ['flow']
    )

    self.assertEqual(len(data), 4)  # 4 readings in 1 hour
    self.assertIn('flow', data[0])

  def test_get_aggregated_flow_data_daily(self):
    data = FlowQueries.get_aggregated_flow_data(
      self.station.id,
      1230768000,
      1230854400,  # 1 day
      'daily',
      ['flow'],
      ['avg', 'min', 'max']
    )

    self.assertGreater(len(data), 0)
    self.assertIn('flow_avg', data[0])
    self.assertIn('flow_min', data[0])
    self.assertIn('flow_max', data[0])

11.2 Integration Tests

# riverflows/tests/test_bulk_download.py

from django.test import TestCase
from rest_framework.test import APIClient
from django.contrib.auth.models import User

class BulkDownloadTestCase(TestCase):
  def setUp(self):
    self.client = APIClient()
    self.user = User.objects.create_user('testuser', password='testpass')

    # Get auth token
    response = self.client.post('/api/auth/token/', {
      'username': 'testuser',
      'password': 'testpass'
    })
    self.token = response.json()['token']
    self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token}')

  def test_submit_bulk_download(self):
    response = self.client.post('/api/bulk-download/', {
      'station_ids': [1],
      'start_time': 1230768000,
      'end_time': 1230854400,
      'parameters': ['flow'],
      'aggregation': {'type': 'none'}
    })

    self.assertEqual(response.status_code, 200)
    self.assertIn('query_id', response.json())
    self.assertEqual(response.json()['status'], 'queued')

  def test_get_query_status(self):
    # First submit a query
    submit_response = self.client.post('/api/bulk-download/', { ... })
    query_id = submit_response.json()['query_id']

    # Then check status
    status_response = self.client.get(f'/api/bulk-download/{query_id}/status/')

    self.assertEqual(status_response.status_code, 200)
    self.assertIn('status', status_response.json())

11.3 Load Testing

Use Apache Bench or Locust to test:

Concurrent API requests
Large query processing
Database performance under load

# Test API endpoint
ab -n 100 -c 10 -H "Authorization: Token xxx" \
  http://localhost:8000/api/stations/

# Test query submission
# Use Locust for more complex scenarios

12. Future Enhancements (Not in Current Scope)

12.1 Azure Queue Storage

Replace polling with queue-based processing:

Frontend submits query → added to Azure Queue
Worker listens to queue → processes messages
Benefits: Better scalability, retry logic, DLQ

12.2 Azure Functions

Move query processing to serverless:

Query submission triggers Azure Function
Function reads from queue, processes, writes to Blob
Benefits: Auto-scaling, pay-per-execution

12.3 Continuous Aggregates

Replace DailyRiverFlow table with TimescaleDB continuous aggregates:

Auto-updating materialized views
Always up-to-date without manual aggregation
Better performance

12.4 GraphQL API

Alternative to REST for more flexible querying:

Single endpoint for all queries
Client specifies exact fields needed
Reduces over-fetching

12.5 Public API

Open API to partners/public:

Rate limiting per API key
Usage tracking and quotas
Documentation portal

13. Appendix

13.1 Database Schema Diagram

13.2 API Endpoint Summary

Endpoint	Method	Auth	Purpose
`/api/auth/token/`	POST	No	Obtain auth token
`/api/auth/refresh/`	POST	Yes	Refresh token
`/api/rivers/`	GET	Yes	List rivers
`/api/reaches/`	GET	Yes	List reaches
`/api/stations/`	GET	Yes	List/search stations
`/api/daily-flows/`	GET	Yes	Get daily aggregates
`/api/bulk-download/`	POST	Yes	Submit bulk query
`/api/bulk-download/{id}/status/`	GET	Yes	Get query status
`/api/bulk-download/queries/`	GET	Yes	List user queries
`/api/bulk-download/{id}/`	DELETE	Yes	Delete query

13.3 Environment Variables Reference

# Django
SECRET_KEY="..."
DEBUG=False
ALLOWED_HOSTS="flows.orsanco.org,localhost"

# Database
DB_NAME="riverflows_db"
DB_USER="riverflows_user"
DB_PASSWORD="..."
DB_HOST="localhost"
DB_PORT="5432"

# Azure Storage
AZURE_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;..."
AZURE_BLOB_CONTAINER_NAME="query-results"
AZURE_BLOB_SAS_EXPIRY_HOURS="720"

# SFTP (existing)
SFTP_HOST="..."
SFTP_USERNAME="..."
SFTP_PASSWORD="..."

13.4 TimescaleDB Useful Queries

-- Check hypertable info
SELECT * FROM timescaledb_information.hypertables;

-- Check chunk sizes
SELECT * FROM timescaledb_information.chunks;

-- Check compression status
SELECT * FROM timescaledb_information.compression_settings;

-- Manually compress old chunks
SELECT compress_chunk(i) FROM show_chunks('riverflows_riverflow', older_than => INTERVAL '30 days') i;

-- Check continuous aggregates
SELECT * FROM timescaledb_information.continuous_aggregates;

-- Get table size
SELECT pg_size_pretty(hypertable_size('riverflows_riverflow'));

Document Version: 1.0 Last Updated: November 12, 2025 Author: Jake Status: Draft for Review Next Steps: Review with team, begin Phase 1 implementation

ORSANCO Flows - Technical Specifications Document

1. Executive Summary​

2. System Architecture​

2.1 Current Architecture​

2.2 Enhanced Architecture (This Upgrade)​

2.3 Future Architecture (Not in Current Scope)​

3. Database Schema & TimescaleDB Implementation​

3.1 Current Schema Issues​

3.2 Proper TimescaleDB Implementation​

Migration 0002: Convert to Hypertables​

3.3 Updated Models​

3.4 Raw SQL Query Module​

4. API Endpoints​

4.1 Authentication​

4.1.1 Token Authentication Endpoint​

4.2 Existing Endpoints (Enhanced)​

4.2.1 Get Rivers​

4.2.2 Get Reaches​

4.2.3 Get Stations​

4.2.4 Get Daily Flows (Existing)​

4.3 New Bulk Download Endpoints​

4.3.1 Submit Bulk Download Query​

4.3.2 Get Query Status​

4.3.3 List User Queries​

4.3.4 Delete Query​

5. Azure Storage Integration​

5.1 Azure Blob Storage (CSV Results)​

5.2 Azure Table Storage (Query Metadata)​

6. Query Processing Worker​

6.1 Worker Architecture​

6.2 Worker Implementation​

6.3 Running the Worker​

7. Performance Optimization​

7.1 Database Query Optimization​

7.2 Query Worker Optimization​

7.3 Frontend Optimization​

8. Security Considerations​

8.1 Authentication​

8.2 Authorization​

8.3 Input Validation​

8.4 Azure Storage Security​

8.5 Rate Limiting​

9. Monitoring & Logging​

9.1 Logging Configuration​

9.2 Metrics to Monitor​

10. Deployment Plan​

10.1 Prerequisites​

10.2 Database Migration Steps​

10.3 Deployment Steps​

10.4 Rollback Plan​

11. Testing Strategy​

11.1 Unit Tests​

11.2 Integration Tests​

11.3 Load Testing​

12. Future Enhancements (Not in Current Scope)​

12.1 Azure Queue Storage​

12.2 Azure Functions​

12.3 Continuous Aggregates​

12.4 GraphQL API​

12.5 Public API​

13. Appendix​

13.1 Database Schema Diagram​

13.2 API Endpoint Summary​

13.3 Environment Variables Reference​

13.4 TimescaleDB Useful Queries​

1. Executive Summary

2. System Architecture

2.1 Current Architecture

2.2 Enhanced Architecture (This Upgrade)

2.3 Future Architecture (Not in Current Scope)

3. Database Schema & TimescaleDB Implementation

3.1 Current Schema Issues

3.2 Proper TimescaleDB Implementation

Migration 0002: Convert to Hypertables

3.3 Updated Models

3.4 Raw SQL Query Module

4. API Endpoints

4.1 Authentication

4.1.1 Token Authentication Endpoint

4.2 Existing Endpoints (Enhanced)

4.2.1 Get Rivers

4.2.2 Get Reaches

4.2.3 Get Stations

4.2.4 Get Daily Flows (Existing)

4.3 New Bulk Download Endpoints

4.3.1 Submit Bulk Download Query

4.3.2 Get Query Status

4.3.3 List User Queries

4.3.4 Delete Query

5. Azure Storage Integration

5.1 Azure Blob Storage (CSV Results)

5.2 Azure Table Storage (Query Metadata)

6. Query Processing Worker

6.1 Worker Architecture

6.2 Worker Implementation

6.3 Running the Worker

7. Performance Optimization

7.1 Database Query Optimization

7.2 Query Worker Optimization

7.3 Frontend Optimization

8. Security Considerations

8.1 Authentication

8.2 Authorization

8.3 Input Validation

8.4 Azure Storage Security

8.5 Rate Limiting

9. Monitoring & Logging

9.1 Logging Configuration

9.2 Metrics to Monitor

10. Deployment Plan

10.1 Prerequisites

10.2 Database Migration Steps

10.3 Deployment Steps

10.4 Rollback Plan

11. Testing Strategy

11.1 Unit Tests

11.2 Integration Tests

11.3 Load Testing

12. Future Enhancements (Not in Current Scope)

12.1 Azure Queue Storage

12.2 Azure Functions

12.3 Continuous Aggregates

12.4 GraphQL API

12.5 Public API

13. Appendix

13.1 Database Schema Diagram

13.2 API Endpoint Summary

13.3 Environment Variables Reference

13.4 TimescaleDB Useful Queries