sync.push и sync.pull в боевом API: контракт, доступ, старые клиенты

import 'reflect-metadata';
import { createHmac, timingSafeEqual } from 'node:crypto';
import { Module } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { FastifyAdapter, type NestFastifyApplication } from '@nestjs/platform-fastify';
import { initTRPC, TRPCError } from '@trpc/server';
import { fastifyTRPCPlugin, type FastifyTRPCPluginOptions } from '@trpc/server/adapters/fastify';
import { createTRPCClient, httpBatchLink, TRPCClientError } from '@trpc/client';
import { z } from 'zod';

const SECRET = 'учебный-секрет-сервера-не-для-прода';
const sign = (userId: string): string => {
  const body = Buffer.from(JSON.stringify({ sub: userId })).toString('base64url');
  return `${body}.${createHmac('sha256', SECRET).update(body).digest('base64url')}`;
};
// проверка подписи — до всякого похода в базу: см. главу про доступ
function subjectOf(token: string | undefined): string | null {
  const [body = '', signature = ''] = (token ?? '').split('.');
  const expected = Buffer.from(createHmac('sha256', SECRET).update(body).digest('base64url'));
  const received = Buffer.from(signature);
  if (received.length !== expected.length || !timingSafeEqual(received, expected)) return null;
  return JSON.parse(Buffer.from(body, 'base64url').toString()).sub as string;
}

type Ctx = { token: string | undefined };
const t = initTRPC.context<Ctx>().create();
const authed = t.procedure.use(({ ctx, next }) => {
  const userId = subjectOf(ctx.token);
  if (userId === null) throw new TRPCError({ code: 'UNAUTHORIZED' });
  return next({ ctx: { userId } });
});

const op = z.object({ opId: z.string(), rowId: z.string(), fields: z.record(z.string(), z.unknown()), at: z.number(), schema: z.number().optional() });
const appRouter = t.router({
  sync: t.router({
    push: authed.input(z.object({ batch: z.array(op) })).mutation(({ input, ctx }) => ({
      verdicts: input.batch.map((one) => ({ opId: one.opId, status: 'applied' as const, by: ctx.userId })),
    })),
    pull: authed.input(z.object({ cursor: z.number(), limit: z.number() })).query(({ input, ctx }) => ({
      changes: [], cursor: input.cursor, hasMore: false, by: ctx.userId,
    })),
  }),
});
type AppRouter = typeof appRouter;

@Module({})
class AppModule {}

const app = await NestFactory.create<NestFastifyApplication>(AppModule, new FastifyAdapter(), { logger: false });
const fastify = app.getHttpAdapter().getInstance();
const seen: string[] = [];
fastify.addHook('onRequest', (request, _reply, done) => { seen.push(`${request.method} ${request.url}`); done(); });
const options: FastifyTRPCPluginOptions<AppRouter> = {
  prefix: '/trpc',
  trpcOptions: { router: appRouter, createContext: ({ req }) => ({ token: (req.headers.authorization ?? '').replace(/^Bearer /, '') || undefined }) },
};
await fastify.register(fastifyTRPCPlugin, options);
await app.listen(0, '127.0.0.1');
const url = `${await app.getUrl()}/trpc`;

const client = (token?: string) => createTRPCClient<AppRouter>({
  links: [httpBatchLink({ url, headers: () => (token === undefined ? {} : { authorization: `Bearer ${token}` }) })],
});

// обёртка вызова: неудачи телефон различает по коду, а не по тексту
function fail(text: string, code: number): Error & { code: number } {
  return Object.assign(new Error(text), { code });
}
async function call<T>(work: () => Promise<T>): Promise<T> {
  try {
    return await work();
  } catch (error) {
    if (error instanceof TRPCClientError && error.data?.httpStatus === 401) throw fail('нужен вход', 401);
    throw fail('ответа нет', 0);
  }
}

const guest = client();
try {
  await guest.sync.pull.query({ cursor: 0, limit: 100 });
} catch (error) {
  const trpc = error as TRPCClientError<AppRouter>;
  console.log('без токена:', trpc.constructor.name, '| свои поля:', JSON.stringify(Object.keys(trpc)));
  console.log('  error.code:', JSON.stringify((trpc as unknown as { code?: number }).code),
    '| error.data.httpStatus:', JSON.stringify(trpc.data?.httpStatus),
    '| error.data.code:', JSON.stringify(trpc.data?.code));
}
try {
  await call(() => guest.sync.pull.query({ cursor: 0, limit: 100 }));
} catch (error) {
  const wrapped = error as Error & { code: number };
  console.log('через обёртку:', wrapped.message, '| code === 401:', wrapped.code === 401);
}

