140 lines
13 KiB
TypeScript
140 lines
13 KiB
TypeScript
import { createHash } from 'node:crypto';
|
|
import type { Pool, PoolConnection } from 'mariadb';
|
|
import { customerSchema, deliverySchema, totalPrice, type Change, type Delivery, type Operation, type PullResponse } from '@model/shared';
|
|
|
|
export class SyncError extends Error {
|
|
constructor(public code: string, public statusCode: number, message: string, public current?: Change) { super(message); }
|
|
}
|
|
function json<T>(value: string | T): T { return typeof value === 'string' ? JSON.parse(value) as T : value; }
|
|
function requestHashes(operation: Operation): string[] {
|
|
const hashes = [createHash('sha256').update(JSON.stringify(operation)).digest('hex')];
|
|
const addCompatibleHash = (compatible: unknown) => {
|
|
const hash = createHash('sha256').update(JSON.stringify(compatible)).digest('hex');
|
|
if (!hashes.includes(hash)) hashes.push(hash);
|
|
};
|
|
// V2 clients did not send the optional profile/structured-address defaults added in V3.
|
|
// Accept only that exact old representation for an already-owned idempotency receipt.
|
|
if (operation.kind === 'customer.create' && !operation.data.contactName && !operation.data.phone && !operation.data.comments && Object.values(operation.data.deliveryWindows).every(value => value === null)) {
|
|
const legacy = { name: operation.data.name };
|
|
addCompatibleHash({ ...operation, data: legacy });
|
|
}
|
|
if (operation.kind === 'address.create' && !operation.data.city && !operation.data.street && !operation.data.buildingNumber) {
|
|
const legacy = { customerId: operation.data.customerId, label: operation.data.label, address: operation.data.address };
|
|
addCompatibleHash({ ...operation, data: legacy });
|
|
}
|
|
let timingCompatible: Operation | undefined;
|
|
if ((operation.kind === 'delivery.create' || operation.kind === 'delivery.update') && operation.data.deliverySlot == null && operation.data.specificTime == null && operation.data.deliveryWindow == null) {
|
|
const { deliverySlot: _deliverySlot, specificTime: _specificTime, deliveryWindow: _deliveryWindow, ...legacy } = operation.data;
|
|
timingCompatible = { ...operation, data: legacy } as Operation;
|
|
addCompatibleHash(timingCompatible);
|
|
}
|
|
if (operation.kind === 'delivery.create' && operation.data.scheduledDate === null) {
|
|
const { scheduledDate: _scheduledDate, ...legacy } = operation.data;
|
|
addCompatibleHash({ ...operation, data: legacy });
|
|
if (timingCompatible?.kind === 'delivery.create') {
|
|
const { scheduledDate: _legacyScheduledDate, ...oldest } = timingCompatible.data;
|
|
addCompatibleHash({ ...timingCompatible, data: oldest });
|
|
}
|
|
}
|
|
return hashes;
|
|
}
|
|
|
|
export async function appendChange(conn: PoolConnection, change: Change) {
|
|
// Caller holds sync_state FOR UPDATE through commit. Sequence order is commit order.
|
|
await conn.query('UPDATE sync_state SET sequence = sequence + 1 WHERE id = 1');
|
|
await conn.query('INSERT INTO sync_changes(sequence, entity_type, entity_id, change_json) SELECT sequence, ?, ?, ? FROM sync_state WHERE id = 1', [change.entityType, change.record.id, JSON.stringify(change)]);
|
|
}
|
|
|
|
export async function applyOperation(pool: Pool, operation: Operation, actorUserId: string | null = null) {
|
|
const [requestHash, ...compatibleHashes] = requestHashes(operation);
|
|
const conn = await pool.getConnection();
|
|
try {
|
|
await conn.beginTransaction();
|
|
const state = await conn.query('SELECT next_delivery_number FROM sync_state WHERE id = 1 FOR UPDATE');
|
|
if (!state.length) throw new Error('Database is not migrated');
|
|
const previous = await conn.query('SELECT request_hash, result_json, actor_user_id FROM sync_operations WHERE operation_id = ?', [operation.operationId]);
|
|
if (previous.length) {
|
|
if (previous[0].request_hash !== requestHash && !compatibleHashes.includes(previous[0].request_hash)) throw new SyncError('OPERATION_REUSED', 409, 'Operation ID has different content');
|
|
if (previous[0].actor_user_id && previous[0].actor_user_id !== actorUserId) throw new SyncError('OPERATION_OWNER_CHANGED', 409, 'Operation belongs to another account');
|
|
await conn.commit();
|
|
return { change: json<Change>(previous[0].result_json), replayed: true };
|
|
}
|
|
const entityType = operation.kind.split('.')[0]!;
|
|
const existing = await conn.query('SELECT change_json FROM sync_changes WHERE entity_type = ? AND entity_id = ? ORDER BY sequence DESC LIMIT 1', [entityType, operation.entityId]);
|
|
if (existing.length && operation.kind.endsWith('.create')) throw new SyncError('VERSION_CONFLICT', 409, 'Entity already exists', json<Change>(existing[0].change_json));
|
|
|
|
const meta = { id: operation.entityId, version: 1, createdAt: new Date().toISOString(), deletedAt: null };
|
|
let change: Change;
|
|
if (operation.kind === 'customer.create') {
|
|
const record = customerSchema.parse({ ...operation.data, ...meta, updatedAt: null });
|
|
const windows = record.deliveryWindows;
|
|
await conn.query('INSERT INTO customers(id, name, contact_name, phone, comments, morning_start, morning_end, afternoon_start, afternoon_end, late_night_start, late_night_end, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', [record.id, record.name, record.contactName, record.phone, record.comments, windows.morning?.start ?? null, windows.morning?.end ?? null, windows.afternoon?.start ?? null, windows.afternoon?.end ?? null, windows.lateNight?.start ?? null, windows.lateNight?.end ?? null, record.createdAt]);
|
|
change = { entityType: 'customer', record };
|
|
} else if (operation.kind === 'customer.update' || operation.kind === 'customer.remove') {
|
|
if (!existing.length) throw new SyncError('CUSTOMER_MISSING', 404, 'Customer is not available');
|
|
const current = customerSchema.parse(json<Change>(existing[0].change_json).record);
|
|
const currentChange: Change = { entityType: 'customer', record: current };
|
|
if (current.version !== operation.baseVersion) throw new SyncError('VERSION_CONFLICT', 409, 'Customer changed on another device', currentChange);
|
|
if (current.deletedAt) throw new SyncError('CUSTOMER_REMOVED', 409, 'Customer is archived', currentChange);
|
|
const record = customerSchema.parse(operation.kind === 'customer.update'
|
|
? { ...current, ...operation.data, version: current.version + 1, updatedAt: meta.createdAt }
|
|
: { ...current, version: current.version + 1, updatedAt: meta.createdAt, deletedAt: meta.createdAt });
|
|
const windows = record.deliveryWindows;
|
|
await conn.query('UPDATE customers SET name = ?, contact_name = ?, phone = ?, comments = ?, morning_start = ?, morning_end = ?, afternoon_start = ?, afternoon_end = ?, late_night_start = ?, late_night_end = ?, version = ?, updated_at = ?, deleted_at = ? WHERE id = ? AND version = ?', [record.name, record.contactName, record.phone, record.comments, windows.morning?.start ?? null, windows.morning?.end ?? null, windows.afternoon?.start ?? null, windows.afternoon?.end ?? null, windows.lateNight?.start ?? null, windows.lateNight?.end ?? null, record.version, record.updatedAt, record.deletedAt, record.id, operation.baseVersion]);
|
|
change = { entityType: 'customer', record };
|
|
} else if (operation.kind === 'address.create') {
|
|
const record = { ...operation.data, ...meta };
|
|
const customer = await conn.query('SELECT id FROM customers WHERE id = ? AND deleted_at IS NULL', [record.customerId]);
|
|
if (!customer.length) throw new SyncError('CUSTOMER_MISSING', 422, 'Customer is not available');
|
|
await conn.query('INSERT INTO customer_addresses(id, customer_id, label, address, city, street, building_number, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)', [record.id, record.customerId, record.label, record.address, record.city, record.street, record.buildingNumber, record.createdAt]);
|
|
change = { entityType: 'address', record };
|
|
} else if (operation.kind === 'delivery.create') {
|
|
const data = operation.data;
|
|
const address = await conn.query('SELECT a.id FROM customer_addresses a JOIN customers c ON c.id = a.customer_id WHERE a.id = ? AND a.customer_id = ? AND a.deleted_at IS NULL AND c.deleted_at IS NULL', [data.addressId, data.customerId]);
|
|
if (!address.length) throw new SyncError('ADDRESS_MISMATCH', 422, 'Select an active address assigned to this customer');
|
|
const types = await conn.query('SELECT version, shipment_unit_hundredths FROM delivery_types WHERE id = ? AND deleted_at IS NULL', [data.deliveryTypeId]);
|
|
if (!types.length || types[0].version !== data.deliveryTypeVersion || types[0].shipment_unit_hundredths !== data.shipmentUnitHundredths) {
|
|
throw new SyncError('CATALOG_CONFLICT', 409, 'Delivery type changed; the saved draft needs review');
|
|
}
|
|
const record = deliverySchema.parse({ ...data, ...meta, deliveryNumber: state[0].next_delivery_number as number, totalPriceAgorot: totalPrice(data.unitPriceAgorot, data.shipmentUnitHundredths), status: 'created' });
|
|
await conn.query('UPDATE sync_state SET next_delivery_number = next_delivery_number + 1 WHERE id = 1');
|
|
await conn.query('INSERT INTO deliveries(id, delivery_number, customer_id, address_id, scheduled_date, delivery_slot, specific_time, delivery_window_start, delivery_window_end, delivery_type_id, delivery_type_version, shipment_unit_hundredths, unit_price_agorot, total_price_agorot, notes, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', [record.id, record.deliveryNumber, record.customerId, record.addressId, record.scheduledDate, record.deliverySlot, record.specificTime, record.deliveryWindow?.start ?? null, record.deliveryWindow?.end ?? null, record.deliveryTypeId, record.deliveryTypeVersion, record.shipmentUnitHundredths, record.unitPriceAgorot, record.totalPriceAgorot, record.notes, record.createdAt]);
|
|
change = { entityType: 'delivery', record };
|
|
} else {
|
|
if (!existing.length) throw new SyncError('DELIVERY_MISSING', 404, 'Delivery is not available');
|
|
const saved = json<Change>(existing[0].change_json);
|
|
const current = deliverySchema.parse(saved.record);
|
|
const currentChange: Change = { entityType: 'delivery', record: current };
|
|
if (current.version !== operation.baseVersion) throw new SyncError('VERSION_CONFLICT', 409, 'Delivery changed on another device', currentChange);
|
|
if (current.status === 'cancelled' || current.deletedAt) throw new SyncError('DELIVERY_CANCELLED', 409, 'Cancelled deliveries cannot be changed', currentChange);
|
|
let record: Delivery = { ...current, version: current.version + 1, updatedAt: meta.createdAt };
|
|
if (operation.kind === 'delivery.update') {
|
|
const data = operation.data;
|
|
const address = await conn.query('SELECT a.id FROM customer_addresses a JOIN customers c ON c.id = a.customer_id WHERE a.id = ? AND a.customer_id = ? AND a.deleted_at IS NULL AND c.deleted_at IS NULL', [data.addressId, data.customerId]);
|
|
if (!address.length) throw new SyncError('ADDRESS_MISMATCH', 422, 'Select an active address assigned to this customer');
|
|
record = deliverySchema.parse({ ...record, ...data, scheduledDate: data.scheduledDate ?? current.scheduledDate });
|
|
} else record = { ...record, status: 'cancelled', cancelledAt: meta.createdAt, cancellationReason: operation.data.reason };
|
|
await conn.query('UPDATE deliveries SET customer_id = ?, address_id = ?, scheduled_date = ?, delivery_slot = ?, specific_time = ?, delivery_window_start = ?, delivery_window_end = ?, notes = ?, status = ?, version = ?, updated_at = ?, cancelled_at = ?, cancellation_reason = ? WHERE id = ? AND version = ?', [record.customerId, record.addressId, record.scheduledDate, record.deliverySlot, record.specificTime, record.deliveryWindow?.start ?? null, record.deliveryWindow?.end ?? null, record.notes, record.status, record.version, record.updatedAt, record.cancelledAt, record.cancellationReason, record.id, operation.baseVersion]);
|
|
change = { entityType: 'delivery', record };
|
|
}
|
|
await appendChange(conn, change);
|
|
await conn.query('INSERT INTO sync_operations(operation_id, device_id, request_hash, result_json, created_at, actor_user_id) VALUES (?, ?, ?, ?, ?, ?)', [operation.operationId, operation.deviceId, requestHash, JSON.stringify(change), meta.createdAt, actorUserId]);
|
|
await conn.commit();
|
|
return { change, replayed: false };
|
|
} catch (error) {
|
|
await conn.rollback();
|
|
throw error;
|
|
} finally { conn.release(); }
|
|
}
|
|
|
|
export async function pullChanges(pool: Pool, cursor: string, limit = 100): Promise<PullResponse> {
|
|
const conn = await pool.getConnection();
|
|
try {
|
|
const state = await conn.query('SELECT sequence FROM sync_state WHERE id = 1');
|
|
if (BigInt(cursor) > BigInt(state[0].sequence)) throw new SyncError('CURSOR_AHEAD', 409, 'Server history is older than this device; restore or reconcile before syncing');
|
|
const rows = await conn.query('SELECT sequence, change_json FROM sync_changes WHERE sequence > ? ORDER BY sequence LIMIT ?', [cursor, limit + 1]);
|
|
const changes: PullResponse['changes'] = rows.slice(0, limit).map((row: { sequence: bigint; change_json: string }) => ({ sequence: String(row.sequence), change: json<Change>(row.change_json) }));
|
|
return { changes, nextCursor: changes.at(-1)?.sequence ?? cursor, hasMore: rows.length > limit };
|
|
} finally { conn.release(); }
|
|
}
|