Internal
Public Access
Added comprehensive rate limiting to prevent abuse and DoS attacks: Configuration (config/settings/base.py): - Anon rate: 100 requests/hour for anonymous users - User rate: 1000 requests/hour for authenticated users - Login rate: 10 attempts/hour (prevents brute force) - Register rate: 5 attempts/hour (prevents account spam) - Sync rate: 100 requests/hour (prevents sync flooding) Custom throttle classes (users/throttles.py): - LoginRateThrottle: Strict limit for login endpoint - RegisterRateThrottle: Strict limit for registration endpoint - SyncRateThrottle: Moderate limit for sync endpoint Applied throttling to sensitive endpoints: - users.views.RegisterAPIView: 5/hour - users.views.TokenObtainPairView: 10/hour - sync.views.sync: 100/hour All other API endpoints use default throttles (anon: 100/hour, user: 1000/hour). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
470 lines
16 KiB
Python
470 lines
16 KiB
Python
"""
|
|
Sync views for KeepItGoing.
|
|
|
|
Handles data synchronization between clients and server.
|
|
"""
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime
|
|
from django.utils import timezone
|
|
from django.utils.dateparse import parse_datetime
|
|
from rest_framework import status
|
|
from rest_framework.decorators import api_view, permission_classes, throttle_classes
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.response import Response
|
|
|
|
from tasks.models import Task, Tag, TimeEntry
|
|
from tasks.serializers import (
|
|
TaskSerializer,
|
|
TagSerializer,
|
|
TimeEntrySerializer,
|
|
TaskSyncSerializer,
|
|
TagSyncSerializer,
|
|
TimeEntrySyncSerializer,
|
|
)
|
|
from .models import SyncLog, SyncConflict
|
|
from .serializers import SyncConflictSerializer
|
|
from users.throttles import SyncRateThrottle
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@api_view(['POST'])
|
|
@permission_classes([IsAuthenticated])
|
|
@throttle_classes([SyncRateThrottle])
|
|
def sync(request):
|
|
"""
|
|
Main sync endpoint.
|
|
|
|
Accepts client changes and returns server changes since last sync.
|
|
|
|
Request body:
|
|
{
|
|
"device_id": "unique-device-id",
|
|
"last_sync_token": "previous-sync-token-or-null",
|
|
"changes": {
|
|
"tasks": [...],
|
|
"tags": [...],
|
|
"time_entries": [...]
|
|
}
|
|
}
|
|
|
|
Response:
|
|
{
|
|
"sync_token": "new-sync-token",
|
|
"server_changes": {
|
|
"tasks": [...],
|
|
"tags": [...],
|
|
"time_entries": [...]
|
|
},
|
|
"conflicts": [...]
|
|
}
|
|
"""
|
|
user = request.user
|
|
device_id = request.data.get('device_id')
|
|
last_sync_token = request.data.get('last_sync_token')
|
|
changes = request.data.get('changes', {})
|
|
|
|
# Debug logging
|
|
task_changes = changes.get('tasks', [])
|
|
deleted_task_count = sum(1 for t in task_changes if t.get('is_deleted', False))
|
|
logger.debug(f"Received sync request with {len(task_changes)} tasks, {deleted_task_count} marked as deleted")
|
|
|
|
if not device_id:
|
|
return Response(
|
|
{'error': 'device_id is required'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
# Get last sync time
|
|
last_sync_at = None
|
|
if last_sync_token:
|
|
try:
|
|
sync_log = SyncLog.objects.get(sync_token=last_sync_token, user=user)
|
|
last_sync_at = sync_log.last_sync_at
|
|
except SyncLog.DoesNotExist:
|
|
pass # Full sync
|
|
|
|
# Process incoming changes
|
|
conflicts = []
|
|
conflicts.extend(process_tag_changes(user, changes.get('tags', []), last_sync_at))
|
|
conflicts.extend(process_task_changes(user, changes.get('tasks', []), last_sync_at))
|
|
conflicts.extend(process_time_entry_changes(user, changes.get('time_entries', []), last_sync_at))
|
|
|
|
# Get server changes since last sync
|
|
server_changes = get_server_changes(user, last_sync_at)
|
|
|
|
# Generate new sync token
|
|
new_sync_token = str(uuid.uuid4())
|
|
current_time = timezone.now()
|
|
|
|
# Update or create sync log
|
|
SyncLog.objects.update_or_create(
|
|
user=user,
|
|
device_id=device_id,
|
|
defaults={
|
|
'last_sync_at': current_time,
|
|
'sync_token': new_sync_token,
|
|
}
|
|
)
|
|
|
|
return Response({
|
|
'sync_token': new_sync_token,
|
|
'server_time': current_time.isoformat(),
|
|
'server_changes': server_changes,
|
|
'conflicts': SyncConflictSerializer(conflicts, many=True).data,
|
|
})
|
|
|
|
|
|
def process_task_changes(user, task_changes, last_sync_at):
|
|
"""Process task changes from client."""
|
|
conflicts = []
|
|
|
|
for task_data in task_changes:
|
|
sync_id = task_data.get('sync_id')
|
|
is_deleted = task_data.get('is_deleted', False)
|
|
|
|
# Debug logging
|
|
if is_deleted:
|
|
logger.debug(f"Received deletion for task sync_id: {sync_id}")
|
|
|
|
try:
|
|
existing = Task.all_objects.get(sync_id=sync_id, user=user)
|
|
|
|
if is_deleted:
|
|
# Soft delete the task
|
|
logger.debug(f"Soft deleting task: {existing.title} (id: {existing.id})")
|
|
existing.is_deleted = True
|
|
existing.save()
|
|
logger.debug(f"Task soft deleted successfully")
|
|
continue
|
|
|
|
# Check for conflict
|
|
if last_sync_at and existing.updated_at > last_sync_at:
|
|
# Server version was modified since last sync
|
|
conflict = SyncConflict.objects.create(
|
|
user=user,
|
|
entity_type='task',
|
|
entity_id=existing.id,
|
|
local_data=task_data,
|
|
server_data=TaskSerializer(existing).data,
|
|
)
|
|
conflicts.append(conflict)
|
|
else:
|
|
# Safe to update
|
|
update_task_from_data(existing, task_data)
|
|
|
|
except Task.DoesNotExist:
|
|
if not is_deleted:
|
|
# Create new task
|
|
create_task_from_data(user, task_data)
|
|
|
|
return conflicts
|
|
|
|
|
|
def process_tag_changes(user, tag_changes, last_sync_at):
|
|
"""Process tag changes from client."""
|
|
conflicts = []
|
|
|
|
for tag_data in tag_changes:
|
|
sync_id = tag_data.get('sync_id')
|
|
is_deleted = tag_data.get('is_deleted', False)
|
|
|
|
try:
|
|
existing = Tag.all_objects.get(sync_id=sync_id, user=user)
|
|
|
|
if is_deleted:
|
|
existing.is_deleted = True
|
|
existing.save()
|
|
continue
|
|
|
|
if last_sync_at and existing.updated_at > last_sync_at:
|
|
conflict = SyncConflict.objects.create(
|
|
user=user,
|
|
entity_type='tag',
|
|
entity_id=existing.id,
|
|
local_data=tag_data,
|
|
server_data=TagSerializer(existing).data,
|
|
)
|
|
conflicts.append(conflict)
|
|
else:
|
|
existing.name = tag_data.get('name', existing.name)
|
|
existing.description = tag_data.get('description', existing.description)
|
|
existing.color = tag_data.get('color', existing.color)
|
|
existing.icon = tag_data.get('icon', existing.icon)
|
|
existing.sort_order = tag_data.get('sort_order', existing.sort_order)
|
|
existing.is_archived = tag_data.get('is_archived', existing.is_archived)
|
|
existing.save()
|
|
|
|
except Tag.DoesNotExist:
|
|
if not is_deleted:
|
|
Tag.objects.create(
|
|
user=user,
|
|
sync_id=sync_id,
|
|
name=tag_data.get('name'),
|
|
description=tag_data.get('description', ''),
|
|
color=tag_data.get('color', '#3B82F6'),
|
|
icon=tag_data.get('icon', ''),
|
|
sort_order=tag_data.get('sort_order', 0),
|
|
is_archived=tag_data.get('is_archived', False),
|
|
)
|
|
|
|
return conflicts
|
|
|
|
|
|
def process_time_entry_changes(user, entry_changes, last_sync_at):
|
|
"""Process time entry changes from client."""
|
|
conflicts = []
|
|
|
|
logger.debug(f"Processing {len(entry_changes)} time entry changes")
|
|
|
|
for entry_data in entry_changes:
|
|
sync_id = entry_data.get('sync_id')
|
|
is_deleted = entry_data.get('is_deleted', False)
|
|
logger.debug(f"Time entry sync_id: {sync_id}, is_deleted: {is_deleted}")
|
|
|
|
try:
|
|
existing = TimeEntry.all_objects.get(sync_id=sync_id, user=user)
|
|
|
|
if is_deleted:
|
|
existing.is_deleted = True
|
|
existing.save()
|
|
continue
|
|
|
|
if last_sync_at and existing.updated_at > last_sync_at:
|
|
conflict = SyncConflict.objects.create(
|
|
user=user,
|
|
entity_type='time_entry',
|
|
entity_id=existing.id,
|
|
local_data=entry_data,
|
|
server_data=TimeEntrySerializer(existing).data,
|
|
)
|
|
conflicts.append(conflict)
|
|
else:
|
|
# Parse datetime strings if needed
|
|
started_at = entry_data.get('started_at', existing.started_at)
|
|
if isinstance(started_at, str):
|
|
started_at = parse_datetime(started_at)
|
|
|
|
ended_at = entry_data.get('ended_at', existing.ended_at)
|
|
if ended_at and isinstance(ended_at, str):
|
|
ended_at = parse_datetime(ended_at)
|
|
|
|
existing.started_at = started_at
|
|
existing.ended_at = ended_at
|
|
existing.notes = entry_data.get('notes', existing.notes)
|
|
existing.save()
|
|
|
|
except TimeEntry.DoesNotExist:
|
|
if not is_deleted:
|
|
task_sync_id = entry_data.get('task_sync_id')
|
|
logger.debug(f"Creating new time entry for task_sync_id: {task_sync_id}")
|
|
logger.debug(f" started_at: {entry_data.get('started_at')}, ended_at: {entry_data.get('ended_at')}")
|
|
try:
|
|
task = Task.objects.get(sync_id=task_sync_id, user=user)
|
|
|
|
# Parse datetime strings to datetime objects
|
|
started_at = entry_data.get('started_at')
|
|
if isinstance(started_at, str):
|
|
started_at = parse_datetime(started_at)
|
|
|
|
ended_at = entry_data.get('ended_at')
|
|
if ended_at and isinstance(ended_at, str):
|
|
ended_at = parse_datetime(ended_at)
|
|
|
|
new_entry = TimeEntry.objects.create(
|
|
user=user,
|
|
task=task,
|
|
sync_id=sync_id,
|
|
started_at=started_at,
|
|
ended_at=ended_at,
|
|
notes=entry_data.get('notes', ''),
|
|
)
|
|
logger.debug(f"Created time entry: {new_entry.id}, duration_seconds: {new_entry.duration_seconds}")
|
|
except Task.DoesNotExist:
|
|
logger.debug(f"Task not found for sync_id: {task_sync_id}")
|
|
pass # Skip if task doesn't exist
|
|
|
|
return conflicts
|
|
|
|
|
|
def update_task_from_data(task, data):
|
|
"""Update a task from sync data."""
|
|
task.title = data.get('title', task.title)
|
|
task.description = data.get('description', task.description)
|
|
task.status = data.get('status', task.status)
|
|
task.priority = data.get('priority', task.priority)
|
|
task.due_date = data.get('due_date', task.due_date)
|
|
task.due_time = data.get('due_time', task.due_time)
|
|
task.reminder_at = data.get('reminder_at', task.reminder_at)
|
|
task.recurrence = data.get('recurrence', task.recurrence)
|
|
task.recurrence_rule = data.get('recurrence_rule', task.recurrence_rule)
|
|
task.sort_order = data.get('sort_order', task.sort_order)
|
|
|
|
task.save()
|
|
|
|
# Handle tags
|
|
tag_sync_ids = data.get('tag_sync_ids', [])
|
|
if tag_sync_ids is not None:
|
|
tags = Tag.objects.filter(sync_id__in=tag_sync_ids, user=task.user)
|
|
task.tags.set(tags)
|
|
|
|
|
|
def create_task_from_data(user, data):
|
|
"""Create a new task from sync data."""
|
|
parent = None
|
|
parent_sync_id = data.get('parent_sync_id')
|
|
if parent_sync_id:
|
|
try:
|
|
parent = Task.all_objects.get(sync_id=parent_sync_id, user=user)
|
|
except Task.DoesNotExist:
|
|
pass
|
|
|
|
# Get sync_id and convert to UUID if provided as string
|
|
sync_id = data.get('sync_id')
|
|
if sync_id and isinstance(sync_id, str):
|
|
sync_id = uuid.UUID(sync_id)
|
|
|
|
task = Task.objects.create(
|
|
user=user,
|
|
sync_id=sync_id,
|
|
parent=parent,
|
|
title=data.get('title'),
|
|
description=data.get('description', ''),
|
|
status=data.get('status', 'pending'),
|
|
priority=data.get('priority', 'medium'),
|
|
due_date=data.get('due_date'),
|
|
due_time=data.get('due_time'),
|
|
reminder_at=data.get('reminder_at'),
|
|
recurrence=data.get('recurrence', 'none'),
|
|
recurrence_rule=data.get('recurrence_rule', ''),
|
|
sort_order=data.get('sort_order', 0),
|
|
)
|
|
|
|
# Handle tags
|
|
tag_sync_ids = data.get('tag_sync_ids', [])
|
|
if tag_sync_ids:
|
|
tags = Tag.objects.filter(sync_id__in=tag_sync_ids, user=user)
|
|
task.tags.set(tags)
|
|
|
|
return task
|
|
|
|
|
|
def get_server_changes(user, last_sync_at):
|
|
"""Get all changes on server since last sync."""
|
|
logger.debug(f"get_server_changes: last_sync_at={last_sync_at}")
|
|
filter_kwargs = {'user': user}
|
|
|
|
# For incremental sync, filter by updated_at
|
|
# IMPORTANT: Use all_objects to include deleted items so clients can delete them too
|
|
if last_sync_at:
|
|
filter_kwargs['updated_at__gt'] = last_sync_at
|
|
tasks = Task.all_objects.filter(**filter_kwargs).prefetch_related('tags')
|
|
tags = Tag.all_objects.filter(**filter_kwargs)
|
|
time_entries = TimeEntry.all_objects.filter(**filter_kwargs)
|
|
else:
|
|
# Full sync - send everything (including deleted items for cleanup)
|
|
tasks = Task.all_objects.filter(user=user).prefetch_related('tags')
|
|
tags = Tag.all_objects.filter(user=user)
|
|
time_entries = TimeEntry.all_objects.filter(user=user)
|
|
|
|
logger.debug(f"Returning {len(tasks)} tasks, {len(tags)} tags, {len(time_entries)} time entries")
|
|
if time_entries:
|
|
logger.debug(f"Time entries being sent to client:")
|
|
for entry in time_entries:
|
|
logger.debug(f" sync_id: {entry.sync_id}, task: {entry.task.title}, duration: {entry.duration_seconds}s, updated_at: {entry.updated_at}")
|
|
|
|
return {
|
|
'tasks': TaskSyncSerializer(tasks, many=True).data,
|
|
'tags': TagSyncSerializer(tags, many=True).data,
|
|
'time_entries': TimeEntrySyncSerializer(time_entries, many=True).data,
|
|
}
|
|
|
|
|
|
@api_view(['GET'])
|
|
@permission_classes([IsAuthenticated])
|
|
def get_conflicts(request):
|
|
"""Get pending sync conflicts for user."""
|
|
conflicts = SyncConflict.objects.filter(
|
|
user=request.user,
|
|
status='pending'
|
|
)
|
|
return Response(SyncConflictSerializer(conflicts, many=True).data)
|
|
|
|
|
|
@api_view(['POST'])
|
|
@permission_classes([IsAuthenticated])
|
|
def resolve_conflict(request, conflict_id):
|
|
"""
|
|
Resolve a sync conflict.
|
|
|
|
Request body:
|
|
{
|
|
"resolution": "local" | "server" | "merged",
|
|
"merged_data": {...} // Only if resolution is "merged"
|
|
}
|
|
"""
|
|
try:
|
|
conflict = SyncConflict.objects.get(
|
|
id=conflict_id,
|
|
user=request.user,
|
|
status='pending'
|
|
)
|
|
except SyncConflict.DoesNotExist:
|
|
return Response(
|
|
{'error': 'Conflict not found'},
|
|
status=status.HTTP_404_NOT_FOUND
|
|
)
|
|
|
|
resolution = request.data.get('resolution')
|
|
|
|
if resolution == 'local':
|
|
# Apply local changes
|
|
apply_conflict_data(conflict.entity_type, conflict.entity_id, conflict.local_data)
|
|
conflict.status = 'resolved_local'
|
|
elif resolution == 'server':
|
|
# Keep server version (no action needed)
|
|
conflict.status = 'resolved_server'
|
|
elif resolution == 'merged':
|
|
merged_data = request.data.get('merged_data')
|
|
if not merged_data:
|
|
return Response(
|
|
{'error': 'merged_data is required for merged resolution'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
apply_conflict_data(conflict.entity_type, conflict.entity_id, merged_data)
|
|
conflict.status = 'resolved_merged'
|
|
else:
|
|
return Response(
|
|
{'error': 'Invalid resolution type'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
conflict.resolved_at = timezone.now()
|
|
conflict.save()
|
|
|
|
return Response({'status': 'resolved'})
|
|
|
|
|
|
def apply_conflict_data(entity_type, entity_id, data):
|
|
"""Apply conflict resolution data to entity."""
|
|
if entity_type == 'task':
|
|
try:
|
|
task = Task.objects.get(id=entity_id)
|
|
update_task_from_data(task, data)
|
|
except Task.DoesNotExist:
|
|
pass
|
|
elif entity_type == 'tag':
|
|
try:
|
|
tag = Tag.objects.get(id=entity_id)
|
|
tag.name = data.get('name', tag.name)
|
|
tag.description = data.get('description', tag.description)
|
|
tag.color = data.get('color', tag.color)
|
|
tag.icon = data.get('icon', tag.icon)
|
|
tag.sort_order = data.get('sort_order', tag.sort_order)
|
|
tag.is_archived = data.get('is_archived', tag.is_archived)
|
|
tag.save()
|
|
except Tag.DoesNotExist:
|
|
pass
|