const phone = client(sign('u1'));
console.log('с токеном, pull:', JSON.stringify(await phone.sync.pull.query({ cursor: 0, limit: 100 })));
console.log('с токеном, push:', JSON.stringify(await phone.sync.push.mutate({ batch: [{ opId: 'A:1', rowId: 'A-1', fields: { reps: 5 }, at: 1 }] })));

const path = (line: string): string => line.split('?')[0] ?? line;
console.log('как ехали вызовы:', seen.map(path).join(' | '));
console.log('вход выдачи — в адресе:', decodeURIComponent(seen[0] ?? ''));
await app.close();

без токена: TRPCClientError | свои поля: ["cause","shape","data","meta","name"]
  error.code: undefined | error.data.httpStatus: 401 | error.data.code: "UNAUTHORIZED"
через обёртку: нужен вход | code === 401: true
с токеном, pull: {"changes":[],"cursor":0,"hasMore":false,"by":"u1"}
с токеном, push: {"verdicts":[{"opId":"A:1","status":"applied","by":"u1"}]}
как ехали вызовы: GET /trpc/sync.pull | GET /trpc/sync.pull | GET /trpc/sync.pull | POST /trpc/sync.push
вход выдачи — в адресе: GET /trpc/sync.pull?batch=1&input={"0":{"cursor":0,"limit":100}}

import 'reflect-metadata';
import { createHmac } from 'node:crypto';
import { Module } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { FastifyAdapter, type NestFastifyApplication } from '@nestjs/platform-fastify';
import { initTRPC, TRPCError } from '@trpc/server';
import { fastifyTRPCPlugin, type FastifyTRPCPluginOptions } from '@trpc/server/adapters/fastify';
import { createTRPCClient, httpBatchLink, TRPCClientError } from '@trpc/client';
import { z } from 'zod';

const SECRET = 'учебный-секрет-сервера-не-для-прода';
const sign = (userId: string): string => {
  const body = Buffer.from(JSON.stringify({ sub: userId })).toString('base64url');
  return `${body}.${createHmac('sha256', SECRET).update(body).digest('base64url')}`;
};
const subjectOf = (token: string | undefined): string | null => {
  const [body = '', signature = ''] = (token ?? '').split('.');
  if (createHmac('sha256', SECRET).update(body).digest('base64url') !== signature) return null;
  return JSON.parse(Buffer.from(body, 'base64url').toString()).sub as string;
};

/** Потолки на окно: у выдачи он выше, потому что первая синхронизация идёт страницами. */
const SYNC: Record<string, number> = { 'sync.pull': 600, 'sync.push': 120 };
/** Гостевой потолок автора — один на все гостевые процедуры. */
const GUEST: Record<string, number> = { 'sync.pull': 10, 'sync.push': 10 };
const WINDOW = 60_000;
const BLOCK = 60_000;
type Counter = { hits: number; expiresAt: number; blockedUntil: number };

/** Окно и пауза — той же формы, что у гостевого ограничителя; меняется только ключ. */
function createLimiter(keyOf: (path: string, subject: Subject) => string, ceiling: Record<string, number>) {
  const counters = new Map<string, Counter>();
  return (path: string, subject: Subject, now: number): boolean => {
    const key = keyOf(path, subject);
    const was = counters.get(key);
    const counter = was === undefined || now >= was.expiresAt
      ? { hits: 0, expiresAt: now + WINDOW, blockedUntil: 0 }
      : was;
    counters.set(key, counter);
    if (now < counter.blockedUntil) return false;
    counter.hits += 1;
    if (counter.hits <= (ceiling[path] ?? 10)) return true;
    counter.blockedUntil = now + BLOCK;
    return false;
  };
}

type Ctx = { token: string | undefined; ip: string };
type Subject = { userId: string | null; ip: string };
const t = initTRPC.context<Ctx>().create();

