Appendix

# Project Feline - Cloud Task Scheduler

## Development Environment
- Use Node.js 18+ with nvm for version management
- Run `npm run setup` for initial environment configuration
- All database changes require migration files in `/migrations`

## Claude Code Guidelines
- Always run tests before committing: `npm test`
- Use TypeScript for all new modules
- Follow our API naming convention: `/api/v2/tasks/{resource}`
- Security: Never commit API keys or database credentials in code

## Architecture Notes
- Task scheduling uses Redis queues for job management
- All external API calls must include retry logic and timeouts
- Database queries should use connection pooling for performance

## Current Sprint Focus
- Implementing distributed task execution
- Focus on fault-tolerance and recovery patterns
- All scheduling logic requires unit tests + integration tests

# Terminal command to Claude Code

claude-code --task "Create comprehensive tests for a user authentication system with registration, login, password reset, and JWT token management"

# tests/test_auth_system.py
import pytest
import jwt
from datetime import datetime, timedelta
from unittest.mock import Mock, patch
from auth_system import AuthSystem, User, AuthenticationError, ValidationError

class TestUserAuthentication:
    @pytest.fixture
    def auth_system(self):
        return AuthSystem(secret_key="test_secret", db_connection=Mock())

    @pytest.fixture
    def valid_user_data(self):
        return {
            "email": "user@example.com",
            "password": "SecurePass123!",
            "confirm_password": "SecurePass123!"
        }

    # Registration Tests
    def test_register_user_success(self, auth_system, valid_user_data):
        """Test successful user registration"""
        with patch.object(auth_system.db, 'user_exists', return_value=False):
            with patch.object(auth_system.db, 'create_user', return_value=True):
                result = auth_system.register(**valid_user_data)
                assert result['success'] == True
                assert result['user_id'] is not None
                assert result['message'] == "User registered successfully"

    def test_register_user_duplicate_email(self, auth_system, valid_user_data):
        """Test registration with existing email"""
        with patch.object(auth_system.db, 'user_exists', return_value=True):
    def test_register_invalid_email_format(self, auth_system):
        """Test registration with invalid email format"""
        invalid_data = {
            "email": "invalid-email",
            "password": "SecurePass123!",
            "confirm_password": "SecurePass123!"
        }
        with pytest.raises(ValidationError) as exc_info:
            auth_system.register(**invalid_data)
        assert "Invalid email format" in str(exc_info.value)

    def test_register_weak_password(self, auth_system):
        """Test registration with weak password"""
        weak_password_data = {
            "email": "user@example.com",
            "password": "123",
            "confirm_password": "123"
        }
        with pytest.raises(ValidationError) as exc_info:
            auth_system.register(**weak_password_data)
        assert "Password must be at least 8 characters" in str(exc_info.value)

    # Login Tests
    def test_login_success(self, auth_system):
        """Test successful login with valid credentials"""
        mock_user = Mock()
        mock_user.id = 1
        mock_user.email = "user@example.com"
        mock_user.password_hash = "$2b$12$hashed_password"

        with patch.object(auth_system.db, 'get_user_by_email', return_value=mock_user):
            with patch.object(auth_system, '_verify_password', return_value=True):
                result = auth_system.login("user@example.com", "correct_password")
                assert result['success'] == True
                assert 'access_token' in result
                assert 'refresh_token' in result
                assert result['user']['email'] == "user@example.com"

    def test_login_invalid_credentials(self, auth_system):
        """Test login with invalid credentials"""
        with patch.object(auth_system.db, 'get_user_by_email', return_value=None):
            with pytest.raises(AuthenticationError) as exc_info:
                auth_system.login("nonexistent@example.com", "any_password")
            assert "Invalid credentials" in str(exc_info.value)

    def test_login_wrong_password(self, auth_system):
        """Test login with correct email but wrong password"""
        mock_user = Mock()
        mock_user.email = "user@example.com"

        with patch.object(auth_system.db, 'get_user_by_email', return_value=mock_user):
            with patch.object(auth_system, '_verify_password', return_value=False):
                with pytest.raises(AuthenticationError) as exc_info:
                    auth_system.login("user@example.com", "wrong_password")
                assert "Invalid credentials" in str(exc_info.value)

    # JWT Token Tests
    def test_generate_access_token(self, auth_system):
        """Test JWT access token generation"""
        user_id = 1
        token = auth_system.generate_access_token(user_id)

        decoded = jwt.decode(token, "test_secret", algorithms=["HS256"])
        assert decoded['user_id'] == user_id
        assert decoded['type'] == 'access'
        assert 'exp' in decoded

    def test_verify_valid_token(self, auth_system):
        """Test verification of valid JWT token"""
        user_id = 1
        token = auth_system.generate_access_token(user_id)

        result = auth_system.verify_token(token)
        assert result['valid'] == True
        assert result['user_id'] == user_id

    def test_verify_expired_token(self, auth_system):
        """Test verification of expired JWT token"""
        # Create token that expires in -1 hour
        payload = {
            'user_id': 1,
            'type': 'access',
            'exp': datetime.utcnow() - timedelta(hours=1)
        }
        expired_token = jwt.encode(payload, "test_secret", algorithm="HS256")
        result = auth_system.verify_token(expired_token)
        assert result['valid'] == False
        assert result['error'] == 'Token expired'

    # Password Reset Tests
    def test_request_password_reset(self, auth_system):
        """Test password reset request"""
        mock_user = Mock()
        mock_user.id = 1
        mock_user.email = "user@example.com"

        with patch.object(auth_system.db, 'get_user_by_email', return_value=mock_user):
            with patch.object(auth_system, '_send_reset_email') as mock_send:
                result = auth_system.request_password_reset("user@example.com")
                assert result['success'] == True
                mock_send.assert_called_once()

    def test_reset_password_success(self, auth_system):
        """Test successful password reset with valid token"""
        reset_token = "valid_reset_token"
        new_password = "NewSecurePass123!"

        with patch.object(auth_system, '_verify_reset_token', return_value={'valid': True, 'user_id': 1}):
            with patch.object(auth_system.db, 'update_password', return_value=True):
                result = auth_system.reset_password(reset_token, new_password)
                assert result['success'] == True
                assert result['message'] == "Password reset successfully"

