feat(web-sync): add background sync worker and retry strategy
This commit is contained in:
@@ -47,10 +47,33 @@ export type LocalTaskDraftRecord = {
|
||||
updatedAt: number;
|
||||
};
|
||||
|
||||
export type LocalSyncStateRecord = {
|
||||
userId: string;
|
||||
cursor: string | null;
|
||||
lastSyncedAt: number | null;
|
||||
updatedAt: number;
|
||||
};
|
||||
|
||||
export type LocalSyncInboxRecord = {
|
||||
opId: string;
|
||||
userId: string;
|
||||
entityId: string;
|
||||
entityType: SyncEntityType;
|
||||
action: SyncActionType;
|
||||
payload: string | null;
|
||||
clientTs: number;
|
||||
deviceId: string;
|
||||
serverTs: number;
|
||||
receivedAt: number;
|
||||
appliedAt: number | null;
|
||||
};
|
||||
|
||||
class TodoLocalDb extends Dexie {
|
||||
declare tasks: Table<LocalTaskRecord, string>;
|
||||
declare opLogs: Table<LocalOpLogRecord, string>;
|
||||
declare taskDrafts: Table<LocalTaskDraftRecord, string>;
|
||||
declare syncStates: Table<LocalSyncStateRecord, string>;
|
||||
declare syncInbox: Table<LocalSyncInboxRecord, string>;
|
||||
|
||||
constructor() {
|
||||
super("todolist-web-db");
|
||||
@@ -66,9 +89,19 @@ class TodoLocalDb extends Dexie {
|
||||
task_drafts: "&taskId,userId,updatedAt"
|
||||
});
|
||||
|
||||
this.version(3).stores({
|
||||
tasks: "&id,userId,status,priority,ddlAt,updatedAt,deletedAt",
|
||||
op_logs: "&opId,entityId,entityType,action,clientTs,syncedAt",
|
||||
task_drafts: "&taskId,userId,updatedAt",
|
||||
sync_states: "&userId,updatedAt,lastSyncedAt",
|
||||
sync_inbox: "&opId,userId,entityId,serverTs,appliedAt"
|
||||
});
|
||||
|
||||
this.tasks = this.table("tasks");
|
||||
this.opLogs = this.table("op_logs");
|
||||
this.taskDrafts = this.table("task_drafts");
|
||||
this.syncStates = this.table("sync_states");
|
||||
this.syncInbox = this.table("sync_inbox");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
import {
|
||||
localDb,
|
||||
type LocalOpLogRecord,
|
||||
type LocalSyncInboxRecord,
|
||||
type LocalSyncStateRecord
|
||||
} from "@/services/local-db";
|
||||
import type { SyncPullItem } from "@/services/sync-api";
|
||||
|
||||
export const MAX_SYNC_RETRY_COUNT = 5;
|
||||
|
||||
export async function listPendingSyncOperations(limit = 20): Promise<LocalOpLogRecord[]> {
|
||||
const records = await localDb.opLogs.orderBy("clientTs").toArray();
|
||||
|
||||
return records
|
||||
.filter((record) => record.syncedAt === null && record.retryCount < MAX_SYNC_RETRY_COUNT)
|
||||
.slice(0, limit);
|
||||
}
|
||||
|
||||
export async function countPendingSyncOperations(): Promise<number> {
|
||||
const records = await localDb.opLogs.toArray();
|
||||
return records.filter(
|
||||
(record) => record.syncedAt === null && record.retryCount < MAX_SYNC_RETRY_COUNT
|
||||
).length;
|
||||
}
|
||||
|
||||
export async function countBlockedSyncOperations(): Promise<number> {
|
||||
const records = await localDb.opLogs.toArray();
|
||||
return records.filter(
|
||||
(record) => record.syncedAt === null && record.retryCount >= MAX_SYNC_RETRY_COUNT
|
||||
).length;
|
||||
}
|
||||
|
||||
export async function markSyncOperationsSucceeded(
|
||||
opIds: string[],
|
||||
syncedAt: number
|
||||
): Promise<void> {
|
||||
if (opIds.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const records = await localDb.opLogs.bulkGet(opIds);
|
||||
const nextRecords = records
|
||||
.filter((record): record is LocalOpLogRecord => record !== undefined)
|
||||
.map((record) => ({
|
||||
...record,
|
||||
syncedAt,
|
||||
errorMessage: null
|
||||
}));
|
||||
|
||||
if (nextRecords.length > 0) {
|
||||
await localDb.opLogs.bulkPut(nextRecords);
|
||||
}
|
||||
}
|
||||
|
||||
export async function markSyncOperationsFailed(
|
||||
failures: Array<{ opId: string; errorMessage: string }>
|
||||
): Promise<void> {
|
||||
if (failures.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const failureMap = new Map(failures.map((failure) => [failure.opId, failure.errorMessage]));
|
||||
const records = await localDb.opLogs.bulkGet(failures.map((failure) => failure.opId));
|
||||
const nextRecords = records
|
||||
.filter((record): record is LocalOpLogRecord => record !== undefined)
|
||||
.map((record) => ({
|
||||
...record,
|
||||
retryCount: record.retryCount + 1,
|
||||
errorMessage: failureMap.get(record.opId) ?? "同步失败"
|
||||
}));
|
||||
|
||||
if (nextRecords.length > 0) {
|
||||
await localDb.opLogs.bulkPut(nextRecords);
|
||||
}
|
||||
}
|
||||
|
||||
export async function getLocalSyncState(userId: string): Promise<LocalSyncStateRecord | undefined> {
|
||||
return localDb.syncStates.get(userId);
|
||||
}
|
||||
|
||||
export async function saveLocalSyncState(input: {
|
||||
userId: string;
|
||||
cursor: string | null;
|
||||
lastSyncedAt: number | null;
|
||||
}): Promise<void> {
|
||||
await localDb.syncStates.put({
|
||||
userId: input.userId,
|
||||
cursor: input.cursor,
|
||||
lastSyncedAt: input.lastSyncedAt,
|
||||
updatedAt: Date.now()
|
||||
});
|
||||
}
|
||||
|
||||
export async function enqueueRemoteSyncOperations(
|
||||
userId: string,
|
||||
operations: SyncPullItem[]
|
||||
): Promise<number> {
|
||||
if (operations.length === 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const receivedAt = Date.now();
|
||||
const records: LocalSyncInboxRecord[] = operations.map((operation) => ({
|
||||
opId: operation.opId,
|
||||
userId,
|
||||
entityId: operation.entityId,
|
||||
entityType: operation.entityType,
|
||||
action: operation.action,
|
||||
payload: operation.payload,
|
||||
clientTs: operation.clientTs,
|
||||
deviceId: operation.deviceId,
|
||||
serverTs: new Date(operation.serverTs).getTime(),
|
||||
receivedAt,
|
||||
appliedAt: null
|
||||
}));
|
||||
|
||||
await localDb.syncInbox.bulkPut(records);
|
||||
return records.length;
|
||||
}
|
||||
|
||||
export async function countPendingRemoteOperations(userId: string): Promise<number> {
|
||||
const records = await localDb.syncInbox.where("userId").equals(userId).toArray();
|
||||
return records.filter((record) => record.appliedAt === null).length;
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
import type { LocalOpLogRecord } from "@/services/local-db";
|
||||
|
||||
export type SyncPushResult = {
|
||||
acceptedCount: number;
|
||||
duplicateCount: number;
|
||||
failedCount: number;
|
||||
results: Array<{
|
||||
opId: string;
|
||||
status: "accepted" | "duplicate" | "failed";
|
||||
serverTs: string | null;
|
||||
reason: string | null;
|
||||
}>;
|
||||
};
|
||||
|
||||
export type SyncPullItem = {
|
||||
opId: string;
|
||||
entityId: string;
|
||||
entityType: "TASK";
|
||||
action: "CREATE" | "UPDATE" | "DELETE";
|
||||
payload: string | null;
|
||||
clientTs: number;
|
||||
deviceId: string;
|
||||
serverTs: string;
|
||||
};
|
||||
|
||||
export type SyncPullResult = {
|
||||
items: SyncPullItem[];
|
||||
nextCursor: string | null;
|
||||
hasMore: boolean;
|
||||
};
|
||||
|
||||
const DEFAULT_API_BASE_URL = "http://localhost:3000";
|
||||
|
||||
function resolveApiBaseUrl(): string {
|
||||
const envBaseUrl = import.meta.env.VITE_API_BASE_URL as string | undefined;
|
||||
if (!envBaseUrl) {
|
||||
return DEFAULT_API_BASE_URL;
|
||||
}
|
||||
|
||||
return envBaseUrl.replace(/\/+$/, "");
|
||||
}
|
||||
|
||||
async function parseErrorMessage(response: Response): Promise<string> {
|
||||
try {
|
||||
const body = (await response.json()) as { message?: string | string[] };
|
||||
if (Array.isArray(body.message)) {
|
||||
return body.message.join("，");
|
||||
}
|
||||
|
||||
if (typeof body.message === "string" && body.message.trim()) {
|
||||
return body.message;
|
||||
}
|
||||
} catch {
|
||||
return `请求失败（${response.status}）`;
|
||||
}
|
||||
|
||||
return `请求失败（${response.status}）`;
|
||||
}
|
||||
|
||||
export async function pushSyncOperations(
|
||||
userId: string,
|
||||
operations: LocalOpLogRecord[]
|
||||
): Promise<SyncPushResult> {
|
||||
const response = await fetch(`${resolveApiBaseUrl()}/sync/push`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
"x-user-id": userId
|
||||
},
|
||||
body: JSON.stringify({
|
||||
operations: operations.map((operation) => ({
|
||||
opId: operation.opId,
|
||||
entityId: operation.entityId,
|
||||
entityType: operation.entityType,
|
||||
action: operation.action,
|
||||
payload: operation.payload,
|
||||
clientTs: operation.clientTs,
|
||||
deviceId: operation.deviceId
|
||||
}))
|
||||
})
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(await parseErrorMessage(response));
|
||||
}
|
||||
|
||||
return (await response.json()) as SyncPushResult;
|
||||
}
|
||||
|
||||
export async function pullSyncOperations(input: {
|
||||
userId: string;
|
||||
cursor: string | null;
|
||||
limit?: number;
|
||||
}): Promise<SyncPullResult> {
|
||||
const requestUrl = new URL(`${resolveApiBaseUrl()}/sync/pull`);
|
||||
|
||||
if (input.cursor) {
|
||||
requestUrl.searchParams.set("cursor", input.cursor);
|
||||
}
|
||||
|
||||
if (input.limit !== undefined) {
|
||||
requestUrl.searchParams.set("limit", String(input.limit));
|
||||
}
|
||||
|
||||
const response = await fetch(requestUrl, {
|
||||
method: "GET",
|
||||
headers: {
|
||||
"x-user-id": input.userId
|
||||
}
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(await parseErrorMessage(response));
|
||||
}
|
||||
|
||||
return (await response.json()) as SyncPullResult;
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
import {
|
||||
enqueueRemoteSyncOperations,
|
||||
getLocalSyncState,
|
||||
listPendingSyncOperations,
|
||||
markSyncOperationsFailed,
|
||||
markSyncOperationsSucceeded,
|
||||
saveLocalSyncState
|
||||
} from "@/services/local-sync-repo";
|
||||
import { pullSyncOperations, pushSyncOperations } from "@/services/sync-api";
|
||||
|
||||
const PUSH_BATCH_LIMIT = 20;
|
||||
const PULL_BATCH_LIMIT = 100;
|
||||
const MAX_PULL_PAGES_PER_CYCLE = 5;
|
||||
|
||||
export type SyncCycleResult = {
|
||||
pushedCount: number;
|
||||
pulledCount: number;
|
||||
lastSyncedAt: number;
|
||||
hasFailures: boolean;
|
||||
failureMessage: string | null;
|
||||
};
|
||||
|
||||
export async function runSyncWorkerCycle(userId: string): Promise<SyncCycleResult> {
|
||||
const lastSyncedAt = Date.now();
|
||||
let pushedCount = 0;
|
||||
let pulledCount = 0;
|
||||
let hasFailures = false;
|
||||
let failureMessage: string | null = null;
|
||||
|
||||
for (;;) {
|
||||
const pendingOperations = await listPendingSyncOperations(PUSH_BATCH_LIMIT);
|
||||
if (pendingOperations.length === 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
const pushResult = await pushSyncOperations(userId, pendingOperations);
|
||||
const syncedOperationIds = pushResult.results
|
||||
.filter((result) => result.status === "accepted" || result.status === "duplicate")
|
||||
.map((result) => result.opId);
|
||||
const failedOperations = pushResult.results
|
||||
.filter((result) => result.status === "failed")
|
||||
.map((result) => ({
|
||||
opId: result.opId,
|
||||
errorMessage: result.reason ?? "同步失败"
|
||||
}));
|
||||
|
||||
await markSyncOperationsSucceeded(syncedOperationIds, lastSyncedAt);
|
||||
await markSyncOperationsFailed(failedOperations);
|
||||
|
||||
pushedCount += syncedOperationIds.length;
|
||||
|
||||
if (failedOperations.length > 0) {
|
||||
hasFailures = true;
|
||||
failureMessage = failedOperations[0]?.errorMessage ?? "同步失败";
|
||||
break;
|
||||
}
|
||||
|
||||
if (pendingOperations.length < PUSH_BATCH_LIMIT) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const currentState = await getLocalSyncState(userId);
|
||||
let nextCursor = currentState?.cursor ?? null;
|
||||
|
||||
for (let page = 0; page < MAX_PULL_PAGES_PER_CYCLE; page += 1) {
|
||||
const pullResult = await pullSyncOperations({
|
||||
userId,
|
||||
cursor: nextCursor,
|
||||
limit: PULL_BATCH_LIMIT
|
||||
});
|
||||
|
||||
if (pullResult.items.length > 0) {
|
||||
pulledCount += await enqueueRemoteSyncOperations(userId, pullResult.items);
|
||||
}
|
||||
|
||||
nextCursor = pullResult.nextCursor;
|
||||
await saveLocalSyncState({
|
||||
userId,
|
||||
cursor: nextCursor,
|
||||
lastSyncedAt
|
||||
});
|
||||
|
||||
if (!pullResult.hasMore) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (currentState === undefined && nextCursor === null) {
|
||||
await saveLocalSyncState({
|
||||
userId,
|
||||
cursor: null,
|
||||
lastSyncedAt
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
pushedCount,
|
||||
pulledCount,
|
||||
lastSyncedAt,
|
||||
hasFailures,
|
||||
failureMessage
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user