diff --git a/apps/web/src/data/db.ts b/apps/web/src/data/db.ts index 5fa1e95..39e17a5 100644 --- a/apps/web/src/data/db.ts +++ b/apps/web/src/data/db.ts @@ -2,11 +2,18 @@ import Dexie, { type EntityTable } from "dexie"; import type { LedgerEntry, SyncOperation } from "@cents/domain"; import type { LedgerRecord, UserPreference } from "./ledgers"; +export type SyncMetadata = { + id: string; + cursor: string; + updatedAt: string; +}; + export const db = new Dexie("cents") as Dexie & { entries: EntityTable; syncOperations: EntityTable; ledgers: EntityTable; userPreferences: EntityTable; + syncMetadata: EntityTable; }; db.version(1).stores({ @@ -20,3 +27,11 @@ db.version(2).stores({ ledgers: "id, updatedAt, archivedAt", userPreferences: "id, userId, key, updatedAt", }); + +db.version(3).stores({ + entries: "id, ledgerId, occurredAt, updatedAt, deletedAt", + syncOperations: "id, ledgerId, createdAt, syncedAt", + ledgers: "id, updatedAt, archivedAt", + userPreferences: "id, userId, key, updatedAt", + syncMetadata: "id, updatedAt", +}); diff --git a/apps/web/src/router.ts b/apps/web/src/router.ts index 34da4b0..8a6445e 100644 --- a/apps/web/src/router.ts +++ b/apps/web/src/router.ts @@ -26,6 +26,7 @@ export const router = createRouter({ path: "/join-ledger", name: "join-ledger", component: () => import("./views/JoinLedgerView.vue"), + meta: { public: true }, }, { path: "/prototype/entry-detail", diff --git a/apps/web/src/stores/entries.ts b/apps/web/src/stores/entries.ts index 7fba325..49b1013 100644 --- a/apps/web/src/stores/entries.ts +++ b/apps/web/src/stores/entries.ts @@ -1,22 +1,121 @@ import { defineStore } from "pinia"; import type { LedgerEntry, SyncOperation } from "@cents/domain"; +import { apiRequest } from "../data/api"; import { db } from "../data/db"; import { createId } from "../data/ids"; import { useAuthStore } from "./auth"; +let syncPromise: Promise | null = null; +let syncTriggersStarted = false; + export const useEntryStore = defineStore("entries", { state: () => ({ entries: [] as LedgerEntry[], + pendingEntryIds: [] as string[], + pendingEntries: [] as Array<{ id: string; ledgerId: string }>, + syncing: false, + syncError: "", + lastSyncedAt: "" as string, }), actions: { async loadEntries() { + await this.refreshLocalEntries(); + this.startSyncTriggers(); + void this.syncEntries(); + }, + async refreshLocalEntries() { this.entries = (await db.entries.orderBy("occurredAt").reverse().toArray()).filter( (entry) => entry.deletedAt === null, ); + const pendingOperations = await db.syncOperations.filter((operation) => operation.syncedAt === null).toArray(); + this.pendingEntryIds = [...new Set(pendingOperations.map((operation) => operation.entityId))]; + this.pendingEntries = [...new Map( + pendingOperations.map((operation) => [operation.entityId, { id: operation.entityId, ledgerId: operation.ledgerId }]), + ).values()]; + }, + isEntryPending(entryId: string) { + return this.pendingEntryIds.includes(entryId); + }, + pendingEntryCount(ledgerId: string) { + return this.pendingEntries.filter((entry) => entry.ledgerId === ledgerId).length; + }, + startSyncTriggers() { + if (syncTriggersStarted || typeof window === "undefined") return; + syncTriggersStarted = true; + window.addEventListener("online", () => void this.syncEntries()); + document.addEventListener("visibilitychange", () => { + if (document.visibilityState === "visible") void this.syncEntries(); + }); + window.setInterval(() => void this.syncEntries(), 60_000); + }, + async syncEntries() { + const auth = useAuthStore(); + if (!auth.user || (typeof navigator !== "undefined" && !navigator.onLine)) { + await this.refreshLocalEntries(); + return; + } + const userId = auth.user.id; + if (syncPromise) return syncPromise; + + syncPromise = (async () => { + this.syncing = true; + this.syncError = ""; + try { + while (true) { + const pending = (await db.syncOperations + .filter((operation) => operation.syncedAt === null) + .toArray()) + .filter((operation) => (operation.userId ?? operation.payload.updatedBy) === userId) + .sort((left, right) => left.createdAt.localeCompare(right.createdAt)) + .slice(0, 200); + if (!pending.length) break; + const pushed = await apiRequest<{ acceptedOperationIds: string[] }>("/api/sync/push", { + method: "POST", + body: { operations: pending }, + }); + if (!pushed.acceptedOperationIds.length) break; + await db.syncOperations.bulkDelete(pushed.acceptedOperationIds); + } + + const metadata = await db.syncMetadata.get(userId); + let cursor = metadata?.cursor ?? ""; + let hasMore = false; + do { + const path = cursor ? `/api/sync/pull?cursor=${encodeURIComponent(cursor)}` : "/api/sync/pull"; + const pulled = await apiRequest<{ entries: LedgerEntry[]; cursor: string; hasMore: boolean }>(path); + const remainingOperations = await db.syncOperations + .filter((operation) => operation.syncedAt === null) + .toArray(); + const locallyChangedIds = new Set(remainingOperations.map((operation) => operation.entityId)); + const safeCloudEntries = pulled.entries.filter((entry) => !locallyChangedIds.has(entry.id)); + await db.transaction("rw", db.entries, db.syncMetadata, async () => { + if (safeCloudEntries.length) await db.entries.bulkPut(safeCloudEntries); + if (pulled.cursor) { + await db.syncMetadata.put({ id: userId, cursor: pulled.cursor, updatedAt: new Date().toISOString() }); + } + }); + cursor = pulled.cursor; + hasMore = pulled.hasMore; + } while (hasMore); + this.lastSyncedAt = new Date().toISOString(); + } catch (error) { + this.syncError = error instanceof Error ? error.message : "同步失败"; + } finally { + await this.refreshLocalEntries(); + this.syncing = false; + } + })(); + try { + await syncPromise; + } finally { + syncPromise = null; + } }, async addEntry(entry: LedgerEntry) { + const auth = useAuthStore(); const operation: SyncOperation = { id: createId(), + userId: auth.user?.id, ledgerId: entry.ledgerId, entity: "entry", entityId: entry.id, @@ -31,7 +130,8 @@ export const useEntryStore = defineStore("entries", { await db.syncOperations.put(operation); }); - await this.loadEntries(); + await this.refreshLocalEntries(); + void this.syncEntries(); }, async updateEntry(entry: LedgerEntry) { const auth = useAuthStore(); @@ -47,6 +147,7 @@ export const useEntryStore = defineStore("entries", { }; const operation: SyncOperation = { id: createId(), + userId: auth.user?.id, ledgerId: updated.ledgerId, entity: "entry", entityId: updated.id, @@ -61,7 +162,8 @@ export const useEntryStore = defineStore("entries", { await db.syncOperations.put(operation); }); - await this.loadEntries(); + await this.refreshLocalEntries(); + void this.syncEntries(); return updated; }, async deleteEntry(entryId: string) { @@ -79,6 +181,7 @@ export const useEntryStore = defineStore("entries", { }; const operation: SyncOperation = { id: createId(), + userId: auth.user?.id, ledgerId: deleted.ledgerId, entity: "entry", entityId: deleted.id, @@ -93,7 +196,8 @@ export const useEntryStore = defineStore("entries", { await db.syncOperations.put(operation); }); - await this.loadEntries(); + await this.refreshLocalEntries(); + void this.syncEntries(); }, }, }); diff --git a/apps/web/src/stores/ledgers.ts b/apps/web/src/stores/ledgers.ts index a8062fd..3e70282 100644 --- a/apps/web/src/stores/ledgers.ts +++ b/apps/web/src/stores/ledgers.ts @@ -100,6 +100,12 @@ export const useLedgerStore = defineStore("ledgers", { memberName(userId: string) { return this.currentMembers.find((member) => member.id === userId)?.name ?? "历史用户"; }, + async createLedger(input: Pick) { + const result = await apiRequest<{ ledger: LedgerRecord }>("/api/ledgers", { method: "POST", body: input }); + await this.reloadLedgers(); + await this.setCurrentLedger(result.ledger.id); + return result.ledger; + }, async updateLedger(input: Pick) { await apiRequest(`/api/ledgers/${encodeURIComponent(input.id)}`, { method: "PATCH", body: input }); await this.reloadLedgers(); diff --git a/apps/web/src/styles.css b/apps/web/src/styles.css index 3077441..3418864 100644 --- a/apps/web/src/styles.css +++ b/apps/web/src/styles.css @@ -309,6 +309,41 @@ input:focus-visible { gap: 3px; } +.sync-notice { + min-height: 44px; + display: flex; + align-items: center; + justify-content: space-between; + gap: 10px; + border-bottom: 1px solid #efd9c9; + padding: 6px 16px; + background: #fff8f2; + color: #8e481e; + font-size: 12px; +} + +.sync-notice > span, +.sync-notice button { + display: inline-flex; + align-items: center; + gap: 5px; +} + +.sync-notice button { + flex: 0 0 auto; + min-height: 32px; + border: 0; + border-radius: 7px; + padding: 0 9px; + background: #f0e2d7; + color: #7a3d1b; + font-weight: 680; +} + +.sync-notice button:disabled { opacity: .6; } +.sync-notice .spinning { animation: sync-spin .8s linear infinite; } +@keyframes sync-spin { to { transform: rotate(360deg); } } + .list-toolbar-heading { display: flex; align-items: baseline; @@ -447,6 +482,13 @@ input:focus-visible { white-space: nowrap; } +.entry-copy .entry-sync-pending { + display: inline-flex; + flex: 0 0 auto; + align-items: center; + color: #b35a24; +} + .entry-copy .entry-category-label::before { margin-right: 7px; color: #c4cecb; diff --git a/apps/web/src/views/EntryDetailView.vue b/apps/web/src/views/EntryDetailView.vue index c285ff4..b8cab3e 100644 --- a/apps/web/src/views/EntryDetailView.vue +++ b/apps/web/src/views/EntryDetailView.vue @@ -8,6 +8,7 @@ import { ChevronDown, ChevronRight, CircleUserRound, + CloudOff, NotebookText, Trash2, WalletCards, @@ -272,7 +273,10 @@ async function deleteEntry() { -

由{{ ledgerStore.memberName(entry.createdBy) }}创建 · {{ createdAtLabel }}

+

+ + 由{{ ledgerStore.memberName(entry.createdBy) }}创建 · {{ createdAtLabel }} +

@@ -578,6 +582,14 @@ async function deleteEntry() { font-size: 12px; } +.entry-sync-pending { + display: inline-flex; + align-items: center; + margin-right: 4px; + color: #b35a24; + vertical-align: -2px; +} + .entry-save-error { margin: 8px 16px; color: #c94635; diff --git a/apps/web/src/views/JoinLedgerView.vue b/apps/web/src/views/JoinLedgerView.vue index 5bb5213..efaa016 100644 --- a/apps/web/src/views/JoinLedgerView.vue +++ b/apps/web/src/views/JoinLedgerView.vue @@ -1,8 +1,9 @@ diff --git a/apps/web/src/views/LoginView.vue b/apps/web/src/views/LoginView.vue index e251a1a..1daff31 100644 --- a/apps/web/src/views/LoginView.vue +++ b/apps/web/src/views/LoginView.vue @@ -18,6 +18,7 @@ const errorMessage = ref(""); const redirect = computed(() => typeof route.query.redirect === "string" && route.query.redirect.startsWith("/") ? route.query.redirect : "/"); +const joiningLedger = computed(() => redirect.value.startsWith("/join-ledger?")); async function login() { if (!name.value.trim() || !password.value || submitting.value) return; @@ -39,14 +40,18 @@ async function login() {
Cents
diff --git a/packages/domain/src/index.ts b/packages/domain/src/index.ts index cc0b78a..cbfc356 100644 --- a/packages/domain/src/index.ts +++ b/packages/domain/src/index.ts @@ -29,6 +29,7 @@ export type SyncOperationAction = "create" | "update" | "delete"; export type SyncOperation = { id: string; + userId?: string; ledgerId: string; entity: "entry"; entityId: string; @@ -37,4 +38,3 @@ export type SyncOperation = { createdAt: string; syncedAt: string | null; }; - diff --git a/services/api/src/db.ts b/services/api/src/db.ts index 488db6f..bbe0cb3 100644 --- a/services/api/src/db.ts +++ b/services/api/src/db.ts @@ -81,6 +81,37 @@ CREATE TABLE IF NOT EXISTS sessions ( ); CREATE INDEX IF NOT EXISTS sessions_lookup ON sessions (token_hash, expires_at); + +CREATE TABLE IF NOT EXISTS entries ( + id text PRIMARY KEY, + ledger_id text NOT NULL REFERENCES ledgers(id) ON DELETE CASCADE, + type varchar(8) NOT NULL CHECK (type IN ('expense', 'income')), + amount integer NOT NULL CHECK (amount > 0), + currency varchar(3) NOT NULL, + base_currency varchar(3) NOT NULL, + base_amount integer NOT NULL CHECK (base_amount > 0), + exchange_rate varchar(32) NOT NULL, + exchange_rate_source varchar(8) NOT NULL CHECK (exchange_rate_source IN ('manual', 'system')), + category_id varchar(100) NOT NULL, + note varchar(500) NOT NULL DEFAULT '', + occurred_at timestamptz NOT NULL, + created_by text REFERENCES users(id) ON DELETE SET NULL, + updated_by text REFERENCES users(id) ON DELETE SET NULL, + created_at timestamptz NOT NULL, + updated_at timestamptz NOT NULL, + deleted_at timestamptz, + version integer NOT NULL CHECK (version > 0), + server_updated_at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS entries_ledger_occurred + ON entries (ledger_id, occurred_at DESC); + +CREATE TABLE IF NOT EXISTS entry_sync_operations ( + id text PRIMARY KEY, + user_id text NOT NULL REFERENCES users(id) ON DELETE CASCADE, + processed_at timestamptz NOT NULL DEFAULT now() +); `; export async function initializeDatabase() { diff --git a/services/api/src/server.ts b/services/api/src/server.ts index 5c9b282..80af9cc 100644 --- a/services/api/src/server.ts +++ b/services/api/src/server.ts @@ -1,6 +1,7 @@ import cookie from "@fastify/cookie"; import helmet from "@fastify/helmet"; import rateLimit from "@fastify/rate-limit"; +import type { LedgerEntry, SyncOperation } from "@cents/domain"; import Fastify, { type FastifyReply, type FastifyRequest } from "fastify"; import { initializeDatabase, pool } from "./db.js"; import { @@ -46,6 +47,7 @@ server.addHook("onSend", async (request, reply, payload) => { request.url.startsWith("/api/auth/") || request.url.startsWith("/api/invitations/") || request.url.startsWith("/api/ledger-invitations/") + || request.url.startsWith("/api/sync/") || request.url === "/api/me" ) { reply.header("Cache-Control", "no-store"); @@ -112,6 +114,59 @@ function invitationStatus(invitation: { expiresAt: Date | string; acceptedAt: Da return "valid"; } +const currencies = new Set(["CNY", "USD", "EUR", "JPY", "HKD"]); + +function validDate(value: unknown) { + return typeof value === "string" && !Number.isNaN(Date.parse(value)); +} + +function validEntry(value: unknown): value is LedgerEntry { + if (!value || typeof value !== "object") return false; + const entry = value as Partial; + return typeof entry.id === "string" && entry.id.length > 0 && entry.id.length <= 100 + && typeof entry.ledgerId === "string" && entry.ledgerId.length > 0 && entry.ledgerId.length <= 100 + && (entry.type === "expense" || entry.type === "income") + && Number.isSafeInteger(entry.amount) && entry.amount! > 0 && entry.amount! <= 2_147_483_647 + && typeof entry.currency === "string" && currencies.has(entry.currency) + && typeof entry.baseCurrency === "string" && currencies.has(entry.baseCurrency) + && Number.isSafeInteger(entry.baseAmount) && entry.baseAmount! > 0 && entry.baseAmount! <= 2_147_483_647 + && typeof entry.exchangeRate === "string" && entry.exchangeRate.length > 0 && entry.exchangeRate.length <= 32 + && (entry.exchangeRateSource === "manual" || entry.exchangeRateSource === "system") + && typeof entry.categoryId === "string" && entry.categoryId.length > 0 && entry.categoryId.length <= 100 + && typeof entry.note === "string" && entry.note.length <= 500 + && validDate(entry.occurredAt) && validDate(entry.createdAt) && validDate(entry.updatedAt) + && (entry.deletedAt === null || validDate(entry.deletedAt)) + && Number.isSafeInteger(entry.version) && entry.version! > 0; +} + +function validSyncOperation(value: unknown): value is SyncOperation { + if (!value || typeof value !== "object") return false; + const operation = value as Partial; + return typeof operation.id === "string" && operation.id.length > 0 && operation.id.length <= 100 + && operation.entity === "entry" + && (operation.action === "create" || operation.action === "update" || operation.action === "delete") + && validEntry(operation.payload) + && operation.entityId === operation.payload.id + && operation.ledgerId === operation.payload.ledgerId + && (operation.action !== "delete" || operation.payload.deletedAt !== null); +} + +type SyncCursor = { serverUpdatedAt: string; id: string }; + +function encodeSyncCursor(cursor: SyncCursor) { + return Buffer.from(JSON.stringify(cursor)).toString("base64url"); +} + +function decodeSyncCursor(value: string): SyncCursor | null { + try { + const cursor = JSON.parse(Buffer.from(value, "base64url").toString("utf8")) as Partial; + if (!validDate(cursor.serverUpdatedAt) || typeof cursor.id !== "string" || cursor.id.length > 100) return null; + return { serverUpdatedAt: cursor.serverUpdatedAt!, id: cursor.id }; + } catch { + return null; + } +} + server.get("/health", async () => ({ ok: true, service: "cents-api" })); server.get("/api/auth/session", async (request, reply) => ({ user: await currentUser(request, reply) })); @@ -176,6 +231,46 @@ server.get("/api/ledgers", async (request, reply) => { return { ledgers: result.rows }; }); +server.post<{ Body: { name?: string; color?: string; defaultCurrency?: string } }>( + "/api/ledgers", + async (request, reply) => { + const user = await requireUser(request, reply); + if (!user) return; + const name = request.body.name?.trim() ?? ""; + const color = request.body.color ?? "#087f72"; + const defaultCurrency = request.body.defaultCurrency ?? "CNY"; + if (!name || name.length > 40) return reply.code(400).send({ error: "账本名称应为 1 至 40 个字符" }); + if (!/^#[0-9a-f]{6}$/i.test(color)) return reply.code(400).send({ error: "账本颜色无效" }); + if (!["CNY", "USD", "EUR", "JPY", "HKD"].includes(defaultCurrency)) { + return reply.code(400).send({ error: "默认币种无效" }); + } + + const client = await pool.connect(); + try { + await client.query("BEGIN"); + const ledgerId = createId(); + const result = await client.query( + `INSERT INTO ledgers (id, name, color, default_currency) + VALUES ($1, $2, $3, $4) + RETURNING id, name, color, default_currency AS "defaultCurrency", + created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt"`, + [ledgerId, name, color, defaultCurrency], + ); + await client.query( + "INSERT INTO ledger_members (ledger_id, user_id, role) VALUES ($1, $2, 'owner')", + [ledgerId, user.id], + ); + await client.query("COMMIT"); + return reply.code(201).send({ ledger: { ...result.rows[0], role: "owner" } }); + } catch (error) { + await client.query("ROLLBACK"); + throw error; + } finally { + client.release(); + } + }, +); + server.get<{ Params: { ledgerId: string } }>("/api/ledgers/:ledgerId/members", async (request, reply) => { const user = await requireUser(request, reply); if (!user) return; @@ -344,6 +439,10 @@ server.post<{ Params: { key: string } }>("/api/ledger-invitations/:key/accept", ON CONFLICT (ledger_id, user_id) DO UPDATE SET removed_at = NULL, joined_at = now()`, [invitation!.ledgerId, user.id, invitation!.role], ); + await client.query( + "UPDATE entries SET server_updated_at = now() WHERE ledger_id = $1", + [invitation!.ledgerId], + ); await client.query( "UPDATE ledger_invitations SET accepted_by = $1, accepted_at = now() WHERE id = $2", [user.id, invitation!.id], @@ -358,14 +457,139 @@ server.post<{ Params: { key: string } }>("/api/ledger-invitations/:key/accept", } }); -server.post("/api/sync/push", async (request, reply) => { - if (!await requireUser(request, reply)) return; - return { accepted: true, operations: [] }; +server.post<{ Body: { operations?: unknown[] } }>("/api/sync/push", async (request, reply) => { + const user = await requireUser(request, reply); + if (!user) return; + const operations = request.body.operations ?? []; + if (!Array.isArray(operations) || operations.length > 200 || !operations.every(validSyncOperation)) { + return reply.code(400).send({ error: "同步数据无效" }); + } + + const client = await pool.connect(); + const acceptedOperationIds: string[] = []; + try { + await client.query("BEGIN"); + for (const operation of operations) { + const alreadyProcessed = await client.query( + "SELECT 1 FROM entry_sync_operations WHERE id = $1", + [operation.id], + ); + if (alreadyProcessed.rowCount) { + acceptedOperationIds.push(operation.id); + continue; + } + + const targetRole = await client.query( + `SELECT role FROM ledger_members + WHERE ledger_id = $1 AND user_id = $2 AND removed_at IS NULL`, + [operation.ledgerId, user.id], + ); + if (!targetRole.rowCount) { + await client.query("ROLLBACK"); + return reply.code(403).send({ error: "无权同步该账本的流水" }); + } + const existing = await client.query<{ ledgerId: string }>( + `SELECT ledger_id AS "ledgerId" FROM entries WHERE id = $1`, + [operation.entityId], + ); + if (existing.rows[0] && existing.rows[0].ledgerId !== operation.ledgerId) { + const sourceRole = await client.query( + `SELECT role FROM ledger_members + WHERE ledger_id = $1 AND user_id = $2 AND removed_at IS NULL`, + [existing.rows[0].ledgerId, user.id], + ); + if (!sourceRole.rowCount) { + await client.query("ROLLBACK"); + return reply.code(403).send({ error: "无权移动该流水" }); + } + } + + const entry = operation.payload; + await client.query( + `INSERT INTO entries ( + id, ledger_id, type, amount, currency, base_currency, base_amount, + exchange_rate, exchange_rate_source, category_id, note, occurred_at, + created_by, updated_by, created_at, updated_at, deleted_at, version + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, + $13, $14, $15, $16, $17, $18 + ) + ON CONFLICT (id) DO UPDATE SET + ledger_id = EXCLUDED.ledger_id, + type = EXCLUDED.type, + amount = EXCLUDED.amount, + currency = EXCLUDED.currency, + base_currency = EXCLUDED.base_currency, + base_amount = EXCLUDED.base_amount, + exchange_rate = EXCLUDED.exchange_rate, + exchange_rate_source = EXCLUDED.exchange_rate_source, + category_id = EXCLUDED.category_id, + note = EXCLUDED.note, + occurred_at = EXCLUDED.occurred_at, + updated_by = EXCLUDED.updated_by, + updated_at = EXCLUDED.updated_at, + deleted_at = EXCLUDED.deleted_at, + version = EXCLUDED.version, + server_updated_at = now() + WHERE entries.version < EXCLUDED.version + OR (entries.version = EXCLUDED.version AND entries.updated_at <= EXCLUDED.updated_at)`, + [ + entry.id, entry.ledgerId, entry.type, entry.amount, entry.currency, entry.baseCurrency, + entry.baseAmount, entry.exchangeRate, entry.exchangeRateSource, entry.categoryId, + entry.note, entry.occurredAt, user.id, user.id, entry.createdAt, entry.updatedAt, + entry.deletedAt, entry.version, + ], + ); + await client.query("UPDATE entries SET server_updated_at = now() WHERE id = $1", [entry.id]); + await client.query( + "INSERT INTO entry_sync_operations (id, user_id) VALUES ($1, $2)", + [operation.id, user.id], + ); + acceptedOperationIds.push(operation.id); + } + await client.query("COMMIT"); + return { acceptedOperationIds }; + } catch (error) { + await client.query("ROLLBACK"); + throw error; + } finally { + client.release(); + } }); -server.get("/api/sync/pull", async (request, reply) => { - if (!await requireUser(request, reply)) return; - return { cursor: null, operations: [] }; +server.get<{ Querystring: { cursor?: string } }>("/api/sync/pull", async (request, reply) => { + const user = await requireUser(request, reply); + if (!user) return; + const cursor = request.query.cursor ? decodeSyncCursor(request.query.cursor) : null; + if (request.query.cursor && !cursor) return reply.code(400).send({ error: "同步游标无效" }); + const result = await pool.query( + `SELECT e.id, e.ledger_id AS "ledgerId", e.type, e.amount, + e.currency, e.base_currency AS "baseCurrency", e.base_amount AS "baseAmount", + e.exchange_rate AS "exchangeRate", e.exchange_rate_source AS "exchangeRateSource", + e.category_id AS "categoryId", e.note, e.occurred_at AS "occurredAt", + e.created_by AS "createdBy", e.updated_by AS "updatedBy", + e.created_at AS "createdAt", e.updated_at AS "updatedAt", + e.deleted_at AS "deletedAt", e.version, + e.server_updated_at AS "serverUpdatedAt" + FROM entries e JOIN ledger_members m ON m.ledger_id = e.ledger_id + WHERE m.user_id = $1 AND m.removed_at IS NULL + AND ($2::timestamptz IS NULL + OR e.server_updated_at > $2::timestamptz + OR (e.server_updated_at = $2::timestamptz AND e.id > $3)) + ORDER BY e.server_updated_at, e.id + LIMIT 501`, + [user.id, cursor?.serverUpdatedAt ?? null, cursor?.id ?? ""], + ); + const pageRows = result.rows.slice(0, 500); + const lastRow = pageRows.at(-1); + const entries = pageRows.map(({ serverUpdatedAt: _serverUpdatedAt, ...entry }) => entry); + return { + entries, + cursor: lastRow + ? encodeSyncCursor({ serverUpdatedAt: lastRow.serverUpdatedAt.toISOString(), id: lastRow.id }) + : request.query.cursor ?? "", + hasMore: result.rows.length > 500, + }; }); const port = Number(process.env.PORT ?? 3000);