/** Сервер собирается дважды: одна и та же форма с ключом по пользователю и по адресу. */
function buildRouter(limiter: (path: string, subject: Subject, now: number) => boolean, counted: string[]) {
  const limited = t.procedure.use(({ ctx, path, next }) => {
    // личность — из подписи токена: она нужна ключу, а в базу за ней не ходим
    const userId = subjectOf(ctx.token);
    counted.push(path);
    if (!limiter(path, { userId, ip: ctx.ip }, Date.now())) throw new TRPCError({ code: 'TOO_MANY_REQUESTS' });
    if (userId === null) throw new TRPCError({ code: 'UNAUTHORIZED' });
    return next({ ctx: { userId } });
  });
  const op = z.object({ opId: z.string(), rowId: z.string(), fields: z.record(z.string(), z.unknown()), at: z.number() });
  return t.router({
    sync: t.router({
      push: limited.input(z.object({ batch: z.array(op) })).mutation(() => ({ verdicts: [] })),
      pull: limited.input(z.object({ cursor: z.number(), limit: z.number() })).query(({ input }) => ({
        changes: [], cursor: input.cursor, hasMore: false,
      })),
    }),
  });
}
type AppRouter = ReturnType<typeof buildRouter>;

@Module({})
class AppModule {}

async function start(keyOf: (path: string, subject: Subject) => string, ceiling: Record<string, number> = SYNC) {
  const counted: string[] = [];
  const requests: string[] = [];
  const router = buildRouter(createLimiter(keyOf, ceiling), counted);
  const app = await NestFactory.create<NestFastifyApplication>(AppModule, new FastifyAdapter(), { logger: false });
  const fastify = app.getHttpAdapter().getInstance();
  fastify.addHook('onRequest', (request, _reply, done) => { requests.push(`${request.method} ${request.url}`); done(); });
  const options: FastifyTRPCPluginOptions<AppRouter> = {
    prefix: '/trpc',
    trpcOptions: { router, createContext: ({ req }) => ({ token: (req.headers.authorization ?? '').replace(/^Bearer /, '') || undefined, ip: req.ip }) },
  };
  await fastify.register(fastifyTRPCPlugin, options);
  await app.listen(0, '127.0.0.1');
  const url = `${await app.getUrl()}/trpc`;
  const client = (userId: string) => createTRPCClient<AppRouter>({
    links: [httpBatchLink({ url, headers: () => ({ authorization: `Bearer ${sign(userId)}` }) })],
  });
  return { app, client, counted, requests };
}

/** Два ключа: авторизованных считаем по человеку, гостей — по адресу, другого признака нет. */
const byUserKey = (path: string, subject: Subject): string =>
  subject.userId === null ? `${path}:ip:${subject.ip}` : `${path}:u:${subject.userId}`;
const byIpKey = (path: string, subject: Subject): string => `${path}:ip:${subject.ip}`;

const refused = (error: unknown): string =>
  error instanceof TRPCClientError ? String(error.data?.code) : 'не отказ ограничителя';
const pull = (who: ReturnType<Awaited<ReturnType<typeof start>>['client']>) => who.sync.pull.query({ cursor: 0, limit: 100 });
const push = (who: ReturnType<Awaited<ReturnType<typeof start>>['client']>) => who.sync.push.mutate({ batch: [] });
async function tryMany(work: () => Promise<unknown>, times: number) {
  let passed = 0;
  let last = 'отказов не было';
  for (let i = 0; i < times; i += 1) {
    try {
      await work();
      passed += 1;
    } catch (error) {
      last = refused(error);
    }
  }
  return { passed, last };
}

// Сцена 1. Пачка tRPC: один HTTP-запрос — и сколько вызовов в счёте ограничителя
{
  const { app, client, counted, requests } = await start(byUserKey);
  const one = client('u1');
  await Promise.all([pull(one), pull(one), pull(one)]);
  console.log('пачка из трёх выдач: запросов', requests.length, '| в счёте ограничителя', counted.length);
  console.log('  адрес пачки:', decodeURIComponent((requests[0] ?? '').split('&input=')[0] ?? ''));
  await app.close();
}

// Сцена 2. Потолки процедур и второй пользователь за тем же ключом-адресом
{
  const { app, client } = await start(byUserKey);
  const one = client('u1');
  const two = client('u2');
  const pulls = await tryMany(() => pull(one), 600);
  console.log('\nвыдача, потолок 600: прошло', pulls.passed, '| последний отказ:', pulls.last);
  console.log('  601-я выдача:', await pull(one).then(() => 'прошла').catch(refused));
  const pushes = await tryMany(() => push(one), 120);
  console.log('приём, потолок 120: прошло', pushes.passed, '| последний отказ:', pushes.last);
  console.log('  121-й приём:', await push(one).then(() => 'прошёл').catch(refused));
  console.log('  первый приём второго человека:', await push(two).then(() => 'прошёл').catch(refused));
  await app.close();
}

