added latest models

app/routes/forms.py

@@ -703,13 +703,16 @@ async def chat_stream(
         except Exception as e:
             logger.warning(f"Could not load responses: {e}")
     
+    # Capture user ID as plain int to avoid session detachment issues
+    user_id_for_save = current_user.id
+    
     # Determine chat type based on whether this is likely a response analysis
     chat_type = "form_editing"  # Default
     
     # Save user message to database
     user_chat_message = ChatMessageModel(
         form_id=form_id,
-        user_id=current_user.id,
+        user_id=user_id_for_save,
         chat_type=chat_type,
         role="user",
         content=chat_data.message
@@ -929,45 +932,42 @@ async def chat_stream(
             logger.error(f"Stream error: {e}", exc_info=True)
             collected_response['content'] = f"Error: {str(e)}"
             yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n"
-        finally:
-            if conn:
-                conn.close()
+        # Note: Don't close conn - it's cached for session reuse
     
-    async def save_assistant_message_background():
-        """Save assistant message after stream completes"""
-        # Create a new db session for background task
+    def save_assistant_message_sync(form_id_val: int, user_id_val: int, response_dict: dict):
+        """Save assistant message in a completely isolated session"""
         from ..core.db import SessionLocal
+        session = None
         try:
-            background_db = SessionLocal()
-            if collected_response.get('content'):
-                # Update user message with correct chat_type if it was response analysis
-                if collected_response.get('route') == 'analyze_responses':
-                    user_chat_message.chat_type = "response_analysis"
-                    background_db.merge(user_chat_message)
-                
-                assistant_message = ChatMessageModel(
-                    form_id=form_id,
-                    user_id=current_user.id,
-                    chat_type="response_analysis" if collected_response.get('route') == 'analyze_responses' else "form_editing",
+            session = SessionLocal()
+            if response_dict.get('content'):
+                msg = ChatMessageModel(
+                    form_id=form_id_val,
+                    user_id=user_id_val,
+                    chat_type="response_analysis" if response_dict.get('route') == 'analyze_responses' else "form_editing",
                     role="assistant",
-                    content=collected_response.get('content', ''),
-                    query_type=collected_response.get('query_type'),
-                    sql_query=collected_response.get('sql_query'),
-                    result_data=collected_response.get('result_data')
+                    content=response_dict.get('content', ''),
+                    query_type=response_dict.get('query_type'),
+                    sql_query=response_dict.get('sql_query'),
+                    result_data=response_dict.get('result_data')
                 )
-                background_db.add(assistant_message)
-                background_db.commit()
+                session.add(msg)
+                session.commit()
+                logger.debug(f"Saved assistant message for form {form_id_val}")
         except Exception as e:
             logger.error(f"Failed to save chat message: {e}")
+            if session:
+                session.rollback()
         finally:
-            background_db.close()
+            if session:
+                session.close()
     
     async def streaming_with_save():
         """Wrap generator to save message after completion"""
         async for event in event_generator():
             yield event
-        # Save after streaming completes
-        await save_assistant_message_background()
+        # Save after streaming completes - pass values explicitly to avoid closure issues
+        save_assistant_message_sync(form_id, user_id_for_save, collected_response.copy())
     
     return StreamingResponse(
         streaming_with_save(),
@@ -1109,8 +1109,7 @@ async def chat_edit_form(
                     if status_breakdown:
                         assistant_response += f"**Status:** {', '.join([f'{k}: {v}' for k, v in status_breakdown.items()])}"
                 
-                if conn:
-                    conn.close()
+                # Note: Don't close conn - it's cached for session reuse
                 
                 return ChatResponse(
                     route="analyze_responses",
@@ -1123,17 +1122,14 @@ async def chat_edit_form(
                 
             except Exception as e:
                 logger.error(f"Response analysis failed: {e}", exc_info=True)
-                if conn:
-                    conn.close()
+                # Note: Don't close conn - it's cached for session reuse
                 return ChatResponse(
                     route="analyze_responses",
                     response={"message": f"Error analyzing responses: {str(e)}"},
                     changes_made=None
                 )
         
-        # Close conn if opened but not used
-        if conn:
-            conn.close()
+        # Note: Don't close conn - it's cached for session reuse
         
         # === Handle No Responses ===
         if route == 'no_responses':

app/routes/responses.py

@@ -27,6 +27,7 @@ from ..services.validation_service import validation_service
 from ..services.analytics_service import analytics_service
 from ..services.webhook_service import webhook_service
 from ..services.s3_service import s3_service
+from ..services.response_chat_service import invalidate_duckdb_cache
 from ..middleware.rate_limiter import rate_limiter
 
 import logging
@@ -270,6 +271,9 @@ async def submit_form(
         db.commit()
         db.refresh(form_response)
         
+        # Invalidate DuckDB cache so new data is reflected in analytics
+        invalidate_duckdb_cache(form.id)
+        
         return SubmissionResponse(
             id=form_response.id,
             form_id=form.id,
@@ -408,6 +412,9 @@ async def autosave_submission(
         db.commit()
         db.refresh(form_response)
         
+        # Invalidate DuckDB cache so new data is reflected in analytics
+        invalidate_duckdb_cache(form.id)
+        
         # Validate answers (non-blocking)
         validation_result = validation_service.validate_submission(
             db=db,

app/services/response_chat_service.py

@@ -11,6 +11,7 @@ import json
 import logging
 import os
 import re
+import time
 from typing import Dict, Any, List, Optional, Tuple
 from datetime import datetime
 from sqlalchemy.orm import Session
@@ -20,6 +21,76 @@ from ..models import Form, FormQuestion, FormResponse, ResponseAnswer, ChatMessa
 logger = logging.getLogger(__name__)
 
 
+# Simple session cache for DuckDB connections (per form)
+# Structure: {form_id: {'conn': connection, 'columns': [...], 'question_map': {...}, 'timestamp': float, 'count': int}}
+_duckdb_sessions: Dict[int, Dict[str, Any]] = {}
+_SESSION_TTL = 1200  # 20 minutes
+
+
+def get_cached_duckdb(form_id: int) -> Optional[Dict[str, Any]]:
+    """Get cached DuckDB session if still valid (< 5 mins old)."""
+    if form_id not in _duckdb_sessions:
+        return None
+    
+    session = _duckdb_sessions[form_id]
+    age = time.time() - session['timestamp']
+    
+    if age > _SESSION_TTL:
+        # Expired - remove it
+        try:
+            session['conn'].close()
+        except:
+            pass
+        del _duckdb_sessions[form_id]
+        logger.debug(f"DuckDB session expired for form {form_id} (age: {age:.0f}s)")
+        return None
+    
+    # Verify connection is still valid
+    try:
+        session['conn'].execute("SELECT 1").fetchone()
+        logger.debug(f"Using cached DuckDB session for form {form_id} (age: {age:.0f}s)")
+        return session
+    except Exception as e:
+        logger.warning(f"DuckDB connection invalid for form {form_id}: {e}")
+        try:
+            session['conn'].close()
+        except:
+            pass
+        del _duckdb_sessions[form_id]
+        return None
+
+
+def set_cached_duckdb(form_id: int, conn: duckdb.DuckDBPyConnection, 
+                      columns: List[str], question_map: Dict[int, str], count: int) -> None:
+    """Cache a DuckDB session for a form."""
+    # Close old connection if exists
+    if form_id in _duckdb_sessions:
+        try:
+            _duckdb_sessions[form_id]['conn'].close()
+        except:
+            pass
+    
+    _duckdb_sessions[form_id] = {
+        'conn': conn,
+        'columns': columns,
+        'question_map': question_map,
+        'count': count,
+        'timestamp': time.time()
+    }
+    logger.debug(f"Cached DuckDB session for form {form_id} with {count} responses")
+
+
+def invalidate_duckdb_cache(form_id: int) -> None:
+    """Invalidate cached DuckDB session for a form."""
+    if form_id in _duckdb_sessions:
+        try:
+            _duckdb_sessions[form_id]['conn'].close()
+        except:
+            pass
+        del _duckdb_sessions[form_id]
+        logger.debug(f"Invalidated DuckDB cache for form {form_id}")
+
+
 class ResponseChatService:
     """Service for analyzing form responses via chat interface using DuckDB."""
     
@@ -37,6 +108,25 @@ class ResponseChatService:
         Returns:
             Tuple of (DuckDB connection, column names, question_id to column name mapping)
         """
+        from sqlalchemy import func
+        
+        # Get current response count to check if cache is valid
+        current_count = db.query(func.count(FormResponse.id)).filter(
+            FormResponse.form_id == form_id,
+            FormResponse.status.in_(["complete", "partial"])
+        ).scalar() or 0
+        
+        # Check cache first
+        cached = get_cached_duckdb(form_id)
+        if cached is not None and cached['count'] == current_count:
+            # Cache is valid and count matches
+            return cached['conn'], cached['columns'], cached['question_map']
+        
+        # Cache miss or count changed - need to reload
+        if cached is not None:
+            logger.info(f"Response count changed for form {form_id}: {cached['count']} -> {current_count}, reloading")
+            invalidate_duckdb_cache(form_id)
+        
         # Get form and questions
         form = db.query(Form).filter(Form.id == form_id).first()
         if not form:
@@ -46,11 +136,14 @@ class ResponseChatService:
             FormQuestion.form_id == form_id
         ).order_by(FormQuestion.question_order).all()
         
-        # Get responses
+        # Get responses - only complete and partial
         responses = db.query(FormResponse).filter(
-            FormResponse.form_id == form_id
+            FormResponse.form_id == form_id,
+            FormResponse.status.in_(["complete", "partial"])
         ).all()
         
+        logger.info(f"Loading {len(responses)} responses into DuckDB for form {form_id}")
+        
         # Create column mapping from questions
         # Column names: response_id, submitted_at, status, q1_<sanitized_text>, q2_<sanitized_text>, ...
         question_id_to_col = {}
@@ -102,6 +195,9 @@ class ResponseChatService:
             col_defs = ", ".join([f'"{c}" VARCHAR' for c in columns])
             conn.execute(f"CREATE TABLE responses ({col_defs})")
         
+        # Cache for reuse within session (5 min TTL)
+        set_cached_duckdb(form_id, conn, columns, question_id_to_col, len(responses))
+        
         return conn, columns, question_id_to_col
     
     def _extract_answer_text(self, answer_value: Any, question_type: str) -> str:
@@ -109,7 +205,15 @@ class ResponseChatService:
         if answer_value is None:
             return ""
         
+        # Handle plain numeric types directly (int, float)
+        if isinstance(answer_value, (int, float)):
+            return str(answer_value)
+        
         if isinstance(answer_value, dict):
+            # Empty dict - return empty string
+            if not answer_value:
+                return ""
+            
             # Handle different answer formats
             if "text" in answer_value:
                 val = answer_value["text"]
@@ -117,6 +221,14 @@ class ResponseChatService:
             if "number" in answer_value:
                 val = answer_value["number"]
                 return str(val) if val is not None else ""
+            # Also check for 'value' key (common for sliders/scales)
+            if "value" in answer_value:
+                val = answer_value["value"]
+                return str(val) if val is not None else ""
+            # Check for 'scale' key
+            if "scale" in answer_value:
+                val = answer_value["scale"]
+                return str(val) if val is not None else ""
             if "choices" in answer_value:
                 choices = answer_value["choices"]
                 if choices is None:
@@ -145,7 +257,8 @@ class ResponseChatService:
                 if isinstance(items, list):
                     return " > ".join(str(i) for i in items)
                 return str(items)
-            # Fallback for other dict types
+            # Fallback for other dict types - log and dump
+            logger.debug(f"Unknown answer format for {question_type}: {answer_value}")
             return json.dumps(answer_value)
         
         return str(answer_value)
@@ -166,6 +279,72 @@ class ResponseChatService:
             value_rows.append(f"({', '.join(values)})")
         return ", ".join(value_rows)
     
+    def get_responses_direct(
+        self,
+        db: Session,
+        form_id: int,
+        limit: int = 100
+    ) -> Tuple[List[Dict], List[str], Dict[str, Any]]:
+        """
+        Get responses directly from database (fallback when DuckDB fails).
+        
+        Returns:
+            Tuple of (rows, columns, summary_dict)
+        """
+        # Get form questions for column names
+        questions = db.query(FormQuestion).filter(
+            FormQuestion.form_id == form_id
+        ).order_by(FormQuestion.question_order).all()
+        
+        # Get responses with status filter
+        responses = db.query(FormResponse).filter(
+            FormResponse.form_id == form_id,
+            FormResponse.status.in_(["complete", "partial"])
+        ).order_by(FormResponse.submitted_at.desc()).limit(limit).all()
+        
+        if not responses:
+            return [], ["response_id", "submitted_at", "status"], {"total_responses": 0, "status_breakdown": {}}
+        
+        # Build column names
+        columns = ["response_id", "submitted_at", "status"]
+        question_id_to_col = {}
+        for i, q in enumerate(questions):
+            sanitized = re.sub(r'[^a-zA-Z0-9]', '_', q.question_text[:30]).lower()
+            col_name = f"q{i+1}_{sanitized}"
+            columns.append(col_name)
+            question_id_to_col[q.id] = col_name
+        
+        # Build rows
+        rows = []
+        status_counts = {}
+        for response in responses:
+            # Get answers for this response
+            answers = db.query(ResponseAnswer).filter(
+                ResponseAnswer.form_response_id == response.id
+            ).all()
+            
+            row = {
+                "response_id": str(response.id),
+                "submitted_at": response.submitted_at.isoformat() if response.submitted_at else None,
+                "status": response.status
+            }
+            
+            status_counts[response.status] = status_counts.get(response.status, 0) + 1
+            
+            answer_map = {a.form_question_id: a.answer_value for a in answers}
+            for q in questions:
+                col_name = question_id_to_col[q.id]
+                answer_value = answer_map.get(q.id)
+                row[col_name] = self._extract_answer_text(answer_value, q.question_type.value)
+            rows.append(row)
+        
+        summary = {
+            "total_responses": len(responses),
+            "status_breakdown": status_counts
+        }
+        
+        return rows, columns, summary
+    
     def execute_query(
         self,
         conn: duckdb.DuckDBPyConnection,

app/stripe_routes.py

@@ -83,7 +83,7 @@ def create_checkout_session(
         
         # Known promotion codes mapping (code -> Stripe promotion code ID)
         KNOWN_PROMO_CODES = {
-            "FIRST100": "promo_1SrJbKBACqQSnujJriPjszSv",
+            "NEWYEARS": "promo_1SrJE0BACqQSnujJY6Q9o3X2",
         }
         
         # Apply promo code if provided