⚡ Bolt: [performance improvement] Implement serialization caching and blockchain audit logs

backend/cache.py

@@ -184,3 +184,7 @@ def invalidate(self):
 audit_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=2)
 closure_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=1)
 user_issues_cache = ThreadSafeCache(ttl=300, max_size=50) # 5 minutes TTL
+grievances_list_cache = ThreadSafeCache(ttl=60, max_size=50) # 1 minute TTL
+grievance_stats_cache = ThreadSafeCache(ttl=300, max_size=10) # 5 minutes TTL
+visit_stats_cache = ThreadSafeCache(ttl=300, max_size=10) # 5 minutes TTL
+evidence_audit_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=1)

backend/init_db.py

@@ -259,6 +259,19 @@ def index_exists(table, index_name):
                     conn.execute(text("ALTER TABLE resolution_proof_tokens ADD COLUMN valid_until DATETIME"))
                     logger.info("Added valid_until column to resolution_proof_tokens")
 
+            # Evidence Audit Logs Table Migrations
+            if inspector.has_table("evidence_audit_logs"):
+                if not column_exists("evidence_audit_logs", "integrity_hash"):
+                    conn.execute(text("ALTER TABLE evidence_audit_logs ADD COLUMN integrity_hash VARCHAR"))
+                    logger.info("Added integrity_hash column to evidence_audit_logs")
+
+                if not column_exists("evidence_audit_logs", "previous_integrity_hash"):
+                    conn.execute(text("ALTER TABLE evidence_audit_logs ADD COLUMN previous_integrity_hash VARCHAR"))
+                    logger.info("Added previous_integrity_hash column to evidence_audit_logs")
+
+                if not index_exists("evidence_audit_logs", "ix_evidence_audit_logs_previous_integrity_hash"):
+                    conn.execute(text("CREATE INDEX IF NOT EXISTS ix_evidence_audit_logs_previous_integrity_hash ON evidence_audit_logs (previous_integrity_hash)"))
+
             logger.info("Database migration check completed successfully.")
 
     except Exception as e:

backend/models.py

@@ -336,5 +336,9 @@ class EvidenceAuditLog(Base):
     actor_email = Column(String, nullable=True)
     timestamp = Column(DateTime, default=lambda: datetime.datetime.now(datetime.timezone.utc), index=True)
 
+    # Blockchain integrity fields
+    integrity_hash = Column(String, nullable=True)
+    previous_integrity_hash = Column(String, nullable=True, index=True)
+
     # Relationship
     evidence = relationship("ResolutionEvidence", back_populates="audit_logs")

backend/resolution_proof_service.py

@@ -25,7 +25,7 @@
     EvidenceAuditLog, VerificationStatus, GrievanceStatus
 )
 from backend.config import get_config
-from backend.cache import resolution_last_hash_cache
+from backend.cache import resolution_last_hash_cache, evidence_audit_last_hash_cache
 
 logger = logging.getLogger(__name__)
 
@@ -604,16 +604,37 @@ def _create_audit_log(
         actor_email: str,
         db: Session
     ) -> EvidenceAuditLog:
-        """Create an append-only audit log entry."""
+        """
+        Create an append-only audit log entry with blockchain integrity.
+        Optimized: Uses evidence_audit_last_hash_cache for O(1) chaining.
+        """
+        # Blockchain feature: calculate integrity hash for the audit log
+        prev_hash = evidence_audit_last_hash_cache.get("last_hash")
+        if prev_hash is None:
+            # Cache miss: Fetch only the last hash from DB
+            last_audit = db.query(EvidenceAuditLog.integrity_hash).order_by(EvidenceAuditLog.id.desc()).first()
+            prev_hash = last_audit[0] if last_audit and last_audit[0] else ""
+            evidence_audit_last_hash_cache.set(data=prev_hash, key="last_hash")
+
+        # Chaining logic: hash(evidence_id|action|actor_email|prev_hash)
+        hash_content = f"{evidence_id}|{action}|{actor_email}|{prev_hash}"
+        integrity_hash = hashlib.sha256(hash_content.encode()).hexdigest()
+
         log = EvidenceAuditLog(
             evidence_id=evidence_id,
             action=action,
             details=details,
             actor_email=actor_email,
+            integrity_hash=integrity_hash,
+            previous_integrity_hash=prev_hash
         )
         db.add(log)
         db.commit()
         db.refresh(log)