// Сцена 3. Тот же ограничитель с гостевым ключом: зал за одним адресом
{
  const byIp = await start(byIpKey, GUEST);
  const byUser = await start(byUserKey);
  const hall = Array.from({ length: 30 }, (_, i) => `u${i + 1}`);
  const count = async (stand: Awaited<ReturnType<typeof start>>) => {
    let no = 0;
    for (const person of hall) {
      const who = stand.client(person);
      for (const work of [() => pull(who), () => push(who)]) {
        try {
          await work();
        } catch {
          no += 1;
        }
      }
    }
    return `отказов ${no} из ${hall.length * 2}`;
  };
  console.log('\nзал: 30 человек × (приём + выдача) за одним адресом');
  console.log('  гостевой ключ «процедура + адрес», потолок 10:', await count(byIp));
  console.log('  ключ «процедура + пользователь», потолки 600 и 120:', await count(byUser));
  await byIp.app.close();
  await byUser.app.close();
}

пачка из трёх выдач: запросов 1 | в счёте ограничителя 3
  адрес пачки: GET /trpc/sync.pull,sync.pull,sync.pull?batch=1

выдача, потолок 600: прошло 600 | последний отказ: отказов не было
  601-я выдача: TOO_MANY_REQUESTS
приём, потолок 120: прошло 120 | последний отказ: отказов не было
  121-й приём: TOO_MANY_REQUESTS
  первый приём второго человека: прошёл

зал: 30 человек × (приём + выдача) за одним адресом
  гостевой ключ «процедура + адрес», потолок 10: отказов 40 из 60
  ключ «процедура + пользователь», потолки 600 и 120: отказов 0 из 60

import 'reflect-metadata';
import { Module } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { FastifyAdapter, type NestFastifyApplication } from '@nestjs/platform-fastify';
import { initTRPC } from '@trpc/server';
import { fastifyTRPCPlugin, type FastifyTRPCPluginOptions } from '@trpc/server/adapters/fastify';
import { createTRPCClient, httpLink, TRPCClientError } from '@trpc/client';
import { z } from 'zod';

type Op = { opId: string; rowId: string; fields: Record<string, string | number>; at: number };

const t = initTRPC.create();
const op = z.object({ opId: z.string(), rowId: z.string(), fields: z.record(z.string(), z.unknown()), at: z.number() });
const appRouter = t.router({
  sync: t.router({
    // за один запрос разбирается не больше тысячи операций: остаток телефон досылает сам
    push: t.procedure.input(z.object({ batch: z.array(op) })).mutation(({ input }) => ({
      got: input.batch.length,
      verdicts: input.batch.slice(0, 1000).map((one) => ({ opId: one.opId, status: 'applied' as const })),
    })),
  }),
});
type AppRouter = typeof appRouter;

@Module({})
class AppModule {}

async function start(bodyLimit?: number) {
  const adapter = bodyLimit === undefined ? new FastifyAdapter() : new FastifyAdapter({ bodyLimit });
  const app = await NestFactory.create<NestFastifyApplication>(AppModule, adapter, { logger: false });
  const fastify = app.getHttpAdapter().getInstance();
  const options: FastifyTRPCPluginOptions<AppRouter> = { prefix: '/trpc', trpcOptions: { router: appRouter } };
  await fastify.register(fastifyTRPCPlugin, options);
  await app.listen(0, '127.0.0.1');
  const url = `${await app.getUrl()}/trpc`;
  const client = createTRPCClient<AppRouter>({ links: [httpLink({ url })] });
  return { app, client, url, limit: fastify.initialConfig.bodyLimit };
}

/*
 * Очередь дневника без сети. Имя устройства — UUID («Журнал: имена и номера
 * рождаются на телефоне»), поэтому имена операции и строки длинные: они и весят.
 */