# Run the tests - they will fail because implementation doesn't exist yet
pytest tests/test_auth_system.py -v

# Expected output: All tests fail with ImportError or similar

claude-code --task "Implement the AuthSystem class to pass all the existing tests. Focus on security best practices for password hashing and JWT token management."

# auth_system.py
import bcrypt
import jwt
import re
from datetime import datetime, timedelta
from typing import Dict, Optional, Any
import secrets
import smtplib
from email.mime.text import MIMEText

class AuthenticationError(Exception):
    """Raised when authentication fails"""
    pass

class ValidationError(Exception):
    """Raised when input validation fails"""
    pass

class User:
    def __init__(self, id: int, email: str, password_hash: str):
        self.id = id
        self.email = email
        self.password_hash = password_hash

class AuthSystem:
    def __init__(self, secret_key: str, db_connection,
                 access_token_expire_hours: int = 1,
                 refresh_token_expire_days: int = 30):
        self.secret_key = secret_key
        self.db = db_connection
        self.access_token_expire_hours = access_token_expire_hours
        self.refresh_token_expire_days = refresh_token_expire_days

        # Password requirements
        self.min_password_length = 8
        self.password_pattern = r'^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[@$!%*?&])[A-Za-z\d@$!%*?&]'

    def register(self, email: str, password: str, confirm_password: str) -> Dict[str, Any]:
        """Register a new user"""
        # Validate input
        self._validate_registration_input(email, password, confirm_password)

        # Check if user already exists
        if self.db.user_exists(email):
            raise ValidationError("Email already registered")

        # Hash password
        password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')

        # Create user in database
        user_id = self.db.create_user(email, password_hash)

        return {
            'success': True,
            'user_id': user_id,
            'message': 'User registered successfully'
        }

    def login(self, email: str, password: str) -> Dict[str, Any]:
        """Authenticate user and return tokens"""
        # Get user from database
        user = self.db.get_user_by_email(email)
        if not user:
            raise AuthenticationError("Invalid credentials")

        # Verify password
        if not self._verify_password(password, user.password_hash):
            raise AuthenticationError("Invalid credentials")

        # Generate tokens
        access_token = self.generate_access_token(user.id)
        refresh_token = self.generate_refresh_token(user.id)

        return {
            'success': True,
            'access_token': access_token,
            'refresh_token': refresh_token,
            'user': {
                'id': user.id,
                'email': user.email
            }
        }

    def generate_access_token(self, user_id: int) -> str:
        """Generate JWT access token"""
        payload = {
            'user_id': user_id,
            'type': 'access',
            'exp': datetime.utcnow() + timedelta(hours=self.access_token_expire_hours),
            'iat': datetime.utcnow()
        }
        return jwt.encode(payload, self.secret_key, algorithm='HS256')

    def generate_refresh_token(self, user_id: int) -> str:
        """Generate JWT refresh token"""
        payload = {
            'user_id': user_id,
            'type': 'refresh',
            'exp': datetime.utcnow() + timedelta(days=self.refresh_token_expire_days),
            'iat': datetime.utcnow()
        }
        return jwt.encode(payload, self.secret_key, algorithm='HS256')

    def verify_token(self, token: str) -> Dict[str, Any]:
        """Verify JWT token"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            return {
                'valid': True,
                'user_id': payload['user_id'],
                'type': payload['type']
            }
        except jwt.ExpiredSignatureError:
            return {'valid': False, 'error': 'Token expired'}
        except jwt.InvalidTokenError:
            return {'valid': False, 'error': 'Invalid token'}

    def request_password_reset(self, email: str) -> Dict[str, Any]:
        """Request password reset"""
        user = self.db.get_user_by_email(email)
        if not user:
            # Don't reveal if email exists or not
            return {'success': True, 'message': 'Reset instructions sent if email exists'}

        # Generate reset token
        reset_token = self._generate_reset_token(user.id)

        # Send reset email
        self._send_reset_email(user.email, reset_token)

        return {'success': True, 'message': 'Reset instructions sent'}

    def reset_password(self, reset_token: str, new_password: str) -> Dict[str, Any]:
        """Reset user password with token"""
        # Verify reset token
        token_data = self._verify_reset_token(reset_token)
        if not token_data['valid']:
            raise AuthenticationError("Invalid or expired reset token")

        # Validate new password
        self._validate_password(new_password)

        # Hash new password
        password_hash = bcrypt.hashpw(new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')

        # Update password in database
        self.db.update_password(token_data['user_id'], password_hash)

        return {'success': True, 'message': 'Password reset successfully'}

    def _validate_registration_input(self, email: str, password: str, confirm_password: str):
        """Validate registration input"""
        if not self._is_valid_email(email):
            raise ValidationError("Invalid email format")

        if password != confirm_password:
            raise ValidationError("Passwords do not match")

        self._validate_password(password)

    def _validate_password(self, password: str):
        """Validate password strength"""
        if len(password) < self.min_password_length:
            raise ValidationError(f"Password must be at least {self.min_password_length} characters")

        if not re.match(self.password_pattern, password):
            raise ValidationError("Password must contain uppercase, lowercase, digit, and special character")

    def _is_valid_email(self, email: str) -> bool:
        """Validate email format"""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email))

    def _verify_password(self, password: str, password_hash: str) -> bool:
        """Verify password against hash"""
        return bcrypt.checkpw(password.encode('utf-8'), password_hash.encode('utf-8'))

    def _generate_reset_token(self, user_id: int) -> str:
        """Generate password reset token"""
        payload = {
            'user_id': user_id,
            'type': 'reset',
            'exp': datetime.utcnow() + timedelta(hours=1),  # 1 hour expiry
            'iat': datetime.utcnow()
        }
        return jwt.encode(payload, self.secret_key, algorithm='HS256')

    def _verify_reset_token(self, token: str) -> Dict[str, Any]:
        """Verify password reset token"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            if payload['type'] != 'reset':
                return {'valid': False, 'error': 'Invalid token type'}
            return {'valid': True, 'user_id': payload['user_id']}
        except jwt.ExpiredSignatureError:
            return {'valid': False, 'error': 'Token expired'}
        except jwt.InvalidTokenError:
            return {'valid': False, 'error': 'Invalid token'}

    def _send_reset_email(self, email: str, reset_token: str):
        """Send password reset email"""
        # In production, implement actual email sending
        # For now, this is a placeholder
        pass

