This commit is contained in:
@@ -0,0 +1,381 @@
|
||||
import { Router, Request, Response } from 'express';
|
||||
import { asyncHandler } from '../utils/asyncHandler';
|
||||
import { db } from '../db';
|
||||
import { categories, tasks, subtasks, repeatProfiles, users, friendships } from '../db/schema';
|
||||
import { eq, and, gte, lte, asc, or, inArray, sql } from 'drizzle-orm';
|
||||
import { authMiddleware } from '../utils/auth';
|
||||
import { AppError } from '../middleware/errorHandler';
|
||||
import { syncQuerySchema, pushChangesSchema, canCompleteTask } from '../utils/validation';
|
||||
|
||||
const router = Router();
|
||||
|
||||
router.use(authMiddleware);
|
||||
|
||||
router.get('/', asyncHandler(async (req: Request, res: Response) => {
|
||||
const { since } = syncQuerySchema.parse(req.query);
|
||||
const userId = req.user!.userId;
|
||||
|
||||
const sinceDate = since;
|
||||
|
||||
// Fetch categories changed since timestamp
|
||||
const changedCategories = await db
|
||||
.select()
|
||||
.from(categories)
|
||||
.where(and(eq(categories.userId, userId), gte(categories.updatedAt, sinceDate)))
|
||||
.orderBy(asc(categories.updatedAt));
|
||||
|
||||
// Fetch tasks changed since timestamp
|
||||
const changedTasks = await db
|
||||
.select()
|
||||
.from(tasks)
|
||||
.where(and(eq(tasks.userId, userId), gte(tasks.updatedAt, sinceDate)))
|
||||
.orderBy(asc(tasks.updatedAt));
|
||||
|
||||
// Fetch subtasks changed since timestamp
|
||||
const taskIds = changedTasks.map(t => t.id);
|
||||
let changedSubtasks: any[] = [];
|
||||
|
||||
if (taskIds.length > 0) {
|
||||
changedSubtasks = await db
|
||||
.select()
|
||||
.from(subtasks)
|
||||
.where(and(eq(subtasks.userId, userId), gte(subtasks.updatedAt, sinceDate)))
|
||||
.orderBy(asc(subtasks.updatedAt));
|
||||
} else {
|
||||
// Also fetch subtasks for tasks that might have been deleted (we track by updatedAt)
|
||||
changedSubtasks = await db
|
||||
.select()
|
||||
.from(subtasks)
|
||||
.where(and(eq(subtasks.userId, userId), gte(subtasks.updatedAt, sinceDate)))
|
||||
.orderBy(asc(subtasks.updatedAt));
|
||||
}
|
||||
|
||||
// Fetch repeat profiles changed since timestamp
|
||||
const changedRepeatProfiles = await db
|
||||
.select()
|
||||
.from(repeatProfiles)
|
||||
.where(and(eq(repeatProfiles.userId, userId), gte(repeatProfiles.updatedAt, sinceDate)))
|
||||
.orderBy(asc(repeatProfiles.updatedAt));
|
||||
|
||||
// Fetch friendships changed since timestamp
|
||||
const changedFriendships = await db
|
||||
.select()
|
||||
.from(friendships)
|
||||
.where(and(
|
||||
or(eq(friendships.userId, userId), eq(friendships.friendId, userId)),
|
||||
gte(friendships.updatedAt, sinceDate)
|
||||
))
|
||||
.orderBy(asc(friendships.updatedAt));
|
||||
|
||||
const timestamp = Date.now();
|
||||
|
||||
res.json({
|
||||
categories: changedCategories,
|
||||
tasks: changedTasks,
|
||||
subtasks: changedSubtasks,
|
||||
repeatProfiles: changedRepeatProfiles,
|
||||
friendships: changedFriendships,
|
||||
timestamp,
|
||||
});
|
||||
}));
|
||||
|
||||
router.post('/push', asyncHandler(async (req: Request, res: Response) => {
|
||||
const data = pushChangesSchema.parse(req.body);
|
||||
const userId = req.user!.userId;
|
||||
const conflicts: any[] = [];
|
||||
const timestamp = Date.now();
|
||||
|
||||
try {
|
||||
await db.transaction(async (tx) => {
|
||||
// Ensure every referenced category exists (FK integrity) so a stale or
|
||||
// never-synced category reference cannot fail the entire push. Missing
|
||||
// categories are recreated as a fallback and the client heals on pull.
|
||||
const referencedCategoryIds = new Set<string>();
|
||||
for (const task of data.changes.tasks ?? []) {
|
||||
if (task.categoryId) referencedCategoryIds.add(task.categoryId);
|
||||
}
|
||||
for (const sub of data.changes.subtasks ?? []) {
|
||||
const task = (data.changes.tasks ?? []).find((t) => t.id === sub.taskId);
|
||||
if (task?.categoryId) referencedCategoryIds.add(task.categoryId);
|
||||
}
|
||||
if (referencedCategoryIds.size > 0) {
|
||||
const existing = await tx
|
||||
.select({ id: categories.id })
|
||||
.from(categories)
|
||||
.where(inArray(categories.id, [...referencedCategoryIds]));
|
||||
const existingIds = new Set(existing.map((c) => c.id));
|
||||
const missing = [...referencedCategoryIds].filter((id) => !existingIds.has(id));
|
||||
if (missing.length > 0) {
|
||||
const rows = await tx
|
||||
.select({ max: sql<number>`max(${categories.order})` })
|
||||
.from(categories)
|
||||
.where(eq(categories.userId, userId));
|
||||
const startOrder = (rows[0]?.max ?? -1) + 1;
|
||||
await tx.insert(categories).values(
|
||||
missing.map((id, i) => ({
|
||||
id,
|
||||
userId,
|
||||
name: 'Default',
|
||||
color: '#9E9E9E',
|
||||
order: startOrder + i,
|
||||
createdAt: Date.now(),
|
||||
updatedAt: Date.now(),
|
||||
}))
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Process categories
|
||||
if (data.changes.categories && data.changes.categories.length > 0) {
|
||||
for (const cat of data.changes.categories) {
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(categories)
|
||||
.where(and(eq(categories.id, cat.id), eq(categories.userId, userId)))
|
||||
.limit(1);
|
||||
|
||||
if (existing.length > 0) {
|
||||
// Check for conflict
|
||||
if (existing[0].updatedAt > cat.updatedAt) {
|
||||
conflicts.push({
|
||||
entity: 'categories',
|
||||
id: cat.id,
|
||||
serverVersion: existing[0],
|
||||
clientVersion: cat,
|
||||
resolution: 'server_wins',
|
||||
});
|
||||
continue; // Server wins
|
||||
}
|
||||
|
||||
await tx
|
||||
.update(categories)
|
||||
.set({
|
||||
name: cat.name,
|
||||
color: cat.color,
|
||||
order: cat.order,
|
||||
updatedAt: cat.updatedAt,
|
||||
})
|
||||
.where(and(eq(categories.id, cat.id), eq(categories.userId, userId)));
|
||||
} else {
|
||||
await tx.insert(categories).values({
|
||||
...cat,
|
||||
userId,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Process tasks
|
||||
if (data.changes.tasks && data.changes.tasks.length > 0) {
|
||||
for (const task of data.changes.tasks) {
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(tasks)
|
||||
.where(and(eq(tasks.id, task.id), eq(tasks.userId, userId)))
|
||||
.limit(1);
|
||||
|
||||
if (existing.length > 0) {
|
||||
// Check for conflict
|
||||
if (existing[0].updatedAt > task.updatedAt) {
|
||||
conflicts.push({
|
||||
entity: 'tasks',
|
||||
id: task.id,
|
||||
serverVersion: existing[0],
|
||||
clientVersion: task,
|
||||
resolution: 'server_wins',
|
||||
});
|
||||
continue; // Server wins
|
||||
}
|
||||
|
||||
// Prevent completing tasks with future due dates
|
||||
if (task.completed === true) {
|
||||
const effectiveDueDate = task.dueDate ?? existing[0].dueDate;
|
||||
if (!canCompleteTask(effectiveDueDate)) {
|
||||
conflicts.push({
|
||||
entity: 'tasks',
|
||||
id: task.id,
|
||||
serverVersion: existing[0],
|
||||
clientVersion: task,
|
||||
resolution: 'server_wins',
|
||||
});
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
await tx
|
||||
.update(tasks)
|
||||
.set({
|
||||
title: task.title,
|
||||
description: task.description,
|
||||
categoryId: task.categoryId,
|
||||
priority: task.priority,
|
||||
completed: task.completed,
|
||||
dueDate: task.dueDate,
|
||||
dueTime: task.dueTime ?? '',
|
||||
endTime: task.endTime ?? '',
|
||||
allDay: task.allDay ?? false,
|
||||
repeat: task.repeat ?? 'none',
|
||||
repeatInterval: task.repeatInterval ?? 1,
|
||||
repeatDays: task.repeatDays ?? '',
|
||||
seriesId: task.seriesId ?? '',
|
||||
reminder: task.reminder ?? 'none',
|
||||
reminders: task.reminders ?? '',
|
||||
assigneeId: task.assigneeId ?? null,
|
||||
updatedAt: task.updatedAt,
|
||||
})
|
||||
.where(and(eq(tasks.id, task.id), eq(tasks.userId, userId)));
|
||||
} else {
|
||||
await tx.insert(tasks).values({
|
||||
...task,
|
||||
allDay: task.allDay ?? false,
|
||||
userId,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Process subtasks
|
||||
if (data.changes.subtasks && data.changes.subtasks.length > 0) {
|
||||
for (const sub of data.changes.subtasks) {
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(subtasks)
|
||||
.where(and(eq(subtasks.id, sub.id), eq(subtasks.userId, userId)))
|
||||
.limit(1);
|
||||
|
||||
if (existing.length > 0) {
|
||||
// Check for conflict
|
||||
if (existing[0].updatedAt > sub.updatedAt) {
|
||||
conflicts.push({
|
||||
entity: 'subtasks',
|
||||
id: sub.id,
|
||||
serverVersion: existing[0],
|
||||
clientVersion: sub,
|
||||
resolution: 'server_wins',
|
||||
});
|
||||
continue; // Server wins
|
||||
}
|
||||
|
||||
await tx
|
||||
.update(subtasks)
|
||||
.set({
|
||||
taskId: sub.taskId,
|
||||
title: sub.title,
|
||||
description: sub.description ?? '',
|
||||
priority: sub.priority ?? 'none',
|
||||
completed: sub.completed,
|
||||
dueDate: sub.dueDate ?? 0,
|
||||
dueTime: sub.dueTime ?? '',
|
||||
endTime: sub.endTime ?? '',
|
||||
allDay: sub.allDay ?? false,
|
||||
repeat: sub.repeat ?? 'none',
|
||||
repeatInterval: sub.repeatInterval ?? 1,
|
||||
repeatDays: sub.repeatDays ?? '',
|
||||
seriesId: sub.seriesId ?? '',
|
||||
reminder: sub.reminder ?? 'none',
|
||||
reminders: sub.reminders ?? '',
|
||||
assigneeId: sub.assigneeId ?? null,
|
||||
order: sub.order,
|
||||
updatedAt: sub.updatedAt,
|
||||
})
|
||||
.where(and(eq(subtasks.id, sub.id), eq(subtasks.userId, userId)));
|
||||
} else {
|
||||
await tx.insert(subtasks).values({
|
||||
...sub,
|
||||
userId,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Process repeat profiles
|
||||
if (data.changes.repeatProfiles && data.changes.repeatProfiles.length > 0) {
|
||||
for (const profile of data.changes.repeatProfiles) {
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(repeatProfiles)
|
||||
.where(and(eq(repeatProfiles.id, profile.id), eq(repeatProfiles.userId, userId)))
|
||||
.limit(1);
|
||||
|
||||
if (existing.length > 0) {
|
||||
// Check for conflict
|
||||
if (existing[0].updatedAt > profile.updatedAt) {
|
||||
conflicts.push({
|
||||
entity: 'repeatProfiles',
|
||||
id: profile.id,
|
||||
serverVersion: existing[0],
|
||||
clientVersion: profile,
|
||||
resolution: 'server_wins',
|
||||
});
|
||||
continue; // Server wins
|
||||
}
|
||||
|
||||
await tx
|
||||
.update(repeatProfiles)
|
||||
.set({
|
||||
name: profile.name,
|
||||
repeat: profile.repeat,
|
||||
repeatInterval: profile.repeatInterval,
|
||||
repeatDays: profile.repeatDays,
|
||||
updatedAt: profile.updatedAt,
|
||||
})
|
||||
.where(and(eq(repeatProfiles.id, profile.id), eq(repeatProfiles.userId, userId)));
|
||||
} else {
|
||||
await tx.insert(repeatProfiles).values({
|
||||
...profile,
|
||||
userId,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Process friendships
|
||||
if (data.changes.friendships && data.changes.friendships.length > 0) {
|
||||
for (const friendship of data.changes.friendships) {
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(friendships)
|
||||
.where(eq(friendships.id, friendship.id))
|
||||
.limit(1);
|
||||
|
||||
if (existing.length > 0) {
|
||||
// Check for conflict
|
||||
if (existing[0].updatedAt > friendship.updatedAt) {
|
||||
conflicts.push({
|
||||
entity: 'friendships',
|
||||
id: friendship.id,
|
||||
serverVersion: existing[0],
|
||||
clientVersion: friendship,
|
||||
resolution: 'server_wins',
|
||||
});
|
||||
continue; // Server wins
|
||||
}
|
||||
|
||||
await tx
|
||||
.update(friendships)
|
||||
.set({
|
||||
userId: friendship.userId,
|
||||
friendId: friendship.friendId,
|
||||
status: friendship.status,
|
||||
updatedAt: friendship.updatedAt,
|
||||
})
|
||||
.where(eq(friendships.id, friendship.id));
|
||||
} else {
|
||||
await tx.insert(friendships).values(friendship);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('Sync push error:', error);
|
||||
throw new AppError('SERVER_ERROR', 'Failed to process sync push', 500);
|
||||
}
|
||||
|
||||
res.json({
|
||||
success: true,
|
||||
timestamp,
|
||||
conflicts,
|
||||
});
|
||||
}));
|
||||
|
||||
export default router;
|
||||
Reference in New Issue
Block a user