const device = '7f3c2a9e-5b1d-4e8f-a6c0-2d9b8e1f4a37';
const YEAR = 6015;
const queue: Op[] = [];
for (let set = 1; queue.length < YEAR * 5; set += 1) {
  const rowId = `${device}-${set}`;
  const put = (fields: Op['fields'], at: number) =>
    queue.push({ opId: `${device}:${queue.length + 1}`, rowId, fields, at });
  put({ exercise: 'жим лежа', weightKg: 82.5, reps: 5 }, set * 2);
  put({ reps: 6 }, set * 2 + 1);
}
const bytes = (count: number): number => Buffer.byteLength(JSON.stringify({ batch: queue.slice(0, count) }));

const answer = async (client: Awaited<ReturnType<typeof start>>['client'], count: number) => {
  try {
    const out = await client.sync.push.mutate({ batch: queue.slice(0, count) });
    return `200, принято операций ${out.got}, вердиктов ${out.verdicts.length}`;
  } catch (error) {
    // печатаем только устойчивую часть: текст этой ошибки от захода к заходу разный
    const trpc = error instanceof TRPCClientError;
    return `отказ | ошибка tRPC: ${trpc} | data: ${trpc ? JSON.stringify((error as TRPCClientError<AppRouter>).data) : '—'}`;
  }
};

console.log('очередь за год без сети:', YEAR, 'операций, тело', bytes(YEAR), 'Б');

{
  const stand = await start();
  const limit = stand.limit ?? 0;
  console.log('\nпредел тела по умолчанию: у Fastify', limit, 'Б — запас', limit - bytes(YEAR), 'Б');
  console.log('  очередь за год:', await answer(stand.client, YEAR));
  console.log('  она же на повтор:', await answer(stand.client, YEAR));
  // где проходит граница: делим пополам, пока принятое и отвергнутое не сойдутся
  let ok = YEAR;
  let no = YEAR * 5;
  while (no - ok > 1) {
    const middle = Math.floor((ok + no) / 2);
    if ((await answer(stand.client, middle)).startsWith('200')) ok = middle;
    else no = middle;
  }
  console.log('  последняя принятая пачка:', ok, 'операций, тело', bytes(ok), 'Б');
  console.log('  первая непринятая:', no, 'операций, тело', bytes(no), 'Б');
  console.log('  что видит клиент:', await answer(stand.client, no));
  /*
   * Тот же запрос руками: код ответа и тело — чтобы видеть, кто отказал и почему.
   * Повтор здесь не украшение: тело больше предела, Fastify отвечает, не дочитав
   * его до конца, и сброс соединения успевает дойти до клиента раньше ответа —
   * тогда fetch бросает ECONNRESET или EPIPE (22 броска на 60 попытках). Сам
   * ответ, когда он доходит, от этого не меняется
   */
  const byHand = async () => {
    const body = JSON.stringify({ batch: queue.slice(0, no) });
    for (let attempt = 1; ; attempt += 1) {
      try {
        return await fetch(`${stand.url}/sync.push`, { method: 'POST', headers: { 'content-type': 'application/json' }, body });
      } catch (error) {
        if (attempt === 20) throw error;
      }
    }
  };
  const raw = await byHand();
  console.log('  тот же POST руками:', raw.status, await raw.text());
  await stand.app.close();
}
{
  const stand = await start(16 * 1024 * 1024);
  console.log('\nпредел поднят переменной окружения: у Fastify', stand.limit, 'Б');
  console.log('  очередь за пять лет:', YEAR * 5, 'операций, тело', bytes(YEAR * 5), 'Б');
  console.log('  ответ:', await answer(stand.client, YEAR * 5));
  await stand.app.close();
}

очередь за год без сети: 6015 операций, тело 942971 Б

предел тела по умолчанию: у Fastify 1048576 Б — запас 105605 Б
  очередь за год: 200, принято операций 6015, вердиктов 1000
  она же на повтор: 200, принято операций 6015, вердиктов 1000
  последняя принятая пачка: 6685 операций, тело 1048496 Б
  первая непринятая: 6686 операций, тело 1048631 Б
  что видит клиент: отказ | ошибка tRPC: true | data: undefined
  тот же POST руками: 413 {"statusCode":413,"code":"FST_ERR_CTP_BODY_TOO_LARGE","error":"Payload Too Large","message":"Request body is too large"}

предел поднят переменной окружения: у Fastify 16777216 Б
  очередь за пять лет: 30075 операций, тело 4782651 Б
  ответ: 200, принято операций 30075, вердиктов 1000

