708 lines
31 KiB
TypeScript
708 lines
31 KiB
TypeScript
import cookie from "@fastify/cookie";
|
|
import helmet from "@fastify/helmet";
|
|
import rateLimit from "@fastify/rate-limit";
|
|
import { ledgerThemeAccents, type LedgerEntry, type LedgerThemeId, type SyncOperation } from "@cents/domain";
|
|
import Fastify, { type FastifyReply, type FastifyRequest } from "fastify";
|
|
import { createUserWithPersonalLedger, PERSONAL_LEDGER_NAME } from "./accounts.js";
|
|
import { initializeDatabase, pool } from "./db.js";
|
|
import {
|
|
clearSessionCookie,
|
|
createId,
|
|
createSecret,
|
|
hashPassword,
|
|
hashSecret,
|
|
INVITATION_TTL_MS,
|
|
normalizeName,
|
|
SESSION_TTL_MS,
|
|
sessionCookieName,
|
|
setSessionCookie,
|
|
validPassword,
|
|
verifyPassword,
|
|
} from "./security.js";
|
|
|
|
type User = { id: string; name: string };
|
|
type MemberRole = "owner" | "member";
|
|
|
|
const server = Fastify({ logger: true, trustProxy: true });
|
|
await server.register(cookie);
|
|
await server.register(helmet, { contentSecurityPolicy: false });
|
|
await server.register(rateLimit, { max: 120, timeWindow: "1 minute" });
|
|
await initializeDatabase();
|
|
|
|
server.addHook("onRequest", async (request, reply) => {
|
|
if (!["POST", "PATCH", "PUT", "DELETE"].includes(request.method)) return;
|
|
const origin = request.headers.origin;
|
|
if (!origin) return;
|
|
try {
|
|
const configuredOrigin = process.env.PUBLIC_ORIGIN ? new URL(process.env.PUBLIC_ORIGIN).origin : null;
|
|
if (new URL(origin).host !== request.headers.host && new URL(origin).origin !== configuredOrigin) {
|
|
return reply.code(403).send({ error: "不允许跨站提交" });
|
|
}
|
|
} catch {
|
|
return reply.code(403).send({ error: "请求来源无效" });
|
|
}
|
|
});
|
|
|
|
server.addHook("onSend", async (request, reply, payload) => {
|
|
if (
|
|
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");
|
|
}
|
|
return payload;
|
|
});
|
|
|
|
function uniqueViolation(error: unknown) {
|
|
return typeof error === "object" && error !== null && "code" in error && error.code === "23505";
|
|
}
|
|
|
|
async function currentUser(request: FastifyRequest, reply?: FastifyReply): Promise<User | null> {
|
|
const token = request.cookies[sessionCookieName];
|
|
if (!token) return null;
|
|
const tokenHash = hashSecret(token);
|
|
const result = await pool.query<User>(
|
|
`SELECT u.id, u.name
|
|
FROM sessions s JOIN users u ON u.id = s.user_id
|
|
WHERE s.token_hash = $1 AND s.revoked_at IS NULL AND s.expires_at > now()`,
|
|
[tokenHash],
|
|
);
|
|
const user = result.rows[0];
|
|
if (!user) return null;
|
|
await pool.query(
|
|
"UPDATE sessions SET last_seen_at = now(), expires_at = $1 WHERE token_hash = $2",
|
|
[new Date(Date.now() + SESSION_TTL_MS), tokenHash],
|
|
);
|
|
if (reply) setSessionCookie(reply, token);
|
|
return user;
|
|
}
|
|
|
|
async function requireUser(request: FastifyRequest, reply: FastifyReply) {
|
|
const user = await currentUser(request, reply);
|
|
if (!user) {
|
|
clearSessionCookie(reply);
|
|
await reply.code(401).send({ error: "请先登录" });
|
|
return null;
|
|
}
|
|
return user;
|
|
}
|
|
|
|
async function ledgerRole(userId: string, ledgerId: string): Promise<MemberRole | null> {
|
|
const result = await pool.query<{ role: MemberRole }>(
|
|
`SELECT role FROM ledger_members
|
|
WHERE ledger_id = $1 AND user_id = $2 AND removed_at IS NULL`,
|
|
[ledgerId, userId],
|
|
);
|
|
return result.rows[0]?.role ?? null;
|
|
}
|
|
|
|
async function createSession(userId: string) {
|
|
const token = createSecret();
|
|
await pool.query(
|
|
"INSERT INTO sessions (id, user_id, token_hash, expires_at) VALUES ($1, $2, $3, $4)",
|
|
[createId(), userId, hashSecret(token), new Date(Date.now() + SESSION_TTL_MS)],
|
|
);
|
|
return token;
|
|
}
|
|
|
|
function invitationStatus(invitation: { expiresAt: Date | string; acceptedAt: Date | string | null; revokedAt: Date | string | null }) {
|
|
if (invitation.revokedAt) return "revoked";
|
|
if (invitation.acceptedAt) return "accepted";
|
|
if (new Date(invitation.expiresAt).getTime() <= Date.now()) return "expired";
|
|
return "valid";
|
|
}
|
|
|
|
const currencies = new Set(["CNY", "USD", "EUR", "JPY", "HKD"]);
|
|
|
|
function validLedgerTheme(value: unknown): value is LedgerThemeId {
|
|
return typeof value === "string" && Object.hasOwn(ledgerThemeAccents, value);
|
|
}
|
|
|
|
function ledgerThemeFromColor(color: unknown): LedgerThemeId {
|
|
if (typeof color !== "string") return "jade";
|
|
return (Object.entries(ledgerThemeAccents) as Array<[LedgerThemeId, string]>)
|
|
.find(([, accent]) => accent.toLowerCase() === color.toLowerCase())?.[0] ?? "jade";
|
|
}
|
|
|
|
function validDate(value: unknown) {
|
|
return typeof value === "string" && !Number.isNaN(Date.parse(value));
|
|
}
|
|
|
|
function validLedgerIds(value: unknown): value is string[] {
|
|
return Array.isArray(value) && value.length > 0 && value.length <= 20
|
|
&& value.every((ledgerId) => typeof ledgerId === "string" && ledgerId.length > 0 && ledgerId.length <= 100)
|
|
&& new Set(value).size === value.length;
|
|
}
|
|
|
|
function validEntry(value: unknown): value is LedgerEntry {
|
|
if (!value || typeof value !== "object") return false;
|
|
const entry = value as Partial<LedgerEntry>;
|
|
return typeof entry.id === "string" && entry.id.length > 0 && entry.id.length <= 100
|
|
&& typeof entry.ownerId === "string" && entry.ownerId.length > 0 && entry.ownerId.length <= 100
|
|
&& validLedgerIds(entry.ledgerIds)
|
|
&& (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<SyncOperation>;
|
|
const payload = operation.payload;
|
|
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(payload)
|
|
&& operation.entityId === payload.id
|
|
&& validLedgerIds(operation.ledgerIds)
|
|
&& operation.ledgerIds.length === payload.ledgerIds.length
|
|
&& operation.ledgerIds.every((ledgerId) => payload.ledgerIds.includes(ledgerId))
|
|
&& (operation.action !== "delete" || 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<SyncCursor>;
|
|
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) }));
|
|
|
|
server.post<{ Body: { name?: string; password?: string } }>(
|
|
"/api/auth/login",
|
|
{ config: { rateLimit: { max: 10, timeWindow: "1 minute" } } },
|
|
async (request, reply) => {
|
|
const nameKey = normalizeName(request.body.name ?? "");
|
|
const password = request.body.password ?? "";
|
|
const result = await pool.query<{ id: string; name: string; passwordHash: string | null }>(
|
|
`SELECT id, name, password_hash AS "passwordHash" FROM users WHERE name_key = $1`,
|
|
[nameKey],
|
|
);
|
|
const account = result.rows[0];
|
|
const valid = account?.passwordHash
|
|
? await verifyPassword(account.passwordHash, password).catch(() => false)
|
|
: (await hashPassword(password || "invalid-password"), false);
|
|
if (!account || !valid) return reply.code(401).send({ error: "姓名或密码错误" });
|
|
const token = await createSession(account.id);
|
|
setSessionCookie(reply, token);
|
|
return { user: { id: account.id, name: account.name } };
|
|
},
|
|
);
|
|
|
|
server.post("/api/auth/logout", async (request, reply) => {
|
|
const token = request.cookies[sessionCookieName];
|
|
if (token) await pool.query("UPDATE sessions SET revoked_at = now() WHERE token_hash = $1", [hashSecret(token)]);
|
|
clearSessionCookie(reply);
|
|
return { ok: true };
|
|
});
|
|
|
|
server.patch<{ Body: { name?: string } }>("/api/me", async (request, reply) => {
|
|
const user = await requireUser(request, reply);
|
|
if (!user) return;
|
|
const name = request.body.name?.trim() ?? "";
|
|
if (!name || name.length > 40) return reply.code(400).send({ error: "姓名应为 1 至 40 个字符" });
|
|
try {
|
|
const result = await pool.query<User>(
|
|
"UPDATE users SET name = $1, name_key = $2, updated_at = now() WHERE id = $3 RETURNING id, name",
|
|
[name, normalizeName(name), user.id],
|
|
);
|
|
return { user: result.rows[0] };
|
|
} catch (error) {
|
|
if (uniqueViolation(error)) return reply.code(409).send({ error: "该姓名已被使用" });
|
|
throw error;
|
|
}
|
|
});
|
|
|
|
server.get("/api/ledgers", async (request, reply) => {
|
|
const user = await requireUser(request, reply);
|
|
if (!user) return;
|
|
const result = await pool.query(
|
|
`SELECT l.id, l.name, l.color, l.theme, l.default_currency AS "defaultCurrency",
|
|
l.created_at AS "createdAt", l.updated_at AS "updatedAt",
|
|
l.archived_at AS "archivedAt", m.role,
|
|
(l.personal_owner_id IS NOT NULL) AS "isPersonal"
|
|
FROM ledgers l JOIN ledger_members m ON m.ledger_id = l.id
|
|
WHERE m.user_id = $1 AND m.removed_at IS NULL AND l.archived_at IS NULL
|
|
ORDER BY (l.personal_owner_id = $1) DESC, l.updated_at DESC`,
|
|
[user.id],
|
|
);
|
|
return { ledgers: result.rows };
|
|
});
|
|
|
|
server.post<{ Body: { name?: string; color?: string; theme?: string; defaultCurrency?: string } }>(
|
|
"/api/ledgers",
|
|
async (request, reply) => {
|
|
const user = await requireUser(request, reply);
|
|
if (!user) return;
|
|
const name = request.body.name?.trim() ?? "";
|
|
if (request.body.theme !== undefined && !validLedgerTheme(request.body.theme)) {
|
|
return reply.code(400).send({ error: "账本主题无效" });
|
|
}
|
|
const theme = request.body.theme ?? ledgerThemeFromColor(request.body.color);
|
|
const color = ledgerThemeAccents[theme];
|
|
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, theme, default_currency)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
RETURNING id, name, color, theme, default_currency AS "defaultCurrency",
|
|
created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt"`,
|
|
[ledgerId, name, color, theme, 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;
|
|
if (!await ledgerRole(user.id, request.params.ledgerId)) return reply.code(403).send({ error: "无权访问该账本" });
|
|
const result = await pool.query(
|
|
`SELECT u.id, u.name, m.role, m.joined_at AS "joinedAt"
|
|
FROM ledger_members m JOIN users u ON u.id = m.user_id
|
|
WHERE m.ledger_id = $1 AND m.removed_at IS NULL
|
|
ORDER BY CASE m.role WHEN 'owner' THEN 0 ELSE 1 END, m.joined_at`,
|
|
[request.params.ledgerId],
|
|
);
|
|
return { members: result.rows };
|
|
});
|
|
|
|
server.patch<{
|
|
Params: { ledgerId: string };
|
|
Body: { name?: string; color?: string; theme?: string; defaultCurrency?: string };
|
|
}>("/api/ledgers/:ledgerId", async (request, reply) => {
|
|
const user = await requireUser(request, reply);
|
|
if (!user) return;
|
|
if (await ledgerRole(user.id, request.params.ledgerId) !== "owner") {
|
|
return reply.code(403).send({ error: "只有账本拥有者可以修改设置" });
|
|
}
|
|
const personalLedger = await pool.query<{ isPersonal: boolean }>(
|
|
`SELECT (personal_owner_id IS NOT NULL) AS "isPersonal" FROM ledgers WHERE id = $1`,
|
|
[request.params.ledgerId],
|
|
);
|
|
const name = request.body.name?.trim() ?? "";
|
|
if (!name || name.length > 40) return reply.code(400).send({ error: "账本名称无效" });
|
|
if (request.body.theme !== undefined && !validLedgerTheme(request.body.theme)) {
|
|
return reply.code(400).send({ error: "账本主题无效" });
|
|
}
|
|
const theme = request.body.theme ?? ledgerThemeFromColor(request.body.color);
|
|
const color = ledgerThemeAccents[theme];
|
|
const result = await pool.query(
|
|
`UPDATE ledgers SET name = $1, color = $2, theme = $3, default_currency = $4, updated_at = now()
|
|
WHERE id = $5
|
|
RETURNING id, name, color, theme, default_currency AS "defaultCurrency",
|
|
created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt"`,
|
|
[personalLedger.rows[0]?.isPersonal ? PERSONAL_LEDGER_NAME : name, color, theme, request.body.defaultCurrency ?? "CNY", request.params.ledgerId],
|
|
);
|
|
return { ledger: result.rows[0] };
|
|
});
|
|
|
|
// Application invitations create accounts but never grant access to an existing ledger.
|
|
server.get<{ Params: { key: string } }>("/api/invitations/:key", async (request, reply) => {
|
|
const user = await currentUser(request, reply);
|
|
if (user) return { user, alreadyLoggedIn: true };
|
|
if (request.params.key.length < 32) return reply.code(404).send({ error: "邀请链接无效" });
|
|
const result = await pool.query(
|
|
`SELECT expires_at AS "expiresAt", accepted_at AS "acceptedAt", revoked_at AS "revokedAt"
|
|
FROM app_invitations WHERE key_hash = $1`,
|
|
[hashSecret(request.params.key)],
|
|
);
|
|
const invitation = result.rows[0];
|
|
if (!invitation) return reply.code(404).send({ error: "邀请链接无效" });
|
|
return { invitation: { ...invitation, status: invitationStatus(invitation) }, user: null };
|
|
});
|
|
|
|
server.post<{ Params: { key: string }; Body: { name?: string; password?: string } }>(
|
|
"/api/invitations/:key/accept",
|
|
{ config: { rateLimit: { max: 10, timeWindow: "1 minute" } } },
|
|
async (request, reply) => {
|
|
const existingUser = await currentUser(request, reply);
|
|
if (existingUser) return { user: existingUser, created: false };
|
|
const name = request.body.name?.trim() ?? "";
|
|
const password = request.body.password ?? "";
|
|
if (!name || name.length > 40) return reply.code(400).send({ error: "请填写 1 至 40 个字符的姓名" });
|
|
if (!validPassword(password)) return reply.code(400).send({ error: "密码应为 8 至 128 个字符" });
|
|
const client = await pool.connect();
|
|
try {
|
|
await client.query("BEGIN");
|
|
const result = await client.query<{
|
|
id: string; expiresAt: Date; acceptedAt: Date | null; revokedAt: Date | null;
|
|
}>(
|
|
`SELECT id, expires_at AS "expiresAt", accepted_at AS "acceptedAt", revoked_at AS "revokedAt"
|
|
FROM app_invitations WHERE key_hash = $1 FOR UPDATE`,
|
|
[hashSecret(request.params.key)],
|
|
);
|
|
const invitation = result.rows[0];
|
|
const status = invitation ? invitationStatus(invitation) : "missing";
|
|
if (status !== "valid") {
|
|
await client.query("ROLLBACK");
|
|
const code = status === "accepted" ? 409 : status === "missing" ? 404 : 410;
|
|
return reply.code(code).send({ error: status === "accepted" ? "邀请已被使用" : status === "expired" ? "邀请已过期" : status === "revoked" ? "邀请已撤销" : "邀请链接无效" });
|
|
}
|
|
|
|
const user = await createUserWithPersonalLedger(client, name, password);
|
|
await client.query(
|
|
"UPDATE app_invitations SET accepted_by = $1, accepted_at = now() WHERE id = $2",
|
|
[user.id, invitation!.id],
|
|
);
|
|
const sessionToken = createSecret();
|
|
await client.query(
|
|
"INSERT INTO sessions (id, user_id, token_hash, expires_at) VALUES ($1, $2, $3, $4)",
|
|
[createId(), user.id, hashSecret(sessionToken), new Date(Date.now() + SESSION_TTL_MS)],
|
|
);
|
|
await client.query("COMMIT");
|
|
setSessionCookie(reply, sessionToken);
|
|
return { user, created: true };
|
|
} catch (error) {
|
|
await client.query("ROLLBACK");
|
|
if (uniqueViolation(error)) return reply.code(409).send({ error: "该姓名已被使用" });
|
|
throw error;
|
|
} finally {
|
|
client.release();
|
|
}
|
|
},
|
|
);
|
|
|
|
server.post<{ Params: { ledgerId: string } }>("/api/ledgers/:ledgerId/invitations", async (request, reply) => {
|
|
const user = await requireUser(request, reply);
|
|
if (!user) return;
|
|
const personalLedger = await pool.query("SELECT 1 FROM ledgers WHERE id = $1 AND personal_owner_id IS NOT NULL", [request.params.ledgerId]);
|
|
if (personalLedger.rowCount) return reply.code(403).send({ error: "个人账本不能共享" });
|
|
if (await ledgerRole(user.id, request.params.ledgerId) !== "owner") {
|
|
return reply.code(403).send({ error: "只有账本拥有者可以邀请成员" });
|
|
}
|
|
const key = createSecret();
|
|
const expiresAt = new Date(Date.now() + INVITATION_TTL_MS);
|
|
await pool.query(
|
|
`INSERT INTO ledger_invitations (id, ledger_id, key_hash, role, created_by, expires_at)
|
|
VALUES ($1, $2, $3, 'member', $4, $5)`,
|
|
[createId(), request.params.ledgerId, hashSecret(key), user.id, expiresAt],
|
|
);
|
|
const origin = process.env.PUBLIC_ORIGIN ?? `${request.protocol}://${request.headers.host}`;
|
|
return { invitation: { key, url: `${origin}/join-ledger?key=${encodeURIComponent(key)}`, expiresAt } };
|
|
});
|
|
|
|
server.get<{ Params: { key: string } }>("/api/ledger-invitations/:key", async (request, reply) => {
|
|
const result = await pool.query(
|
|
`SELECT i.expires_at AS "expiresAt", i.accepted_at AS "acceptedAt", i.revoked_at AS "revokedAt",
|
|
l.id AS "ledgerId", l.name AS "ledgerName", u.name AS "inviterName"
|
|
FROM ledger_invitations i JOIN ledgers l ON l.id = i.ledger_id
|
|
LEFT JOIN users u ON u.id = i.created_by WHERE i.key_hash = $1`,
|
|
[hashSecret(request.params.key)],
|
|
);
|
|
const invitation = result.rows[0];
|
|
if (!invitation) return reply.code(404).send({ error: "账本邀请无效" });
|
|
return { invitation: { ...invitation, status: invitationStatus(invitation) } };
|
|
});
|
|
|
|
server.post<{ Params: { key: string } }>("/api/ledger-invitations/:key/accept", async (request, reply) => {
|
|
const user = await requireUser(request, reply);
|
|
if (!user) return;
|
|
const client = await pool.connect();
|
|
try {
|
|
await client.query("BEGIN");
|
|
const result = await client.query<{
|
|
id: string; ledgerId: string; role: MemberRole; expiresAt: Date; acceptedAt: Date | null; revokedAt: Date | null;
|
|
personalOwnerId: string | null;
|
|
}>(
|
|
`SELECT i.id, i.ledger_id AS "ledgerId", i.role, i.expires_at AS "expiresAt",
|
|
i.accepted_at AS "acceptedAt", i.revoked_at AS "revokedAt",
|
|
l.personal_owner_id AS "personalOwnerId"
|
|
FROM ledger_invitations i JOIN ledgers l ON l.id = i.ledger_id
|
|
WHERE i.key_hash = $1 FOR UPDATE OF i`,
|
|
[hashSecret(request.params.key)],
|
|
);
|
|
const invitation = result.rows[0];
|
|
if (invitation?.personalOwnerId) {
|
|
await client.query("ROLLBACK");
|
|
return reply.code(403).send({ error: "个人账本不能共享" });
|
|
}
|
|
const status = invitation ? invitationStatus(invitation) : "missing";
|
|
if (status !== "valid") {
|
|
await client.query("ROLLBACK");
|
|
return reply.code(status === "accepted" ? 409 : status === "missing" ? 404 : 410).send({ error: "账本邀请不可用" });
|
|
}
|
|
await client.query(
|
|
`INSERT INTO ledger_members (ledger_id, user_id, role) VALUES ($1, $2, $3)
|
|
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 e SET server_updated_at = now()
|
|
WHERE EXISTS (
|
|
SELECT 1 FROM entry_ledgers el
|
|
WHERE el.entry_id = e.id AND el.ledger_id = $1 AND el.unlinked_at IS NULL
|
|
)`,
|
|
[invitation!.ledgerId],
|
|
);
|
|
await client.query(
|
|
"UPDATE ledger_invitations SET accepted_by = $1, accepted_at = now() WHERE id = $2",
|
|
[user.id, invitation!.id],
|
|
);
|
|
await client.query("COMMIT");
|
|
return { ledgerId: invitation!.ledgerId };
|
|
} catch (error) {
|
|
await client.query("ROLLBACK");
|
|
throw error;
|
|
} finally {
|
|
client.release();
|
|
}
|
|
});
|
|
|
|
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 existing = await client.query<{ ownerId: string; canAccess: boolean }>(
|
|
`SELECT e.owner_id AS "ownerId",
|
|
(e.owner_id = $2 OR EXISTS (
|
|
SELECT 1 FROM entry_ledgers el
|
|
JOIN ledger_members m ON m.ledger_id = el.ledger_id
|
|
WHERE el.entry_id = e.id AND el.unlinked_at IS NULL
|
|
AND m.user_id = $2 AND m.removed_at IS NULL
|
|
)) AS "canAccess"
|
|
FROM entries e WHERE e.id = $1`,
|
|
[operation.entityId, user.id],
|
|
);
|
|
if (existing.rows[0] && !existing.rows[0].canAccess) {
|
|
await client.query("ROLLBACK");
|
|
return reply.code(403).send({ error: "无权修改该流水" });
|
|
}
|
|
const inaccessibleTargets = await client.query<{ count: number }>(
|
|
`SELECT count(*)::int AS count
|
|
FROM unnest($1::text[]) AS target(ledger_id)
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM ledger_members m
|
|
WHERE m.ledger_id = target.ledger_id AND m.user_id = $2 AND m.removed_at IS NULL
|
|
)
|
|
AND NOT ($3::boolean AND EXISTS (
|
|
SELECT 1 FROM entry_ledgers el
|
|
WHERE el.entry_id = $4 AND el.ledger_id = target.ledger_id AND el.unlinked_at IS NULL
|
|
))`,
|
|
[operation.ledgerIds, user.id, existing.rows[0]?.ownerId === user.id, operation.entityId],
|
|
);
|
|
if (inaccessibleTargets.rows[0]?.count) {
|
|
await client.query("ROLLBACK");
|
|
return reply.code(403).send({ error: "无权关联其中一个账本" });
|
|
}
|
|
|
|
const entryOwnerId = existing.rows[0]?.ownerId ?? user.id;
|
|
const personalLedger = await client.query<{ id: string }>(
|
|
"SELECT id FROM ledgers WHERE personal_owner_id = $1",
|
|
[entryOwnerId],
|
|
);
|
|
if (!personalLedger.rows[0]) throw new Error("用户缺少个人账本");
|
|
const targetLedgerIds = [...new Set([...operation.ledgerIds, personalLedger.rows[0].id])];
|
|
|
|
const entry = operation.payload;
|
|
const writeResult = await client.query(
|
|
`INSERT INTO entries (
|
|
id, owner_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
|
|
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, user.id, 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,
|
|
],
|
|
);
|
|
if (operation.action !== "delete" && writeResult.rowCount) {
|
|
if (existing.rows[0]?.ownerId === user.id) {
|
|
await client.query(
|
|
`UPDATE entry_ledgers SET unlinked_at = now(), updated_at = now()
|
|
WHERE entry_id = $1 AND unlinked_at IS NULL
|
|
AND NOT (ledger_id = ANY($2::text[]))`,
|
|
[entry.id, targetLedgerIds],
|
|
);
|
|
} else if (existing.rows[0]) {
|
|
await client.query(
|
|
`UPDATE entry_ledgers el SET unlinked_at = now(), updated_at = now()
|
|
FROM ledger_members m
|
|
WHERE el.entry_id = $1 AND el.unlinked_at IS NULL AND el.ledger_id = m.ledger_id
|
|
AND m.user_id = $2 AND m.removed_at IS NULL
|
|
AND NOT (el.ledger_id = ANY($3::text[]))`,
|
|
[entry.id, user.id, targetLedgerIds],
|
|
);
|
|
}
|
|
await client.query(
|
|
`INSERT INTO entry_ledgers (entry_id, ledger_id)
|
|
SELECT $1, unnest($2::text[])
|
|
ON CONFLICT (entry_id, ledger_id) DO UPDATE
|
|
SET unlinked_at = NULL, updated_at = now()`,
|
|
[entry.id, targetLedgerIds],
|
|
);
|
|
}
|
|
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<{ 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<LedgerEntry & { serverUpdatedAt: Date }>(
|
|
`SELECT e.id, e.owner_id AS "ownerId",
|
|
ARRAY(
|
|
SELECT el.ledger_id
|
|
FROM entry_ledgers el
|
|
LEFT JOIN ledger_members visible_member
|
|
ON visible_member.ledger_id = el.ledger_id
|
|
AND visible_member.user_id = $1
|
|
AND visible_member.removed_at IS NULL
|
|
WHERE el.entry_id = e.id
|
|
AND el.unlinked_at IS NULL
|
|
AND (e.owner_id = $1 OR visible_member.user_id IS NOT NULL)
|
|
ORDER BY el.ledger_id
|
|
) AS "ledgerIds",
|
|
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,
|
|
entry_change.changed_at AS "serverUpdatedAt"
|
|
FROM entries e
|
|
CROSS JOIN LATERAL (
|
|
SELECT GREATEST(
|
|
e.server_updated_at,
|
|
COALESCE(MAX(el.updated_at) FILTER (
|
|
WHERE e.owner_id = $1 OR change_member.user_id IS NOT NULL
|
|
), e.server_updated_at)
|
|
) AS changed_at
|
|
FROM entry_ledgers el
|
|
LEFT JOIN ledger_members change_member
|
|
ON change_member.ledger_id = el.ledger_id
|
|
AND change_member.user_id = $1
|
|
AND change_member.removed_at IS NULL
|
|
WHERE el.entry_id = e.id
|
|
) entry_change
|
|
WHERE (e.owner_id = $1 OR EXISTS (
|
|
SELECT 1 FROM entry_ledgers accessible_link
|
|
JOIN ledger_members accessible_member ON accessible_member.ledger_id = accessible_link.ledger_id
|
|
WHERE accessible_link.entry_id = e.id
|
|
AND accessible_member.user_id = $1
|
|
AND accessible_member.removed_at IS NULL
|
|
))
|
|
AND ($2::timestamptz IS NULL
|
|
OR entry_change.changed_at > $2::timestamptz
|
|
OR (entry_change.changed_at = $2::timestamptz AND e.id > $3))
|
|
ORDER BY entry_change.changed_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);
|
|
const host = process.env.HOST ?? "0.0.0.0";
|
|
await server.listen({ port, host });
|