Implementation Plan

Overview

Complete implementation plan for ImpactPulse enhancements. Since this is a pet project, we can start fresh without migration concerns.

Implementation Phases

Phase 1: Database & Core Models

Tasks

1. Setup MongoDB Collections

   • Create all new collections
   • Set up indexes for performance
   • Configure TTL indexes for cleanup
   • Test connection pooling

2. Implement Core Models

   • User model with organization support
   • Organization model
   • Organization members model
   • Survey model with multi-attempt settings
   • Question model with Maslow categorization
   • Survey session model
   • Survey response model
   • Maslow scores model
   • Email reminder model

3. Update Authentication

   • Make authentication mandatory for surveys
   • Add organization context to JWT
   • Implement email verification
   • Add refresh token support
   • Remove anonymous survey capability

Testing

# Unit tests
pytest tests/models/ -n 32

# Integration tests
pytest tests/integration/database/ -n 16

Phase 2: Organization Management

Tasks

1. Organization CRUD

   • Create organization endpoint
   • List user's organizations
   • Get organization details
   • Update organization settings
   • Delete organization (soft delete)

2. Member Management

   • Invite members endpoint
   • Accept invitation flow
   • List organization members
   • Update member roles
   • Remove members

3. Permissions System

   • Role-based access control (RBAC)
   • Organization-scoped permissions
   • Decorator-based authorization
   • Permission middleware

Testing

# Parallel testing for organizations
pytest tests/organizations/ -n 32

# Load testing
locust -f tests/load/organizations.py --users 100 --spawn-rate 10

Phase 3: Survey Sessions & Multi-Attempts

Tasks

1. Session Management

   • Create session on survey start
   • Track attempt numbers
   • Auto-save partial responses
   • Resume capability
   • Session expiry handling

2. Progress Tracking

   • Question-by-question progress
   • Time tracking
   • Completion percentage
   • Status management

3. Attempt Comparison

   • Store all attempts
   • Calculate deltas
   • Generate insights
   • Comparison API endpoints

4. Maslow Scoring

   • Calculate scores per category
   • Historical score tracking
   • Percentile calculations
   • Trend analysis

Testing

# Session concurrency tests
pytest tests/sessions/ -n 32

# E2E survey flow
pytest tests/e2e/survey_flow.py

Phase 4: Email & Reminder System

Tasks

1. Email Service Integration

   • Resend API setup
   • Email template system
   • Multi-language support preparation
   • HTML email rendering

2. Background Jobs

   • Celery setup with Redis
   • Task scheduling
   • Retry logic
   • Dead letter queue

3. Reminder Logic

   • Schedule reminders (24h, 48h, 7d)
   • Track email status
   • Handle opt-outs
   • Click/open tracking

Background Job Setup

from celery import Celery
from celery.schedules import crontab

app = Celery('impactpulse', broker='redis://localhost:6379')

app.conf.beat_schedule = {
    'send-reminders': {
        'task': 'tasks.send_pending_reminders',
        'schedule': crontab(minute='*/15'),
    },
    'cleanup-sessions': {
        'task': 'tasks.cleanup_abandoned_sessions',
        'schedule': crontab(hour=2, minute=0),
    }
}

Phase 5: Frontend Updates

Tasks

1. Authentication Flow

   • Login/Register pages
   • Email verification UI
   • Password reset flow
   • Session management

2. Organization UI

   • Organization dashboard
   • Member management UI
   • Invitation flow
   • Organization switcher

3. Survey Experience

   • Progress indicators
   • Auto-save feedback
   • Resume UI
   • Attempt selector

4. Analytics & Comparison

   • Comparison charts
   • Progress visualizations
   • Maslow pyramid display
   • Export functionality

Frontend Structure

/frontend/src/
  /features/
    /auth/
    /organizations/
    /surveys/
    /analytics/
  /components/
    /shared/
    /charts/
    /forms/

Code Implementation Examples

Organization Service