pytest tests/test_auth_system.py -v

# Expected output: All tests pass!

claude-code --task "Review the authentication system for security improvements, add rate limiting for login attempts, and enhance error handling. Maintain all existing test compatibility"

claude-code --task "Debug authentication failure in user dashboard - getting 500 errors on login attempts"

Error: Authentication failed during login process
    at UserController.login (/Users/dev/myapp/controllers/userController.js:45)
    at Layer.handle [as handle_request] (/Users/dev/myapp/node_modules/express/lib/router/layer.js:95)
    at next (/Users/dev/myapp/node_modules/express/lib/router/route.js:144)
    at Route.dispatch (/Users/dev/myapp/node_modules/express/lib/router/route.js:114)
    at Layer.handle [as handle_request] (/Users/dev/myapp/node_modules/express/lib/router/layer.js:95)
    at /Users/dev/myapp/node_modules/express/lib/router/index.js:284
    at Function.process_params (/Users/dev/myapp/node_modules/express/lib/router/index.js:346)
    at next (/Users/dev/myapp/node_modules/express/lib/router/index.js:280)
    at /Users/dev/myapp/middleware/auth.js:23
    at Layer.handle [as handle_request] (/Users/dev/myapp/node_modules/express/lib/router/layer.js:95)

Database connection error in logs:
[2024-01-15 14:23:15] ERROR: connection to server at "localhost" (127.0.0.1), port 5432 failed: FATAL:  password authentication failed for user "myapp_user"
[2024-01-15 14:23:15] ERROR: Query execution failed - relation "users" does not exist

