254 lines
9.5 KiB
TypeScript
254 lines
9.5 KiB
TypeScript
import type { RowDataPacket } from "mysql2/promise";
|
|
import { incomingWins } from "./conflict";
|
|
import { getPool, withTransaction } from "./db";
|
|
import { isRowId } from "./ids";
|
|
import { SYNC_TABLES, SYNC_TABLE_NAMES, type Column, type TableSpec } from "./schema";
|
|
import { quoteIdent } from "./sql";
|
|
import { isoToSql, sqlDateToIso } from "./time";
|
|
|
|
export class SyncInputError extends Error {}
|
|
|
|
export type PushResult = {
|
|
accepted: Record<string, string[]>;
|
|
rejected: Record<string, string[]>;
|
|
};
|
|
|
|
export type Cursor = { updatedAt: string; id: string };
|
|
|
|
export type PullResult = {
|
|
tables: Record<string, Record<string, unknown>[]>;
|
|
nextCursors: Record<string, Cursor>;
|
|
};
|
|
|
|
export const EPOCH_ISO = "1970-01-01T00:00:00.000Z";
|
|
export const PULL_LIMIT = 500;
|
|
|
|
type SqlValue = string | number | null;
|
|
|
|
export type PreparedRow = {
|
|
id: string;
|
|
updatedAt: Date;
|
|
columns: string[];
|
|
values: SqlValue[];
|
|
};
|
|
|
|
export function prepareRow(
|
|
table: TableSpec,
|
|
raw: unknown,
|
|
): { ok: true; row: PreparedRow } | { ok: false; reason: "skip" } | { ok: false; reason: "reject"; id: string } {
|
|
if (!raw || typeof raw !== "object" || Array.isArray(raw)) return { ok: false, reason: "skip" };
|
|
const record = raw as Record<string, unknown>;
|
|
if (typeof record.id !== "string" || record.id.length > 36) return { ok: false, reason: "skip" };
|
|
if (!isRowId(record.id)) return { ok: false, reason: "reject", id: record.id };
|
|
|
|
const columns: string[] = [];
|
|
const values: SqlValue[] = [];
|
|
for (const column of table.columns) {
|
|
if (column.name === "id" || column.name === "is_dirty") continue;
|
|
if (!Object.prototype.hasOwnProperty.call(record, column.name)) {
|
|
if (column.nullable) {
|
|
columns.push(column.name);
|
|
values.push(null);
|
|
continue;
|
|
}
|
|
if (column.hasDefault) continue;
|
|
return { ok: false, reason: "reject", id: record.id };
|
|
}
|
|
const coerced = coerce(column, record[column.name]);
|
|
if (!coerced.ok) return { ok: false, reason: "reject", id: record.id };
|
|
columns.push(column.name);
|
|
values.push(coerced.value);
|
|
}
|
|
|
|
const updatedIndex = columns.indexOf("updated_at");
|
|
if (updatedIndex < 0 || typeof values[updatedIndex] !== "string") {
|
|
return { ok: false, reason: "reject", id: record.id };
|
|
}
|
|
const updatedAt = new Date(`${(values[updatedIndex] as string).replace(" ", "T")}Z`);
|
|
if (Number.isNaN(updatedAt.getTime())) return { ok: false, reason: "reject", id: record.id };
|
|
return { ok: true, row: { id: record.id, updatedAt, columns, values } };
|
|
}
|
|
|
|
function coerce(column: Column, value: unknown): { ok: true; value: SqlValue } | { ok: false } {
|
|
if (value === null) {
|
|
if (column.nullable) return { ok: true, value: null };
|
|
return { ok: false };
|
|
}
|
|
switch (column.type) {
|
|
case "uuid": {
|
|
if (typeof value !== "string" || !isRowId(value)) return { ok: false };
|
|
return { ok: true, value };
|
|
}
|
|
case "string":
|
|
case "longtext": {
|
|
let text: string;
|
|
if (typeof value === "string") text = value;
|
|
else if (column.type === "longtext" && value && typeof value === "object") text = JSON.stringify(value);
|
|
else return { ok: false };
|
|
if (column.max !== undefined && text.length > column.max) return { ok: false };
|
|
return { ok: true, value: text };
|
|
}
|
|
case "bool": {
|
|
if (value === true || value === 1) return { ok: true, value: 1 };
|
|
if (value === false || value === 0) return { ok: true, value: 0 };
|
|
return { ok: false };
|
|
}
|
|
case "double": {
|
|
if (typeof value !== "number" || !Number.isFinite(value)) return { ok: false };
|
|
return { ok: true, value };
|
|
}
|
|
case "int": {
|
|
if (typeof value !== "number" || !Number.isInteger(value)) return { ok: false };
|
|
return { ok: true, value };
|
|
}
|
|
case "datetime": {
|
|
if (typeof value !== "string") return { ok: false };
|
|
const sql = isoToSql(value);
|
|
if (!sql) return { ok: false };
|
|
return { ok: true, value: sql };
|
|
}
|
|
default:
|
|
return { ok: false };
|
|
}
|
|
}
|
|
|
|
export async function pushChanges(tables: unknown): Promise<PushResult> {
|
|
// v1 does not allocate invoice numbers. shop_settings.next_invoice_number is
|
|
// last-write-wins like every other column. Two tablets incrementing it will clash.
|
|
if (!tables || typeof tables !== "object" || Array.isArray(tables)) {
|
|
throw new SyncInputError("invalid body");
|
|
}
|
|
const accepted: Record<string, string[]> = {};
|
|
const rejected: Record<string, string[]> = {};
|
|
|
|
await withTransaction(async (conn) => {
|
|
for (const [name, rows] of Object.entries(tables as Record<string, unknown>)) {
|
|
const table = SYNC_TABLES[name];
|
|
if (!table || !Array.isArray(rows)) continue;
|
|
accepted[name] = [];
|
|
rejected[name] = [];
|
|
for (const raw of rows) {
|
|
const prepared = prepareRow(table, raw);
|
|
if (!prepared.ok) {
|
|
if (prepared.reason === "reject") rejected[name].push(prepared.id);
|
|
continue;
|
|
}
|
|
const [found] = await conn.query<RowDataPacket[]>(
|
|
`SELECT updated_at FROM ${quoteIdent(table.name)} WHERE id = ? FOR UPDATE`,
|
|
[prepared.row.id],
|
|
);
|
|
const existing = found[0];
|
|
if (existing) {
|
|
const serverIso = sqlDateToIso(existing.updated_at);
|
|
const serverDate = serverIso ? new Date(serverIso) : null;
|
|
if (serverDate && !incomingWins(serverDate, prepared.row.updatedAt)) {
|
|
rejected[name].push(prepared.row.id);
|
|
continue;
|
|
}
|
|
const sets = prepared.row.columns.map((column) => `${quoteIdent(column)} = ?`);
|
|
sets.push(`${quoteIdent("is_dirty")} = 0`);
|
|
await conn.query(
|
|
`UPDATE ${quoteIdent(table.name)} SET ${sets.join(", ")} WHERE id = ?`,
|
|
[...prepared.row.values, prepared.row.id],
|
|
);
|
|
} else {
|
|
const insertCols = ["id", ...prepared.row.columns, "is_dirty"];
|
|
const insertVals: SqlValue[] = [prepared.row.id, ...prepared.row.values, 0];
|
|
await conn.query(
|
|
`INSERT INTO ${quoteIdent(table.name)} (${insertCols.map((column) => quoteIdent(column)).join(", ")}) VALUES (${insertCols.map(() => "?").join(", ")})`,
|
|
insertVals,
|
|
);
|
|
}
|
|
accepted[name].push(prepared.row.id);
|
|
}
|
|
}
|
|
});
|
|
|
|
return { accepted, rejected };
|
|
}
|
|
|
|
export function resolveCursors(input: unknown): {
|
|
cursors: Record<string, { updatedAt: Date; id: string }>;
|
|
error?: string;
|
|
} {
|
|
const source = input === undefined ? {} : input;
|
|
if (!source || typeof source !== "object" || Array.isArray(source)) {
|
|
return { cursors: {}, error: "invalid cursors" };
|
|
}
|
|
const raw = source as Record<string, unknown>;
|
|
const cursors: Record<string, { updatedAt: Date; id: string }> = {};
|
|
for (const name of SYNC_TABLE_NAMES) {
|
|
if (!Object.prototype.hasOwnProperty.call(raw, name)) {
|
|
cursors[name] = { updatedAt: new Date(EPOCH_ISO), id: "" };
|
|
continue;
|
|
}
|
|
const parsed = parseCursor(raw[name]);
|
|
if (!parsed) return { cursors: {}, error: "invalid cursor" };
|
|
cursors[name] = parsed;
|
|
}
|
|
return { cursors };
|
|
}
|
|
|
|
function parseCursor(value: unknown): { updatedAt: Date; id: string } | null {
|
|
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
|
|
const record = value as Record<string, unknown>;
|
|
if (typeof record.updatedAt !== "string") return null;
|
|
const updatedAt = new Date(record.updatedAt);
|
|
if (Number.isNaN(updatedAt.getTime())) return null;
|
|
const id = record.id === undefined ? "" : record.id;
|
|
if (typeof id !== "string") return null;
|
|
return { updatedAt, id };
|
|
}
|
|
|
|
function serializeValue(column: Column, value: unknown): unknown {
|
|
if (value === null || value === undefined) return null;
|
|
if (column.type === "datetime") return sqlDateToIso(value);
|
|
if (column.type === "bool") {
|
|
if (column.name === "is_dirty") return 0;
|
|
return value === true || value === 1 || value === "1";
|
|
}
|
|
if (column.type === "double" || column.type === "int") {
|
|
return typeof value === "number" ? value : Number(value);
|
|
}
|
|
if (Buffer.isBuffer(value)) return value.toString("utf8");
|
|
return String(value);
|
|
}
|
|
|
|
export async function pullChanges(cursorsInput: unknown): Promise<PullResult> {
|
|
const resolved = resolveCursors(cursorsInput);
|
|
if (resolved.error) throw new SyncInputError(resolved.error);
|
|
|
|
const pool = getPool();
|
|
const tables: Record<string, Record<string, unknown>[]> = {};
|
|
const nextCursors: Record<string, Cursor> = {};
|
|
|
|
for (const name of SYNC_TABLE_NAMES) {
|
|
const table = SYNC_TABLES[name];
|
|
const cursor = resolved.cursors[name];
|
|
const updatedSql = isoToSql(cursor.updatedAt.toISOString());
|
|
if (!updatedSql) throw new SyncInputError("invalid cursor");
|
|
const selectList = table.columns.map((column) => quoteIdent(column.name)).join(", ");
|
|
const [rows] = await pool.query<RowDataPacket[]>(
|
|
`SELECT ${selectList} FROM ${quoteIdent(name)}
|
|
WHERE updated_at > ? OR (updated_at = ? AND id > ?)
|
|
ORDER BY updated_at ASC, id ASC
|
|
LIMIT ${PULL_LIMIT}`,
|
|
[updatedSql, updatedSql, cursor.id],
|
|
);
|
|
const serialized = rows.map((row) => {
|
|
const out: Record<string, unknown> = {};
|
|
for (const column of table.columns) out[column.name] = serializeValue(column, row[column.name]);
|
|
return out;
|
|
});
|
|
tables[name] = serialized;
|
|
const last = serialized[serialized.length - 1];
|
|
if (last && typeof last.updated_at === "string" && typeof last.id === "string") {
|
|
nextCursors[name] = { updatedAt: last.updated_at, id: last.id };
|
|
} else {
|
|
nextCursors[name] = { updatedAt: cursor.updatedAt.toISOString(), id: cursor.id };
|
|
}
|
|
}
|
|
|
|
return { tables, nextCursors };
|
|
}
|