import { createHmac, timingSafeEqual } from 'node:crypto';
import { Inject, Injectable, Module } from '@nestjs/common';
import { initTRPC, TRPCError } from '@trpc/server';
import { z } from 'zod';
import pg from 'pg';
import type { Entry, Fields, Op, Verdict } from '../../_shared/p6-sync/client-v5.mjs';

/**
 * Выпуск сервера: какую схему он понимает и как поднимает до неё операции.
 * Нет поля schema — схема 1. Первый выпуск знает только её и ступенек не имеет.
 */
export const RELEASE: { latest: number; lift: (one: Op) => Op } = { latest: 1, lift: (one) => one };

/*
 * Контракт: файл части V плюс два необязательных поля — schema у операции и row
 * у вердикта. Ни device, ни reset, ни rowIds в нём нет.
 */
const fieldValue = z.union([z.string(), z.number(), z.boolean()]);
const op = z.object({
  opId: z.string().min(1),
  rowId: z.string().min(1),
  fields: z.record(z.string(), fieldValue),
  at: z.number().int(),
  schema: z.number().int().positive().optional(),
});
export const pushInput = z.object({ batch: z.array(op).min(1) });
export const pullInput = z.object({ cursor: z.number().int().min(0), limit: z.number().int().positive() });

const SECRET = 'учебный-секрет-сервера-не-для-прода';
/** Токен главы про доступ, урезанный до двух полей: кто и до какого часа. */
export const sign = (userId: string, secondsLeft: number): string => {
  const body = Buffer.from(JSON.stringify({ sub: userId, exp: Math.floor(Date.now() / 1000) + secondsLeft })).toString('base64url');
  return `${body}.${createHmac('sha256', SECRET).update(body).digest('base64url')}`;
};
/** Подпись и срок — и ни одного запроса в базу: личность отсюда идёт в ключ ограничителя. */
function subjectOf(token: string | undefined): string | null {
  const [body = '', signature = ''] = (token ?? '').split('.');
  const expected = Buffer.from(createHmac('sha256', SECRET).update(body).digest('base64url'));
  const received = Buffer.from(signature);
  if (received.length !== expected.length || !timingSafeEqual(received, expected)) return null;
  const claims = JSON.parse(Buffer.from(body, 'base64url').toString()) as { sub: string; exp: number };
  return Date.now() / 1000 >= claims.exp ? null : claims.sub;
}

/* ---- слияние по полям: то же правило, что у телефона, слово в слово ---- */
type Mark = { at: number; opId: string };
const newer = (a: Mark, b: Mark): boolean => (a.at !== b.at ? a.at > b.at : a.opId > b.opId);
function merge(fields: Fields | undefined, one: Op, marks: Record<string, Mark>): Fields {
  const next: Fields = { ...fields };
  for (const field of Object.keys(one.fields)) {
    const was = marks[field];
    if (was && !newer(one, was)) continue;
    next[field] = one.fields[field] as Fields[string];
    marks[field] = { at: one.at, opId: one.opId };
  }
  return next;
}
const checkRow = (fields: Fields): string | null => {
  const missing = ['exercise', 'weightKg', 'reps'].filter((field) => !(field in fields));
  if (missing.length > 0) return 'нет полей: ' + missing.join(', ');
  return Number(fields.reps) > 100 ? 'повторов больше сотни' : null;
};

/*
 * Ступенька схемы 1 → 2. Пока живы телефоны 1.0, она РАСШИРЯЮЩАЯ: старое поле
 * остаётся, новое появляется рядом. Идемпотентна — операция, уже несущая оба
 * поля, проходит через неё без изменений.
 */
export function expand(one: Op): Op {
  const fields = { ...one.fields };
  if ('reps' in fields && !('repsDone' in fields)) fields.repsDone = fields.reps as Fields[string];
  return { ...one, fields, schema: 2 };
}

type Stored = { rowId: string; fields: Fields; marks: Record<string, Mark>; seq: number };
/** Строка уходит записями по отметкам: поля с общей отметкой — одной записью. */
function entriesOf(stored: Stored): Entry[] {
  const byMark = new Map<string, Entry>();
  for (const field of Object.keys(stored.fields).sort()) {
    const mark = stored.marks[field] as Mark;
    const name = `${mark.at}|${mark.opId}`;
    const entry = byMark.get(name) ?? { opId: mark.opId, rowId: stored.rowId, fields: {}, at: mark.at, seq: stored.seq };
    entry.fields[field] = stored.fields[field] as Fields[string];
    byMark.set(name, entry);
  }
  return [...byMark.values()].sort((a, b) => a.at - b.at || (a.opId < b.opId ? -1 : 1));
}