class OrganizationService:
    def __init__(self, db: AsyncIOMotorDatabase):
        self.db = db

    async def create_organization(
        self,
        org_data: OrganizationCreate,
        user_id: str
    ) -> Organization:
        """Create new organization and make user the owner."""
        slug = await self._generate_unique_slug(org_data.name)

        org_doc = {
            "name": org_data.name,
            "slug": slug,
            "description": org_data.description,
            "created_by_id": user_id,
            "created_at": datetime.utcnow(),
            "settings": org_data.settings or {},
            "subscription_tier": "free"
        }

        result = await self.db.organizations.insert_one(org_doc)

        # Add user as owner
        await self.db.organization_members.insert_one({
            "organization_id": result.inserted_id,
            "user_id": user_id,
            "role": "owner",
            "joined_at": datetime.utcnow()
        })

        return Organization(**org_doc, id=str(result.inserted_id))

Session Management

class SessionService:
    async def create_session(
        self,
        survey_id: str,
        user_id: str,
        organization_id: Optional[str] = None
    ) -> SurveySession:
        """Create new survey session."""
        attempt_number = await self._get_next_attempt_number(user_id, survey_id)
        await self._check_attempt_restrictions(survey_id, user_id, attempt_number)

        session_doc = {
            "user_id": user_id,
            "survey_id": survey_id,
            "organization_id": organization_id,
            "attempt_number": attempt_number,
            "status": "in_progress",
            "started_at": datetime.utcnow(),
            "last_activity_at": datetime.utcnow(),
            "current_question_index": 0,
            "partial_responses": {},
            "time_spent_seconds": 0
        }

        result = await self.db.survey_sessions.insert_one(session_doc)
        await self._schedule_reminders(result.inserted_id, user_id)

        return SurveySession(**session_doc, id=str(result.inserted_id))

    async def save_progress(
        self,
        session_id: str,
        responses: dict,
        current_index: int
    ) -> None:
        """Auto-save survey progress."""
        await self.db.survey_sessions.update_one(
            {"_id": session_id},
            {
                "$set": {
                    "partial_responses": responses,
                    "current_question_index": current_index,
                    "last_activity_at": datetime.utcnow()
                },
                "$inc": {"time_spent_seconds": 30}
            }
        )

Testing Strategy

Unit Testing

@pytest.mark.asyncio
async def test_create_organization(test_db, test_user):
    """Test organization creation."""
    service = OrganizationService(test_db)

    org_data = OrganizationCreate(
        name="Test Org",
        description="Test Description"
    )

    org = await service.create_organization(org_data, test_user.id)

    assert org.name == "Test Org"
    assert org.slug == "test-org"

    member = await test_db.organization_members.find_one({
        "organization_id": org.id,
        "user_id": test_user.id
    })
    assert member["role"] == "owner"

Integration Testing

@pytest.mark.asyncio
async def test_complete_survey_flow(client, test_user):
    """Test complete survey flow with multiple attempts."""
    token = await login_user(client, test_user)

    # Start survey (attempt 1)
    response = await client.post(
        "/api/v1/surveys/maslow_survey_id/start",
        headers={"Authorization": f"Bearer {token}"}
    )
    session1 = response.json()["data"]

    # Save progress
    await save_progress(client, session1["session_id"], responses_partial)

    # Complete survey
    await complete_survey(client, session1["session_id"], responses_full)

    # Start second attempt
    response = await client.post(
        "/api/v1/surveys/maslow_survey_id/start",
        headers={"Authorization": f"Bearer {token}"}
    )
    session2 = response.json()["data"]

    assert session2["attempt_number"] == 2

Deployment Checklist

Pre-deployment

• All tests passing (unit, integration, E2E)
• Performance benchmarks met
• Security audit completed
• Documentation updated
• Environment variables configured

Deployment

• Database indexes created
• Redis configured
• Celery workers running
• Email service configured
• Monitoring enabled

Post-deployment

• Smoke tests passing
• Metrics flowing
• Email delivery verified
• Performance monitoring
• Error tracking active

Success Criteria

Phase 1 Success

• MongoDB collections created and indexed
• Authentication required for all surveys
• Email verification working

Phase 2 Success

• Organizations can be created and managed
• Members can be invited and join
• Role-based permissions enforced

Phase 3 Success

• Users can take surveys multiple times
• Progress is auto-saved
• Attempts can be compared

Phase 4 Success

• Reminders sent on schedule
• Email delivery rate > 95%
• Users can manage preferences

Phase 5 Success

• All UI components responsive
• E2E tests passing
• User feedback positive