POST http://localhost:3000/api/auth/login 500 (Internal Server Error)
Uncaught (in promise) Error: Request failed with status code 500
    at createError (createError.js:16:1)
    at settle (settle.js:17:1)
    at XMLHttpRequest.onloadend (xhr.js:66:1)

const bcrypt = require('bcryptjs');
const jwt = require('jsonwebtoken');
const User = require('../models/User');

exports.login = async (req, res) => {
  try {
    const { email, password } = req.body;

    // Find user by email
    const user = await User.findByEmail(email);
    if (!user) {
      return res.status(401).json({ message: 'Invalid credentials' });
    }

    // Verify password
    const isValidPassword = await bcrypt.compare(password, user.password_hash);
    if (!isValidPassword) {
      return res.status(401).json({ message: 'Invalid credentials' });
    }

    // Generate JWT token - LINE 45 WHERE ERROR OCCURS
    const token = jwt.sign(
      { userId: user.id, email: user.email },
      process.env.JWT_SECRET,
      { expiresIn: '24h' }
    );

    res.json({
      message: 'Login successful',
      token: token,
      user: { id: user.id, email: user.email }
    });

  } catch (error) {
    console.error('Login error:', error);
    res.status(500).json({ message: 'Authentication failed' });
  }
};

const { Pool } = require('pg');

const pool = new Pool({
  user: process.env.DB_USER || 'myapp_user',
  host: process.env.DB_HOST || 'localhost',
  database: process.env.DB_NAME || 'myapp_db',
  password: process.env.DB_PASSWORD || 'defaultpassword',
  port: process.env.DB_PORT || 5432,
});

module.exports = pool;

NODE_ENV=development
PORT=3000
JWT_SECRET=your-secret-key-here
DB_USER=myapp_user
DB_HOST=localhost
DB_NAME=myapp_db
DB_PASSWORD=wrongpassword123
DB_PORT=5432

CREATE TABLE users (
  id SERIAL PRIMARY KEY,
  email VARCHAR(255) UNIQUE NOT NULL,
  password_hash VARCHAR(255) NOT NULL,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

claude-code --task "Build a complete REST API for user management with authentication, validation, and comprehensive testing"

POST   /api/auth/register         - Register new user
POST   /api/auth/login            - User login
POST   /api/auth/logout           - User logout
POST   /api/auth/refresh          - Refresh JWT token
POST   /api/auth/forgot-password  - Request password reset
POST   /api/auth/reset-password   - Reset password with token

GET    /api/users/profile         - Get current user profile
PUT    /api/users/profile         - Update current user profile
DELETE /api/users/profile         - Delete current user account
PUT    /api/users/change-password - Change user password

GET    /api/admin/users           - List all users (admin only)
GET    /api/admin/users/:id       - Get specific user (admin only)
PUT    /api/admin/users/:id/status - Update user status (admin only)
DELETE /api/admin/users/:id       - Delete user (admin only)

// User entity should include:
{
  id: "UUID primary key",
  email: "unique, validated email address",
  password_hash: "bcrypt hashed password",
  first_name: "required string, 2-50 characters",
  last_name: "required string, 2-50 characters",
  role: "enum: 'user' | 'admin', default 'user'",
  status: "enum: 'active' | 'inactive' | 'suspended', default 'active'",
  email_verified: "boolean, default false",
  last_login: "timestamp, nullable",
  created_at: "timestamp, auto-generated",
  updated_at: "timestamp, auto-updated"
}

// Success response format:
{
  "success": true,
  "data": {...},
  "message": "Operation completed successfully"
}

// Error response format:
{
  "success": false,
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Invalid input provided",
    "details": [
      {
        "field": "email",
        "message": "Email format is invalid"
      }
    ]
  }
}

/src
  /controllers    - Request handling logic
  /services       - Business logic layer
  /models         - Data access layer
  /middleware     - Custom middleware (auth, validation, etc.)
  /routes         - API route definitions
  /utils          - Helper functions
  /config         - Configuration files
  /validators     - Input validation schemas
/tests          - Test files mirroring src structure
/docs           - API documentation
/migrations     - Database schema migrations