export const POOL = Symbol('POOL');
/** За запрос разбирается не больше тысячи операций: остаток телефон досылает сам. */
const PER_REQUEST = 1000;
/** Больше пятисот строк за страницу сервер не отдаёт, о чём бы ни просил клиент. */
const PER_PAGE = 500;

@Injectable()
export class SyncService {
  constructor(@Inject(POOL) private readonly pool: pg.Pool) {}

  async push(userId: string, batch: Op[]): Promise<{ verdicts: Verdict[] }> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');
      await client.query('INSERT INTO sync_heads (user_id) VALUES ($1) ON CONFLICT DO NOTHING', [userId]);
      await client.query('SELECT seq FROM sync_heads WHERE user_id = $1 FOR UPDATE', [userId]);
      const verdicts: Verdict[] = [];
      const taken: Op[] = [];
      for (const raw of batch.slice(0, PER_REQUEST)) {
        // операция новее сервера: разбор пачки останавливается на ней
        if ((raw.schema ?? 1) > RELEASE.latest) break;
        const one = RELEASE.lift(raw);
        taken.push(one);
        const { rows: [stored] } = await client.query<Stored>(
          'SELECT fields, marks FROM sync_rows WHERE user_id = $1 AND row_id = $2', [userId, one.rowId]);
        const before = stored?.marks ?? {};
        const marks = { ...before };
        const fields = merge(stored?.fields, one, marks);
        const moved = JSON.stringify(marks) !== JSON.stringify(before);
        const reason = stored && !moved ? null : checkRow(fields);
        const computed: Verdict = reason
          ? { opId: one.opId, status: 'rejected', reason }
          : { opId: one.opId, status: 'applied' };
        const put = await client.query<{ status: string }>(
          `INSERT INTO sync_verdicts (user_id, op_id, status, reason) VALUES ($1, $2, $3, $4)
           ON CONFLICT DO NOTHING RETURNING status`,
          [userId, one.opId, computed.status, computed.reason ?? null]);
        if (put.rowCount === 0) {
          const { rows: [saved] } = await client.query<{ status: string; reason: string | null }>(
            'SELECT status, reason FROM sync_verdicts WHERE user_id = $1 AND op_id = $2', [userId, one.opId]);
          verdicts.push(saved.status === 'applied'
            ? { opId: one.opId, status: 'duplicate' }
            : { opId: one.opId, status: 'rejected', reason: saved.reason ?? undefined });
          continue;
        }
        if (computed.status === 'applied' && moved) {
          const { rows: [head] } = await client.query<{ seq: string }>(
            'UPDATE sync_heads SET seq = seq + 1 WHERE user_id = $1 RETURNING seq', [userId]);
          await client.query(
            `INSERT INTO sync_rows (user_id, row_id, fields, marks, seq) VALUES ($1, $2, $3, $4, $5)
             ON CONFLICT (user_id, row_id) DO UPDATE SET fields = excluded.fields, marks = excluded.marks, seq = excluded.seq`,
            [userId, one.rowId, fields, marks, Number(head.seq)]);
        }
        verdicts.push(computed);
      }
      // разобрать нечего — ошибка запроса, а не 200 с пустыми вердиктами:
      // на пустые вердикты при непустой очереди цикл телефона зовёт себя снова
      if (verdicts.length === 0) throw new TRPCError({ code: 'PRECONDITION_FAILED', message: 'операция новее сервера' });
      // строка целиком — после всей пачки: к этому времени она окончательная
      for (let i = 0; i < verdicts.length; i += 1) {
        if (verdicts[i].status !== 'rejected') continue;
        const { rows: [stored] } = await client.query<Stored>(
          'SELECT row_id, fields, marks, seq FROM sync_rows WHERE user_id = $1 AND row_id = $2',
          [userId, taken[i].rowId]);
        verdicts[i].row = stored === undefined ? [] : entriesOf({ ...stored, seq: Number(stored.seq) });
      }
      await client.query('COMMIT');
      return { verdicts };
    } catch (error) {
      await client.query('ROLLBACK').catch(() => {});
      throw error;
    } finally {
      client.release();
    }
  }

  async pull(userId: string, cursor: number, limit: number): Promise<{ changes: Entry[]; cursor: number; hasMore: boolean }> {
    // limit урезается кодом, а не отказом схемы: отдать меньше просимого законно
    const take = Math.min(limit, PER_PAGE);
    const { rows } = await this.pool.query<Stored>(
      `SELECT row_id AS "rowId", fields, marks, seq FROM sync_rows
       WHERE user_id = $1 AND seq > $2 ORDER BY seq LIMIT $3`,
      [userId, cursor, take + 1]);
    const page = rows.slice(0, take).map((row) => ({ ...row, seq: Number(row.seq) }));
    const last = page.length > 0 ? (page[page.length - 1] as Stored).seq : cursor;
    return { changes: page.flatMap(entriesOf), cursor: last, hasMore: rows.length > page.length };
  }
}