+
+        # Update cache after successful commit
+        evidence_audit_last_hash_cache.set(data=integrity_hash, key="last_hash")
+
         return log
 
     @staticmethod

backend/routers/field_officer.py

@@ -4,12 +4,13 @@
 Issue #288: Field Officer Check-In System With Location Verification
 """
 
-from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, Response
 from sqlalchemy.orm import Session
 from sqlalchemy import func, case
 from typing import List, Optional
 import logging
 import os
+import json
 from datetime import datetime, timezone
 
 from backend.database import get_db
@@ -31,7 +32,7 @@
     calculate_visit_metrics,
     get_geofencing_service
 )
-from backend.cache import visit_last_hash_cache
+from backend.cache import visit_last_hash_cache, visit_stats_cache
 from backend.schemas import BlockchainVerificationResponse
 
 logger = logging.getLogger(__name__)
@@ -424,8 +425,14 @@ def get_visit_statistics(db: Session = Depends(get_db)):
     Get aggregate statistics for all field officer visits using optimized SQL queries
 
     Returns metrics like total visits, verification status, geo-fence compliance, etc.
+    Optimized: Uses serialization caching and a single aggregate SQL query.
     """
     try:
+        # Check cache
+        cached_json = visit_stats_cache.get("default")
+        if cached_json:
+            return Response(content=cached_json, media_type="application/json")
+
         # Optimized: Use a single aggregate query to fetch multiple statistics in one database roundtrip
         stats = db.query(
             func.count(FieldOfficerVisit.id).label('total'),
@@ -449,14 +456,20 @@ def get_visit_statistics(db: Session = Depends(get_db)):
         else:
             average_distance = 0.0
 
-        return VisitStatsResponse(
-            total_visits=total_visits,
-            verified_visits=verified_visits,
-            within_geofence_count=within_geofence_count,
-            outside_geofence_count=outside_geofence_count,
-            unique_officers=unique_officers,
-            average_distance_from_site=average_distance
-        )
+        stats_data = {
+            "total_visits": total_visits,
+            "verified_visits": verified_visits,
+            "within_geofence_count": within_geofence_count,
+            "outside_geofence_count": outside_geofence_count,
+            "unique_officers": unique_officers,
+            "average_distance_from_site": average_distance
+        }
+
+        # Cache serialized JSON to bypass Pydantic overhead on hits
+        json_data = json.dumps(stats_data)
+        visit_stats_cache.set(json_data, "default")
+
+        return Response(content=json_data, media_type="application/json")
 
     except Exception as e:
         logger.error(f"Error calculating visit statistics: {e}", exc_info=True)

backend/routers/grievances.py

@@ -21,6 +21,8 @@
     ClosureStatusResponse,
     BlockchainVerificationResponse
 )
+from backend.cache import grievances_list_cache, grievance_stats_cache
+from fastapi import Response
 from backend.grievance_service import GrievanceService
 from backend.closure_service import ClosureService
 
@@ -38,9 +40,15 @@ def get_grievances(
 ):
     """
     Get list of grievances with escalation history.
-    Optimized: Uses selectinload for audit_logs to avoid Cartesian product and improve O(N) fetching.
+    Optimized: Uses serialization caching and selectinload for audit_logs.
     """
     try:
