Files
carry-your-live/carry-your-live/src/database/sync.ts
T

698 lines
23 KiB
TypeScript

import { AppState, AppStateStatus } from 'react-native';
import { database, collections } from './index';
import AsyncStorage from '@react-native-async-storage/async-storage';
import { Q } from '@nozbe/watermelondb';
import { apiFetch, getAuthToken } from '@/services/auth';
import { cancelTaskReminder } from '@/services/notifications';
import Category from '@/models/Category';
import Task from '@/models/Task';
import Subtask from '@/models/Subtask';
import RepeatProfile from '@/models/RepeatProfile';
import Friendship from '@/models/Friendship';
import { TombstoneEntity } from '@/models/Tombstone';
import { fetchPendingTombstones, removeTombstonesInBatch } from './tombstones';
const LAST_PULLED_AT_KEY = 'sync:lastPulledAt';
const LAST_RUN_AT_KEY = 'sync:lastRunAt';
export interface SyncResult {
pushed: { categories: number; tasks: number; subtasks: number; repeatProfiles: number; friendships: number };
pulled: { categories: number; tasks: number; subtasks: number; repeatProfiles: number; friendships: number };
conflicts: number;
timestamp: number;
}
interface ServerRow {
id: string;
createdAt: number;
updatedAt: number;
[key: string]: any;
}
interface PushConflict {
entity: 'categories' | 'tasks' | 'subtasks' | 'repeatProfiles' | 'friendships';
id: string;
serverVersion: ServerRow;
resolution: string;
}
async function getLastPulledAt(): Promise<number> {
try {
const raw = await AsyncStorage.getItem(LAST_PULLED_AT_KEY);
const value = raw ? Number(raw) : 0;
return Number.isFinite(value) ? value : 0;
} catch {
return 0;
}
}
async function setLastPulledAt(timestamp: number): Promise<void> {
try {
await AsyncStorage.setItem(LAST_PULLED_AT_KEY, String(timestamp));
await AsyncStorage.setItem(LAST_RUN_AT_KEY, String(Date.now()));
} catch {
// ignore
}
}
export async function getLastSyncTime(): Promise<number | null> {
try {
const raw = await AsyncStorage.getItem(LAST_RUN_AT_KEY);
const value = raw ? Number(raw) : null;
return value && Number.isFinite(value) ? value : null;
} catch {
return null;
}
}
async function findOrNull<T>(table: any, id: string): Promise<T | null> {
return table.find(id).catch(() => null);
}
export async function runSync(): Promise<SyncResult> {
const token = await getAuthToken();
if (!token) {
throw new Error('NOT_SIGNED_IN');
}
const result: SyncResult = {
pushed: { categories: 0, tasks: 0, subtasks: 0, repeatProfiles: 0, friendships: 0 },
pulled: { categories: 0, tasks: 0, subtasks: 0, repeatProfiles: 0, friendships: 0 },
conflicts: 0,
timestamp: 0,
};
const { conflicts, pushedDeletions } = await pushChanges();
result.conflicts = conflicts.length;
await applyConflicts(conflicts);
await pruneSyncedTombstones(conflicts, pushedDeletions);
const lastPulledAt = await getLastPulledAt();
const response = await apiFetch(`/sync?since=${lastPulledAt}`);
const data = await response.json();
const pulledCounts = await applyServerChanges(data);
result.pulled = pulledCounts;
result.timestamp = data.timestamp;
await setLastPulledAt(data.timestamp);
return result;
}
async function pushChanges(): Promise<{ conflicts: PushConflict[]; pushedDeletions: { entity: TombstoneEntity; id: string }[] }> {
const lastPulledAt = await getLastPulledAt();
const [tasks, subtasks, repeatProfiles, friendships] = await Promise.all([
collections.tasks.query(Q.where('updated_at', Q.gt(lastPulledAt))).fetch(),
collections.subtasks.query(Q.where('updated_at', Q.gt(lastPulledAt))).fetch(),
collections.repeatProfiles.query(Q.where('updated_at', Q.gt(lastPulledAt))).fetch(),
collections.friendships.query(Q.where('updated_at', Q.gt(lastPulledAt))).fetch(),
]);
const taskPayload = (t: Task) => ({
id: t.id,
title: t.title,
description: t.description,
categoryId: t.categoryId,
tags: t.tags || '',
priority: t.priority,
completed: t.completed,
dueDate: t.dueDate,
dueTime: t.dueTime,
endTime: t.endTime || '',
allDay: t.allDay ?? false,
repeat: t.repeat || 'none',
repeatInterval: t.repeatInterval || 1,
repeatDays: t.repeatDays || '',
seriesId: t.seriesId || '',
reminder: t.reminder || 'none',
reminders: t.reminders || '',
assigneeId: t.assigneeId || null,
completedAt: t.completedAt ?? null,
createdAt: t.createdAt.getTime(),
updatedAt: t.updatedAt.getTime(),
});
const changedTasks: ReturnType<typeof taskPayload>[] = [];
const includedTaskIds = new Set<string>();
const missingTaskIds = new Set<string>();
const referencedCategoryIds = new Set<string>();
for (const t of tasks) {
changedTasks.push(taskPayload(t));
includedTaskIds.add(t.id);
if (t.categoryId) referencedCategoryIds.add(t.categoryId);
}
const changedSubtasks = subtasks.map((s) => ({
id: s.id,
taskId: s.taskId,
parentSubtaskId: s.parentSubtaskId || null,
title: s.title,
description: s.description ?? '',
priority: s.priority ?? 'none',
completed: s.completed,
dueDate: s.dueDate ?? 0,
dueTime: s.dueTime ?? '',
endTime: s.endTime ?? '',
allDay: s.allDay ?? false,
repeat: s.repeat ?? 'none',
repeatInterval: s.repeatInterval ?? 1,
repeatDays: s.repeatDays ?? '',
seriesId: s.seriesId ?? '',
reminder: s.reminder ?? 'none',
reminders: s.reminders ?? '',
assigneeId: s.assigneeId ?? null,
order: s.order,
createdAt: s.createdAt.getTime(),
updatedAt: s.updatedAt.getTime(),
}));
for (const sub of changedSubtasks) {
if (!includedTaskIds.has(sub.taskId)) {
missingTaskIds.add(sub.taskId);
}
}
if (missingTaskIds.size > 0) {
const parentTasks = await collections.tasks
.query(Q.where('id', Q.oneOf(Array.from(missingTaskIds))))
.fetch();
for (const t of parentTasks) {
includedTaskIds.add(t.id);
changedTasks.push(taskPayload(t));
if (t.categoryId) referencedCategoryIds.add(t.categoryId);
}
}
let categoryQuery = collections.categories.query();
if (referencedCategoryIds.size > 0) {
categoryQuery = collections.categories.query(
Q.or(
Q.where('updated_at', Q.gt(lastPulledAt)),
Q.where('id', Q.oneOf(Array.from(referencedCategoryIds)))
)
);
} else {
categoryQuery = collections.categories.query(Q.where('updated_at', Q.gt(lastPulledAt)));
}
const categories = await categoryQuery.fetch();
const changedCategories = categories.map((c) => ({
id: c.id,
name: c.name,
color: c.color,
order: c.order,
createdAt: c.createdAt.getTime(),
updatedAt: c.updatedAt.getTime(),
}));
const changedRepeatProfiles = repeatProfiles.map((p) => ({
id: p.id,
name: p.name,
repeat: p.repeat,
repeatInterval: p.repeatInterval,
repeatDays: p.repeatDays,
createdAt: p.createdAt.getTime(),
updatedAt: p.updatedAt.getTime(),
}));
const changedFriendships = friendships.map((f) => ({
id: f.id,
userId: f.userId,
friendId: f.friendId,
status: f.status,
createdAt: f.createdAt.getTime(),
updatedAt: f.updatedAt.getTime(),
}));
const pendingTombstones = await fetchPendingTombstones(lastPulledAt);
const changedDeletions = pendingTombstones.map((t) => ({
entity: t.entity,
id: t.entityId,
updatedAt: t.deletedAt.getTime(),
}));
if (
changedCategories.length === 0 &&
changedTasks.length === 0 &&
changedSubtasks.length === 0 &&
changedRepeatProfiles.length === 0 &&
changedFriendships.length === 0 &&
changedDeletions.length === 0
) {
return { conflicts: [], pushedDeletions: [] };
}
const response = await apiFetch('/sync/push', {
method: 'POST',
body: JSON.stringify({
changes: {
categories: changedCategories,
tasks: changedTasks,
subtasks: changedSubtasks,
repeatProfiles: changedRepeatProfiles,
friendships: changedFriendships,
},
deleted: changedDeletions,
lastPulledAt,
}),
});
const data = await response.json();
return {
conflicts: (data.conflicts ?? []) as PushConflict[],
pushedDeletions: pendingTombstones.map((t) => ({ entity: t.entity, id: t.entityId })),
};
}
// Remove local tombstones that were just accepted by the server. Tombstones
// whose deletion lost to a newer server row (conflict) are kept so the row
// stays marked as deleted locally after the server version wins.
async function pruneSyncedTombstones(
conflicts: PushConflict[],
pushedDeletions: { entity: TombstoneEntity; id: string }[]
): Promise<void> {
if (pushedDeletions.length === 0) return;
const conflictKeys = new Set(conflicts.map((c) => `${c.entity}:${c.id}`));
const accepted = pushedDeletions.filter((d) => !conflictKeys.has(`${d.entity}:${d.id}`));
const grouped = new Map<TombstoneEntity, string[]>();
for (const t of accepted) {
const list = grouped.get(t.entity) ?? [];
list.push(t.id);
grouped.set(t.entity, list);
}
if (accepted.length === 0) return;
await database.write(async () => {
for (const [entity, ids] of grouped) {
await removeTombstonesInBatch(entity, ids);
}
});
}
async function applyConflicts(conflicts: PushConflict[]): Promise<void> {
if (conflicts.length === 0) return;
await database.write(async () => {
for (const conflict of conflicts) {
const row = conflict.serverVersion;
switch (conflict.entity) {
case 'categories':
await upsertCategory(row, true);
break;
case 'tasks':
await upsertTask(row, true);
break;
case 'subtasks':
await upsertSubtask(row, true);
break;
case 'repeatProfiles':
await upsertRepeatProfile(row, true);
break;
case 'friendships':
await upsertFriendship(row, true);
break;
}
// The server version of this row is newer - drop any pending local
// tombstone so the resurrected row is kept and synced going forward.
await removeTombstonesInBatch(conflict.entity, [conflict.id]);
}
});
}
async function applyServerChanges(data: {
categories: ServerRow[];
tasks: ServerRow[];
subtasks: ServerRow[];
repeatProfiles: ServerRow[];
friendships: ServerRow[];
deleted?: { entity: TombstoneEntity; id: string; updatedAt: number }[];
}): Promise<SyncResult['pulled']> {
const counts = { categories: 0, tasks: 0, subtasks: 0, repeatProfiles: 0, friendships: 0 };
await database.write(async () => {
// Apply deletions first so rows that were removed server-side are gone
// before any upsert re-imports their data.
for (const tomb of data.deleted ?? []) {
await applyRemoteTombstone(tomb);
}
for (const cat of data.categories) {
if (await upsertCategory(cat, false)) counts.categories++;
}
for (const task of data.tasks) {
if (await upsertTask(task, false)) counts.tasks++;
}
for (const sub of data.subtasks) {
if (await upsertSubtask(sub, false)) counts.subtasks++;
}
for (const profile of data.repeatProfiles) {
if (await upsertRepeatProfile(profile, false)) counts.repeatProfiles++;
}
for (const friendship of data.friendships) {
if (await upsertFriendship(friendship, false)) counts.friendships++;
}
});
return counts;
}
// Apply a server-side deletion to the local DB, cascading to every related
// row (category -> tasks -> subtasks, task -> subtasks, subtask -> children).
// LWW: if the local row is newer than the deletion, it survives.
async function applyRemoteTombstone(tomb: { entity: TombstoneEntity; id: string; updatedAt: number }): Promise<void> {
const { entity, id, updatedAt } = tomb;
const row = await findOrNull<any>(collections[entity as keyof typeof collections] as any, id);
if (row) {
if (row.updatedAt.getTime() > updatedAt) {
return; // local edit is newer than the remote deletion - keep it
}
await destroyLocalEntityCascade(entity, id);
} else {
// Row already gone locally - clear the tombstone if the server has it.
await removeTombstonesInBatch(entity, [id]);
}
}
async function destroyLocalEntityCascade(entity: TombstoneEntity, id: string): Promise<void> {
if (entity === 'tasks') {
const subtasks = await collections.subtasks.query(Q.where('task_id', id)).fetch();
for (const s of subtasks) {
await destroyLocalEntityCascade('subtasks', s.id);
}
const task = await findOrNull<any>(collections.tasks, id).catch(() => null);
if (task) {
await task.destroyPermanently();
await cancelTaskReminder(id).catch(() => {});
}
return;
}
if (entity === 'categories') {
const tasks = await collections.tasks.query(Q.where('category_id', id)).fetch();
for (const t of tasks) {
await destroyLocalEntityCascade('tasks', t.id);
}
const cat = await findOrNull<any>(collections.categories, id).catch(() => null);
if (cat) await cat.destroyPermanently();
return;
}
if (entity === 'subtasks') {
const children = await collections.subtasks.query(Q.where('parent_subtask_id', id)).fetch();
for (const c of children) {
await destroyLocalEntityCascade('subtasks', c.id);
}
const sub = await findOrNull<any>(collections.subtasks, id).catch(() => null);
if (sub) {
await sub.destroyPermanently();
await cancelTaskReminder(id).catch(() => {});
}
return;
}
if (entity === 'repeatProfiles') {
const p = await findOrNull<any>(collections.repeatProfiles, id).catch(() => null);
if (p) await p.destroyPermanently();
return;
}
if (entity === 'friendships') {
const f = await findOrNull<any>(collections.friendships, id).catch(() => null);
if (f) await f.destroyPermanently();
}
}
async function upsertCategory(row: ServerRow, force: boolean): Promise<boolean> {
const local = await findOrNull<Category>(collections.categories, row.id);
if (!local) {
await collections.categories.create((c) => {
c.name = String(row.name ?? '');
c.color = String(row.color ?? '#9E9E9E');
c.order = Number(row.order ?? 0);
c.createdAt = new Date(row.createdAt ?? Date.now());
c.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
if (!force && local.updatedAt.getTime() > Number(row.updatedAt ?? 0)) {
return false;
}
await local.update((c) => {
c.name = String(row.name ?? c.name);
c.color = String(row.color ?? c.color);
c.order = Number(row.order ?? c.order);
c.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
async function upsertTask(row: ServerRow, force: boolean): Promise<boolean> {
const local = await findOrNull<Task>(collections.tasks, row.id);
if (!local) {
await collections.tasks.create((t) => {
t.title = String(row.title ?? '');
t.description = String(row.description ?? '');
t.categoryId = String(row.categoryId ?? '');
t.tags = String(row.tags ?? '');
t.priority = row.priority ?? 'none';
t.completed = Boolean(row.completed);
t.dueDate = Number(row.dueDate ?? 0);
t.dueTime = String(row.dueTime ?? '');
t.endTime = String(row.endTime ?? '');
t.repeat = row.repeat ?? 'none';
t.repeatInterval = Number(row.repeatInterval ?? 1);
t.repeatDays = String(row.repeatDays ?? '');
t.seriesId = String(row.seriesId ?? '');
t.reminder = row.reminder ?? 'none';
t.reminders = String(row.reminders ?? '');
t.assigneeId = row.assigneeId ?? null;
t.completedAt = row.completedAt == null ? null : Number(row.completedAt);
t.createdAt = new Date(row.createdAt ?? Date.now());
t.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
if (!force && local.updatedAt.getTime() > Number(row.updatedAt ?? 0)) {
return false;
}
await local.update((t) => {
t.title = String(row.title ?? t.title);
t.description = String(row.description ?? t.description);
t.categoryId = String(row.categoryId ?? t.categoryId);
t.tags = String(row.tags ?? t.tags ?? '');
t.priority = row.priority ?? t.priority;
t.completed = Boolean(row.completed ?? t.completed);
t.dueDate = Number(row.dueDate ?? t.dueDate);
t.dueTime = String(row.dueTime ?? t.dueTime);
t.endTime = String(row.endTime ?? t.endTime);
t.repeat = row.repeat ?? t.repeat;
t.repeatInterval = Number(row.repeatInterval ?? t.repeatInterval);
t.repeatDays = String(row.repeatDays ?? t.repeatDays);
t.seriesId = String(row.seriesId ?? t.seriesId);
t.reminder = row.reminder ?? t.reminder;
t.reminders = String(row.reminders ?? t.reminders ?? '');
t.assigneeId = row.assigneeId ?? t.assigneeId;
t.completedAt = row.completedAt == null ? null : Number(row.completedAt ?? t.completedAt);
t.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
async function upsertRepeatProfile(row: ServerRow, force: boolean): Promise<boolean> {
const local = await findOrNull<RepeatProfile>(collections.repeatProfiles, row.id);
if (!local) {
await collections.repeatProfiles.create((p) => {
p.name = String(row.name ?? '');
p.repeat = row.repeat ?? 'none';
p.repeatInterval = Number(row.repeatInterval ?? 1);
p.repeatDays = String(row.repeatDays ?? '');
p.createdAt = new Date(row.createdAt ?? Date.now());
p.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
if (!force && local.updatedAt.getTime() > Number(row.updatedAt ?? 0)) {
return false;
}
await local.update((p) => {
p.name = String(row.name ?? p.name);
p.repeat = row.repeat ?? p.repeat;
p.repeatInterval = Number(row.repeatInterval ?? p.repeatInterval);
p.repeatDays = String(row.repeatDays ?? p.repeatDays);
p.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
async function upsertSubtask(row: ServerRow, force: boolean): Promise<boolean> {
const task = await findOrNull<Task>(collections.tasks, String(row.taskId ?? ''));
if (!task) return false;
const local = await findOrNull<Subtask>(collections.subtasks, row.id);
if (!local) {
await collections.subtasks.create((s) => {
s.taskId = String(row.taskId ?? '');
s.parentSubtaskId = row.parentSubtaskId == null ? null : String(row.parentSubtaskId);
s.title = String(row.title ?? '');
s.description = String(row.description ?? '');
s.priority = row.priority ?? 'none';
s.completed = Boolean(row.completed);
s.dueDate = Number(row.dueDate ?? 0);
s.dueTime = String(row.dueTime ?? '');
s.endTime = String(row.endTime ?? '');
s.allDay = Boolean(row.allDay ?? false);
s.repeat = row.repeat ?? 'none';
s.repeatInterval = Number(row.repeatInterval ?? 1);
s.repeatDays = String(row.repeatDays ?? '');
s.seriesId = String(row.seriesId ?? '');
s.reminder = row.reminder ?? 'none';
s.assigneeId = row.assigneeId ?? null;
s.order = Number(row.order ?? 0);
s.createdAt = new Date(row.createdAt ?? Date.now());
s.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
if (!force && local.updatedAt.getTime() > Number(row.updatedAt ?? 0)) {
return false;
}
await local.update((s) => {
s.taskId = String(row.taskId ?? s.taskId);
s.parentSubtaskId = row.parentSubtaskId == null ? null : String(row.parentSubtaskId);
s.title = String(row.title ?? s.title);
s.description = String(row.description ?? s.description);
s.priority = row.priority ?? s.priority;
s.completed = Boolean(row.completed ?? s.completed);
s.dueDate = Number(row.dueDate ?? s.dueDate);
s.dueTime = String(row.dueTime ?? s.dueTime);
s.endTime = String(row.endTime ?? s.endTime);
s.allDay = Boolean(row.allDay ?? s.allDay);
s.repeat = row.repeat ?? s.repeat;
s.repeatInterval = Number(row.repeatInterval ?? s.repeatInterval);
s.repeatDays = String(row.repeatDays ?? s.repeatDays);
s.seriesId = String(row.seriesId ?? s.seriesId);
s.reminder = row.reminder ?? s.reminder;
s.assigneeId = row.assigneeId ?? s.assigneeId;
s.order = Number(row.order ?? s.order);
s.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
async function upsertFriendship(row: ServerRow, force: boolean): Promise<boolean> {
const local = await findOrNull<Friendship>(collections.friendships, row.id);
if (!local) {
await collections.friendships.create((f) => {
f.userId = String(row.userId ?? '');
f.friendId = String(row.friendId ?? '');
f.status = (row.status as 'pending' | 'accepted') ?? 'pending';
f.createdAt = new Date(row.createdAt ?? Date.now());
f.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
if (!force && local.updatedAt.getTime() > Number(row.updatedAt ?? 0)) {
return false;
}
await local.update((f) => {
f.userId = String(row.userId ?? f.userId);
f.friendId = String(row.friendId ?? f.friendId);
f.status = (row.status as 'pending' | 'accepted') ?? f.status;
f.updatedAt = new Date(row.updatedAt ?? Date.now());
});
return true;
}
export const DEFAULT_WATCH_INTERVAL_MS = 60_000;
export interface UpdateWatcher {
stop: () => void;
isRunning: () => boolean;
}
let watcherStop: (() => void) | null = null;
let watcherActive = false;
let appInForeground = true;
let syncInFlight = false;
export function watchForUpdates(intervalMs: number = DEFAULT_WATCH_INTERVAL_MS): UpdateWatcher {
if (watcherStop) {
return {
stop: () => stopWatchingForUpdates(),
isRunning: () => watcherActive,
};
}
appInForeground = AppState.currentState === 'active';
const appStateSubscription = AppState.addEventListener('change', (state: AppStateStatus) => {
const next = state === 'active';
const wentForeground = next && !appInForeground;
appInForeground = next;
if (wentForeground) {
maybeSyncOnce().catch(() => {});
}
});
const timer = setInterval(() => {
if (appInForeground && !syncInFlight) {
maybeSyncOnce().catch(() => {});
}
}, intervalMs);
watcherActive = true;
watcherStop = () => {
clearInterval(timer);
appStateSubscription.remove();
watcherStop = null;
watcherActive = false;
};
return {
stop: () => stopWatchingForUpdates(),
isRunning: () => watcherActive,
};
}
async function maybeSyncOnce(): Promise<void> {
if (syncInFlight) return;
const token = await getAuthToken();
if (!token) return;
syncInFlight = true;
try {
await runSync();
} finally {
syncInFlight = false;
}
}
export function stopWatchingForUpdates(): void {
watcherStop?.();
}
export function startAutoSync(intervalMs: number = DEFAULT_WATCH_INTERVAL_MS): UpdateWatcher {
return watchForUpdates(intervalMs);
}
export function stopAutoSync(): void {
stopWatchingForUpdates();
}
export async function runSyncGuarded(): Promise<SyncResult> {
if (syncInFlight) {
throw new Error('SYNC_IN_FLIGHT');
}
syncInFlight = true;
try {
return await runSync();
} finally {
syncInFlight = false;
}
}
export function isSyncInFlight(): boolean {
return syncInFlight;
}
export function isWatchingForUpdates(): boolean {
return watcherActive;
}