type Ctx = { token: string | undefined };
const t = initTRPC.context<Ctx>().create();
/** Синхронизация принимает только Bearer: куки-клиента у неё нет. */
const authed = t.procedure.use(({ ctx, next }) => {
  const userId = subjectOf(ctx.token);
  if (userId === null) throw new TRPCError({ code: 'UNAUTHORIZED' });
  return next({ ctx: { userId } });
});

export function buildRouter(sync: SyncService) {
  return t.router({
    sync: t.router({
      push: authed.input(pushInput).mutation(({ input, ctx }) => sync.push(ctx.userId, input.batch)),
      // выдача остаётся чистым чтением, поэтому Query: в @trpc/client это GET
      pull: authed.input(pullInput).query(({ input, ctx }) => sync.pull(ctx.userId, input.cursor, input.limit)),
    }),
  });
}
export type AppRouter = ReturnType<typeof buildRouter>;
export const contextOf = (headers: Record<string, string | string[] | undefined>): Ctx => ({
  token: String(headers.authorization ?? '').replace(/^Bearer /, '') || undefined,
});

@Module({
  providers: [
    SyncService,
    { provide: POOL, useFactory: () => new pg.Pool({ connectionString: process.env.DATABASE_URL }) },
  ],
  exports: [SyncService],
})
export class SyncModule {}

база: postgres://postgres@127.0.0.1:<порт-базы>/app
модуль синхронизации: http://127.0.0.1:<порт>/trpc

-- приёмка на двух устройствах
  зал: ответа нет
  после зала: на телефоне 2, на вебе 0 | ждут отправки 2 и 0
  дома, после первой синхронизации: на телефоне 2, на вебе 2 | ждут отправки 0 и 0
  после слияния: на телефоне 1, на вебе 1 | ждут отправки 0 и 0
  телефон: [{"id":"A-1","exercise":"жим","weightKg":82.5,"reps":6}]
  веб: [{"id":"A-1","exercise":"жим","reps":6,"weightKg":82.5}]
  эталон сошёлся: true | телефон и веб: по значениям true | строкой JSON false
  строк у сервера: 2 | счётчик: 5

-- токен протух
  синхронизация: нужен вход | code === 401: true | ждут отправки 2
  после входа: ждут отправки 0 | строк у сервера 2

-- порядок выпуска: кто первым
  сервер знает схему: 1
  операция 1.1 до выката сервера: PRECONDITION_FAILED (412)
  выкат сервера: знает схему 2 | ступенька расширяющая, reps остаётся
  что шлёт 1.0: {"opId":"D:1","rowId":"D-1","fields":{"exercise":"жим","weightKg":80,"reps":5},"at":1}
  вердикты 1.0: [{"opId":"D:1","status":"applied"}]
  что шлёт 1.1: {"opId":"E:1","rowId":"E-1","fields":{"exercise":"присед","weightKg":105,"reps":5,"repsDone":5},"at":10,"schema":2}
  вердикты 1.1: [{"opId":"E:1","status":"applied"}]
  поля строк у сервера: exercise+reps+repsDone+weightKg / exercise+reps+repsDone+weightKg
  лента 1.0 (читает reps): жим 5, присед 5
  лента 1.1 (читает repsDone ?? reps): жим 5, присед 5
  операция со schema 3 первой в пачке: PRECONDITION_FAILED (412)
  она же второй: вердиктов 1 из 2 — [{"opId":"F:2","status":"applied"}]