421 lines
17 KiB
TypeScript
421 lines
17 KiB
TypeScript
import { defineStore } from "pinia";
|
|
import { calculateCnyAmount, type CurrencyCode, type LedgerEntry, type SyncOperation } from "@cents/domain";
|
|
import { apiRequest } from "../data/api";
|
|
import { getUserDb } from "../data/db";
|
|
import { localDate } from "../data/entries";
|
|
import { createId } from "../data/ids";
|
|
import { reimbursementsForExpense, reimbursedOriginalAmount } from "../data/reimbursements";
|
|
import { useAuthStore } from "./auth";
|
|
import { useLedgerStore } from "./ledgers";
|
|
|
|
let syncPromise: Promise<void> | null = null;
|
|
let syncTriggersStarted = false;
|
|
let conversionPollTimer: number | null = null;
|
|
let conversionPollAttempts = 0;
|
|
let conversionPollDeadline = 0;
|
|
const conversionPollEntryIds = new Set<string>();
|
|
|
|
function scheduleConversionPoll(entryIds: string[], resetAttempts = true) {
|
|
for (const entryId of entryIds) conversionPollEntryIds.add(entryId);
|
|
if (conversionPollTimer !== null || typeof window === "undefined") return;
|
|
if (resetAttempts) {
|
|
conversionPollAttempts = 0;
|
|
conversionPollDeadline = Date.now() + 60_000;
|
|
}
|
|
const remainingMs = conversionPollDeadline - Date.now();
|
|
if (remainingMs <= 0) {
|
|
conversionPollEntryIds.clear();
|
|
return;
|
|
}
|
|
const delayMs = Math.min(1_500 * 2 ** conversionPollAttempts, remainingMs);
|
|
const poll = async () => {
|
|
conversionPollTimer = null;
|
|
const store = useEntryStore();
|
|
await store.syncEntries();
|
|
await store.refreshLocalEntries();
|
|
const remaining = [...conversionPollEntryIds].filter((entryId) =>
|
|
store.entries.some((entry) => entry.id === entryId && entry.conversionStatus === "pending"),
|
|
);
|
|
conversionPollEntryIds.clear();
|
|
if (!remaining.length || Date.now() >= conversionPollDeadline) return;
|
|
conversionPollAttempts += 1;
|
|
scheduleConversionPoll(remaining, false);
|
|
};
|
|
conversionPollTimer = window.setTimeout(() => void poll(), delayMs);
|
|
}
|
|
|
|
export const useEntryStore = defineStore("entries", {
|
|
state: () => ({
|
|
entries: [] as LedgerEntry[],
|
|
loadedForUserId: "",
|
|
pendingEntryIds: [] as string[],
|
|
pendingEntries: [] as Array<{ id: string; ledgerIds: string[] }>,
|
|
syncing: false,
|
|
syncError: "",
|
|
lastSyncedAt: "" as string,
|
|
}),
|
|
actions: {
|
|
async loadEntries() {
|
|
const auth = useAuthStore();
|
|
if (this.loadedForUserId !== (auth.user?.id ?? "")) {
|
|
this.entries = [];
|
|
this.pendingEntryIds = [];
|
|
this.pendingEntries = [];
|
|
this.loadedForUserId = auth.user?.id ?? "";
|
|
}
|
|
await useLedgerStore().loadLedgers();
|
|
await this.refreshLocalEntries();
|
|
this.startSyncTriggers();
|
|
void this.syncEntries();
|
|
},
|
|
async refreshLocalEntries() {
|
|
const auth = useAuthStore();
|
|
if (!auth.user) {
|
|
this.entries = [];
|
|
this.pendingEntryIds = [];
|
|
this.pendingEntries = [];
|
|
this.loadedForUserId = "";
|
|
return;
|
|
}
|
|
const db = await getUserDb(auth.user.id);
|
|
const allowedLedgerIds = new Set(useLedgerStore().ledgers.map((ledger) => ledger.id));
|
|
const canAccessEntry = (entry: LedgerEntry) =>
|
|
entry.ownerId === auth.user!.id || entry.ledgerIds.some((ledgerId) => allowedLedgerIds.has(ledgerId));
|
|
this.entries = (await db.entries.orderBy("occurredAt").reverse().toArray()).filter(
|
|
(entry) => entry.deletedAt === null && canAccessEntry(entry),
|
|
);
|
|
const pendingOperations = (await db.syncOperations.filter((operation) => operation.syncedAt === null).toArray())
|
|
.filter((operation) =>
|
|
(operation.userId ?? operation.payload.updatedBy) === auth.user!.id,
|
|
);
|
|
this.pendingEntryIds = [...new Set(pendingOperations.map((operation) => operation.entityId))];
|
|
this.pendingEntries = [...new Map(
|
|
pendingOperations.map((operation) => [operation.entityId, { id: operation.entityId, ledgerIds: operation.ledgerIds }]),
|
|
).values()];
|
|
},
|
|
isEntryPending(entryId: string) {
|
|
return this.pendingEntryIds.includes(entryId);
|
|
},
|
|
pendingEntryCount(ledgerId: string) {
|
|
return this.pendingEntries.filter((entry) => entry.ledgerIds.includes(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;
|
|
const db = await getUserDb(userId);
|
|
if (syncPromise) return syncPromise;
|
|
|
|
syncPromise = (async () => {
|
|
this.syncing = true;
|
|
this.syncError = "";
|
|
try {
|
|
const conversionIds = new Set<string>();
|
|
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;
|
|
for (const operation of pending) {
|
|
if (operation.action !== "delete" && operation.payload.currency !== "CNY" && operation.payload.conversionStatus === "pending") {
|
|
conversionIds.add(operation.entityId);
|
|
}
|
|
}
|
|
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) && (entry.ownerId === userId || entry.ledgerIds.length > 0),
|
|
);
|
|
const inaccessibleEntryIds = pulled.entries
|
|
.filter((entry) => !locallyChangedIds.has(entry.id) && entry.ownerId !== userId && entry.ledgerIds.length === 0)
|
|
.map((entry) => entry.id);
|
|
await db.transaction("rw", db.entries, db.syncMetadata, async () => {
|
|
if (safeCloudEntries.length) await db.entries.bulkPut(safeCloudEntries);
|
|
if (inaccessibleEntryIds.length) await db.entries.bulkDelete(inaccessibleEntryIds);
|
|
if (pulled.cursor) {
|
|
await db.syncMetadata.put({ id: userId, cursor: pulled.cursor, updatedAt: new Date().toISOString() });
|
|
}
|
|
});
|
|
cursor = pulled.cursor;
|
|
hasMore = pulled.hasMore;
|
|
} while (hasMore);
|
|
if (conversionIds.size) scheduleConversionPoll([...conversionIds]);
|
|
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();
|
|
if (entry.reimbursementOfEntryId) {
|
|
const expense = this.entries.find((item) => item.id === entry.reimbursementOfEntryId);
|
|
if (
|
|
!expense
|
|
|| expense.deletedAt
|
|
|| expense.type !== "expense"
|
|
|| expense.ownerId !== auth.user?.id
|
|
) throw new Error("关联的原支出无效");
|
|
const sameLedgers = [...entry.ledgerIds].sort().join(",") === [...expense.ledgerIds].sort().join(",");
|
|
if (
|
|
entry.type !== "income"
|
|
|| entry.currency !== expense.currency
|
|
|| entry.categoryId !== expense.categoryId
|
|
|| !sameLedgers
|
|
) throw new Error("报销必须继承原支出的币种、分类和账本");
|
|
if (entry.amount > expense.amount - reimbursedOriginalAmount(expense, this.entries)) {
|
|
throw new Error("报销金额超过剩余可报销金额");
|
|
}
|
|
}
|
|
const db = await getUserDb(auth.user?.id ?? entry.ownerId);
|
|
const operation: SyncOperation = {
|
|
id: createId(),
|
|
userId: auth.user?.id,
|
|
ledgerIds: entry.ledgerIds,
|
|
entity: "entry",
|
|
entityId: entry.id,
|
|
action: "create",
|
|
payload: entry,
|
|
createdAt: new Date().toISOString(),
|
|
syncedAt: null,
|
|
};
|
|
|
|
await db.transaction("rw", db.entries, db.syncOperations, async () => {
|
|
await db.entries.put(entry);
|
|
await db.syncOperations.put(operation);
|
|
});
|
|
|
|
await this.refreshLocalEntries();
|
|
void this.resolveLocalConversion(entry.id);
|
|
void this.syncEntries();
|
|
},
|
|
async updateEntry(entry: LedgerEntry) {
|
|
const auth = useAuthStore();
|
|
const db = await getUserDb(auth.user?.id ?? entry.ownerId);
|
|
const existing = await db.entries.get(entry.id);
|
|
if (!existing || existing.deletedAt) throw new Error("Entry not found");
|
|
if (existing.reimbursementOfEntryId !== entry.reimbursementOfEntryId) {
|
|
throw new Error("不能修改报销关联");
|
|
}
|
|
if (entry.reimbursementOfEntryId) {
|
|
const expense = this.entries.find((item) => item.id === entry.reimbursementOfEntryId);
|
|
if (!expense || expense.deletedAt || expense.ownerId !== auth.user?.id) {
|
|
throw new Error("关联的原支出无效");
|
|
}
|
|
const otherReimbursements = reimbursementsForExpense(this.entries, expense.id)
|
|
.filter((item) => item.id !== entry.id)
|
|
.reduce((sum, item) => sum + item.amount, 0);
|
|
if (
|
|
entry.type !== "income"
|
|
|| entry.currency !== expense.currency
|
|
|| entry.categoryId !== expense.categoryId
|
|
|| entry.amount + otherReimbursements > expense.amount
|
|
) throw new Error("报销金额超过剩余可报销金额");
|
|
} else if (existing.type === "expense") {
|
|
const reimbursed = reimbursedOriginalAmount(existing, this.entries);
|
|
const sameLedgers = [...existing.ledgerIds].sort().join(",") === [...entry.ledgerIds].sort().join(",");
|
|
if (
|
|
reimbursed > entry.amount
|
|
|| (reimbursed > 0 && (
|
|
existing.currency !== entry.currency
|
|
|| existing.categoryId !== entry.categoryId
|
|
|| !sameLedgers
|
|
))
|
|
) {
|
|
throw new Error("存在报销时不能缩小支出金额、改变分类、币种或账本");
|
|
}
|
|
}
|
|
|
|
const exchangeRateDate = localDate(entry.occurredAt);
|
|
const conversionChanged = existing.amount !== entry.amount
|
|
|| existing.currency !== entry.currency
|
|
|| existing.exchangeRateDate !== exchangeRateDate;
|
|
const updated: LedgerEntry = {
|
|
...existing,
|
|
...entry,
|
|
ledgerIds: [...new Set(entry.ledgerIds)],
|
|
baseCurrency: "CNY",
|
|
baseAmount: conversionChanged ? (entry.currency === "CNY" ? entry.amount : null) : existing.baseAmount,
|
|
exchangeRate: conversionChanged ? (entry.currency === "CNY" ? "1" : null) : existing.exchangeRate,
|
|
exchangeRateDate,
|
|
exchangeRateEffectiveDate: conversionChanged
|
|
? (entry.currency === "CNY" ? exchangeRateDate : null)
|
|
: existing.exchangeRateEffectiveDate,
|
|
exchangeRateSource: conversionChanged ? "system" : existing.exchangeRateSource,
|
|
conversionStatus: conversionChanged
|
|
? (entry.currency === "CNY" ? "exact" : "pending")
|
|
: existing.conversionStatus,
|
|
updatedAt: new Date().toISOString(),
|
|
updatedBy: auth.user?.id ?? existing.updatedBy,
|
|
version: existing.version + 1,
|
|
};
|
|
const operation: SyncOperation = {
|
|
id: createId(),
|
|
userId: auth.user?.id,
|
|
ledgerIds: updated.ledgerIds,
|
|
entity: "entry",
|
|
entityId: updated.id,
|
|
action: "update",
|
|
payload: updated,
|
|
createdAt: updated.updatedAt,
|
|
syncedAt: null,
|
|
};
|
|
|
|
await db.transaction("rw", db.entries, db.syncOperations, async () => {
|
|
await db.entries.put(updated);
|
|
await db.syncOperations.put(operation);
|
|
});
|
|
|
|
await this.refreshLocalEntries();
|
|
if (conversionChanged) void this.resolveLocalConversion(updated.id);
|
|
void this.syncEntries();
|
|
return updated;
|
|
},
|
|
async resolveLocalConversion(entryId: string) {
|
|
const auth = useAuthStore();
|
|
if (!auth.user) return;
|
|
const db = await getUserDb(auth.user.id);
|
|
const entry = await db.entries.get(entryId);
|
|
if (!entry || entry.deletedAt || entry.currency === "CNY") return;
|
|
const cacheId = `${entry.currency}:${entry.exchangeRateDate}`;
|
|
let cached = await db.exchangeRates.get(cacheId);
|
|
if (!cached && (typeof navigator === "undefined" || navigator.onLine)) {
|
|
try {
|
|
const result = await apiRequest<{
|
|
rate: {
|
|
currency: CurrencyCode;
|
|
requestedDate: string;
|
|
effectiveDate: string | null;
|
|
cnyPerUnit: string | null;
|
|
status: "exact" | "fallback" | "pending";
|
|
};
|
|
}>(`/api/exchange-rates/${entry.currency}/${entry.exchangeRateDate}`);
|
|
if (result.rate.cnyPerUnit && result.rate.effectiveDate && result.rate.status !== "pending") {
|
|
cached = {
|
|
id: cacheId,
|
|
currency: result.rate.currency,
|
|
requestedDate: result.rate.requestedDate,
|
|
effectiveDate: result.rate.effectiveDate,
|
|
cnyPerUnit: result.rate.cnyPerUnit,
|
|
status: result.rate.status,
|
|
updatedAt: new Date().toISOString(),
|
|
};
|
|
await db.exchangeRates.put(cached);
|
|
}
|
|
} catch {
|
|
// The original-currency entry is already durable; conversion can retry after sync.
|
|
}
|
|
}
|
|
if (!cached) {
|
|
const candidates = await db.exchangeRates.where("currency").equals(entry.currency).toArray();
|
|
cached = candidates
|
|
.sort((left, right) => {
|
|
const leftFuture = left.requestedDate > entry.exchangeRateDate;
|
|
const rightFuture = right.requestedDate > entry.exchangeRateDate;
|
|
if (leftFuture !== rightFuture) return leftFuture ? 1 : -1;
|
|
return Math.abs(Date.parse(left.requestedDate) - Date.parse(entry.exchangeRateDate))
|
|
- Math.abs(Date.parse(right.requestedDate) - Date.parse(entry.exchangeRateDate));
|
|
})[0];
|
|
}
|
|
if (!cached) return;
|
|
|
|
const current = await db.entries.get(entryId);
|
|
if (
|
|
!current
|
|
|| current.deletedAt
|
|
|| current.amount !== entry.amount
|
|
|| current.currency !== entry.currency
|
|
|| current.exchangeRateDate !== entry.exchangeRateDate
|
|
) return;
|
|
const baseAmount = calculateCnyAmount(current.amount, current.currency, cached.cnyPerUnit);
|
|
await db.entries.update(entryId, {
|
|
baseAmount,
|
|
exchangeRate: cached.cnyPerUnit,
|
|
exchangeRateEffectiveDate: cached.effectiveDate,
|
|
exchangeRateSource: "system",
|
|
conversionStatus: cached.requestedDate === current.exchangeRateDate && cached.status === "exact"
|
|
? "exact"
|
|
: "fallback",
|
|
});
|
|
await this.refreshLocalEntries();
|
|
},
|
|
async deleteEntry(entryId: string) {
|
|
const auth = useAuthStore();
|
|
if (!auth.user) return;
|
|
const db = await getUserDb(auth.user.id);
|
|
const existing = await db.entries.get(entryId);
|
|
if (!existing || existing.deletedAt) return;
|
|
if (existing.type === "expense" && reimbursementsForExpense(this.entries, entryId).length) {
|
|
throw new Error("存在报销记录，不能删除原支出");
|
|
}
|
|
|
|
const deletedAt = new Date().toISOString();
|
|
const deleted: LedgerEntry = {
|
|
...existing,
|
|
deletedAt,
|
|
updatedAt: deletedAt,
|
|
updatedBy: auth.user?.id ?? existing.updatedBy,
|
|
version: existing.version + 1,
|
|
};
|
|
const operation: SyncOperation = {
|
|
id: createId(),
|
|
userId: auth.user?.id,
|
|
ledgerIds: deleted.ledgerIds,
|
|
entity: "entry",
|
|
entityId: deleted.id,
|
|
action: "delete",
|
|
payload: deleted,
|
|
createdAt: deletedAt,
|
|
syncedAt: null,
|
|
};
|
|
|
|
await db.transaction("rw", db.entries, db.syncOperations, async () => {
|
|
await db.entries.put(deleted);
|
|
await db.syncOperations.put(operation);
|
|
});
|
|
|
|
await this.refreshLocalEntries();
|
|
void this.syncEntries();
|
|
},
|
|
},
|
|
});
|