+        # Check cache
+        cache_key = f"grievances_{status}_{category}_{limit}_{offset}"
+        cached_json = grievances_list_cache.get(cache_key)
+        if cached_json:
+            return Response(content=cached_json, media_type="application/json")
+
         query = db.query(Grievance).options(
             selectinload(Grievance.audit_logs),
             joinedload(Grievance.jurisdiction)
@@ -53,41 +61,45 @@ def get_grievances(
 
         grievances = query.offset(offset).limit(limit).all()
 
-        # Convert to response format
+        # Convert to response format (dictionaries for faster JSON serialization)
         result = []
         for grievance in grievances:
             escalation_history = [
-                EscalationAuditResponse(
-                    id=audit.id,
-                    grievance_id=audit.grievance_id,
-                    previous_authority=audit.previous_authority,
-                    new_authority=audit.new_authority,
-                    timestamp=audit.timestamp,
-                    reason=audit.reason.value
-                )
+                {
+                    "id": audit.id,
+                    "grievance_id": audit.grievance_id,
+                    "previous_authority": audit.previous_authority,
+                    "new_authority": audit.new_authority,
+                    "timestamp": audit.timestamp.isoformat() if audit.timestamp else None,
+                    "reason": audit.reason.value
+                }
                 for audit in grievance.audit_logs
             ]
 
-            result.append(GrievanceSummaryResponse(
-                id=grievance.id,
-                unique_id=grievance.unique_id,
-                category=grievance.category,
-                severity=grievance.severity.value,
-                pincode=grievance.pincode,
-                city=grievance.city,
-                district=grievance.district,
-                state=grievance.state,
-                current_jurisdiction_id=grievance.current_jurisdiction_id,
-                assigned_authority=grievance.assigned_authority,
-                sla_deadline=grievance.sla_deadline,
-                status=grievance.status.value,
-                created_at=grievance.created_at,
-                updated_at=grievance.updated_at,
-                resolved_at=grievance.resolved_at,
-                escalation_history=escalation_history
-            ))
-
-        return result
+            result.append({
+                "id": grievance.id,
+                "unique_id": grievance.unique_id,
+                "category": grievance.category,
+                "severity": grievance.severity.value,
+                "pincode": grievance.pincode,
+                "city": grievance.city,
+                "district": grievance.district,
+                "state": grievance.state,
+                "current_jurisdiction_id": grievance.current_jurisdiction_id,
+                "assigned_authority": grievance.assigned_authority,
+                "sla_deadline": grievance.sla_deadline.isoformat() if grievance.sla_deadline else None,
+                "status": grievance.status.value,
+                "created_at": grievance.created_at.isoformat() if grievance.created_at else None,
+                "updated_at": grievance.updated_at.isoformat() if grievance.updated_at else None,
+                "resolved_at": grievance.resolved_at.isoformat() if grievance.resolved_at else None,
+                "escalation_history": escalation_history
+            })
+
+        # Cache serialized JSON to bypass Pydantic overhead on hits
+        json_data = json.dumps(result)
+        grievances_list_cache.set(json_data, cache_key)
+
+        return Response(content=json_data, media_type="application/json")
 
     except Exception as e:
         logger.error(f"Error getting grievances: {e}", exc_info=True)
@@ -149,9 +161,14 @@ def get_grievance(grievance_id: int, db: Session = Depends(get_db)):
 def get_escalation_stats(db: Session = Depends(get_db)):
     """
     Get escalation statistics.
-    Optimized: Uses a single GROUP BY query instead of 4 separate count queries.
+    Optimized: Uses serialization caching and a single GROUP BY query.
     """
     try:
+        # Check cache
+        cached_json = grievance_stats_cache.get("default")
+        if cached_json:
+            return Response(content=cached_json, media_type="application/json")
+
         # Perform aggregation in a single query for performance
         status_counts = db.query(
             Grievance.status,
@@ -168,13 +185,19 @@ def get_escalation_stats(db: Session = Depends(get_db)):
 
         escalation_rate = (escalated_grievances / total_grievances * 100) if total_grievances > 0 else 0
 
-        return EscalationStatsResponse(
-            total_grievances=total_grievances,
-            escalated_grievances=escalated_grievances,
-            active_grievances=active_grievances,
-            resolved_grievances=resolved_grievances,
-            escalation_rate=escalation_rate
-        )
+        stats_data = {
+            "total_grievances": total_grievances,
+            "escalated_grievances": escalated_grievances,
+            "active_grievances": active_grievances,
+            "resolved_grievances": resolved_grievances,
+            "escalation_rate": escalation_rate
+        }
+
+        # Cache serialized JSON to bypass Pydantic overhead on hits
+        json_data = json.dumps(stats_data)
+        grievance_stats_cache.set(json_data, "default")
+
+        return Response(content=json_data, media_type="application/json")
 
     except Exception as e:
         logger.error(f"Error getting escalation stats: {e}", exc_info=True)