Курсор — это номер, а не время

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

// Журнал сервера тот же, что у миниатюры части V, но номера выдаёт база.
// Пачка — одна транзакция: её записи не видны никому, пока она не зафиксирована
function createServer() {
  const log: Entry[] = [];
  let sequence = 0;

  return {
    begin() {
      const pending: Entry[] = [];
      return {
        // insert into changes values (nextval('changes_seq'), …) — номер выдан здесь
        write(op: Op): number {
          sequence += 1;
          pending.push({ ...op, seq: sequence });
          return sequence;
        },
        // commit — записи становятся видны разом
        commit(): void {
          log.push(...pending);
        },
      };
    },
    // pull — чистое чтение: только зафиксированное, по возрастанию номера
    pull(cursor: number, limit: number) {
      const changes = log.filter((entry) => entry.seq > cursor).sort((a, b) => a.seq - b.seq).slice(0, limit);
      const next = changes.length > 0 ? changes[changes.length - 1].seq : cursor;
      return { changes, cursor: next, hasMore: log.some((entry) => entry.seq > next) };
    },
  };
}

const server = createServer();
const names = (changes: Entry[]): string => "[" + changes.map((entry) => entry.seq + ":" + entry.opId).join(", ") + "]";

// телефон и веб-версия отправили свои пачки почти одновременно
const phone = server.begin();
phone.write({ opId: "A:1", rowId: "A-1", fields: { exercise: "жим", weightKg: 80, reps: 5 }, at: 1 });
const web = server.begin();
web.write({ opId: "B:1", rowId: "B-1", fields: { exercise: "присед", weightKg: 100, reps: 5 }, at: 2 });
// пачка веб-версии короче, и фиксируется она первой
web.commit();

const first = server.pull(0, 100);
console.log("пока пачка телефона в пути:", names(first.changes), "| курсор", first.cursor);
phone.commit();
const second = server.pull(first.cursor, 100);
console.log("после фиксации телефона:", names(second.changes), "| курсор", second.cursor);
console.log("а в журнале сервера:", names(server.pull(0, 100).changes));

-- первые команды транзакции пачки: строка-счётчик пользователя
insert into sync_heads (user_id) values ($1) on conflict do nothing;
select seq from sync_heads where user_id = $1 for update;

-- номер очередного изменения, уже под взятой строкой
update sync_heads set seq = seq + 1 where user_id = $1 returning seq;

type Fields = Record<string, string | number>;
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Entry = Op & { seq: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };

// Тот же журнал, но номер берётся из строки-счётчика пользователя, а строку
// эту пачка занимает первой же командой
function createServer() {
  const log: (Entry & { user: string })[] = [];
  const heads = new Map<string, number>();
  const held = new Map<string, Promise<void>>();

  async function begin(user: string) {
    // первая команда пачки: select seq from sync_heads where user_id = $1 for update
    const previous = held.get(user);
    let release = (): void => {};
    held.set(user, new Promise<void>((resolve) => { release = resolve; }));
    await previous;
    // insert into sync_heads values ($1) on conflict do nothing: страховка для
    // первой пачки, если строку не завела регистрация
    if (!heads.has(user)) heads.set(user, 0);
    const pending: (Entry & { user: string })[] = [];
    return {
      // update sync_heads set seq = seq + 1 where user_id = $1 returning seq
      write(op: Op): number {
        const seq = (heads.get(user) ?? 0) + 1;
        heads.set(user, seq);
        pending.push({ ...op, seq, user });
        return seq;
      },
      commit(): void {
        log.push(...pending);
        release();
      },
    };
  }

  // номера у каждого свои, поэтому страница отбирается и по пользователю
  async function pull(user: string, cursor: number, limit: number) {
    const mine = log.filter((entry) => entry.user === user && entry.seq > cursor);
    const changes = mine.sort((a, b) => a.seq - b.seq).slice(0, limit);
    const next = changes.length > 0 ? changes[changes.length - 1].seq : cursor;
    return { changes, cursor: next, hasMore: mine.some((entry) => entry.seq > next) };
  }

  return {
    begin,
    pull,
    // соединение от имени пользователя: кто он, сервер знает из токена
    as(user: string) {
      return {
        async push(batch: Op[]): Promise<{ verdicts: Verdict[] }> {
          const tx = await begin(user);
          const verdicts: Verdict[] = batch.map((op) => {
            tx.write(op);
            return { opId: op.opId, status: "applied" };
          });
          tx.commit();
          return { verdicts };
        },
        pull: (cursor: number, limit: number) => pull(user, cursor, limit),
      };
    },
  };
}

const names = (changes: Entry[]): string => "[" + changes.map((entry) => entry.seq + ":" + entry.opId).join(", ") + "]";
// даём соседним запросам все поводы пройти вперёд: десять оборотов очереди микрозадач
const letOthersRun = async (): Promise<void> => {
  for (let i = 0; i < 10; i += 1) await Promise.resolve();
};

async function main(): Promise<void> {
  const server = createServer();
  const phone = await server.begin("u1");
  phone.write({ opId: "A:1", rowId: "A-1", fields: { exercise: "жим", weightKg: 80, reps: 5 }, at: 1 });

  // веб-версия того же человека отправляет свою пачку, не дожидаясь телефона
  let webNumber = 0;
  const webBatch = (async () => {
    const web = await server.begin("u1");
    webNumber = web.write({ opId: "B:1", rowId: "B-1", fields: { exercise: "присед", weightKg: 100, reps: 5 }, at: 2 });
    web.commit();
  })();

  await letOthersRun();
  console.log("номер у веб-версии:", webNumber === 0 ? "ещё не выдан" : webNumber);
  const first = await server.pull("u1", 0, 100);
  console.log("пока пачка телефона в пути:", names(first.changes), "| курсор", first.cursor);

  phone.commit();
  await webBatch;
  const second = await server.pull("u1", first.cursor, 100);
  console.log("после фиксации обеих:", names(second.changes), "| курсор", second.cursor);
}

main();

type Row = { seq: string; op_id: string };

// строки, как их отдал драйвер: bigint приезжает строкой
const rows: Row[] = [];
for (let n = 1; n <= 12; n += 1) rows.push({ seq: String(n), op_id: "A:" + n });

// страница и «ещё есть» посчитаны в JS — как их считает сервер, когда решает
// сам: по курсору клиента и по номерам строк
function page(cursor: string | number, limit: number) {
  const mine = rows.filter((row) => row.seq > cursor);
  const changes = mine.slice(0, limit);
  const last = changes.length > 0 ? changes[changes.length - 1].seq : cursor;
  return { changes, cursor: last, hasMore: mine.some((row) => row.seq > last) };
}

// то же, но номер приведён к числу на границе чтения
function pageNumbers(cursor: number, limit: number) {
  const mine = rows.map((row) => ({ seq: Number(row.seq), opId: row.op_id })).filter((row) => row.seq > cursor);
  const changes = mine.slice(0, limit);
  const last = changes.length > 0 ? changes[changes.length - 1].seq : cursor;
  return { changes, cursor: last, hasMore: mine.some((row) => row.seq > last) };
}

const asIs = page("9", 5);
console.log("курсор «9», номера как приехали:", asIs.changes.map((row) => row.seq).join(",") || "ничего");
console.log("  курсор стал:", JSON.stringify(asIs.cursor), "| ещё есть:", asIs.hasMore);
const fixed = pageNumbers(9, 5);
console.log("курсор 9, номера через Number:", fixed.changes.map((row) => row.seq).join(","));
console.log("  курсор стал:", JSON.stringify(fixed.cursor), "| ещё есть:", fixed.hasMore);

type Fields = Record<string, string | number>;
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Entry = Op & { seq: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
type Page = { changes: Entry[]; cursor: number; hasMore: boolean };
interface Connection {
  push(batch: Op[]): Promise<{ verdicts: Verdict[] }>;
  pull(cursor: number, limit: number): Promise<Page>;
}
interface Server {
  as(user: string): Connection;
}
interface Client {
  journal: { newId(): string; change(rowId: string, fields: Fields): void };
  sync(): Promise<Verdict[]>;
  readonly cursor: number;
}
declare function createServer(): Server;
declare function createClient(device: string, server: Connection): Client;

// Водяной знак: сервер держит при себе самые свежие записи — за номером свежей
// записи может прятаться транзакция в полёте с номером поменьше
type Mode = { name: string; settledUpTo: number; withShift: boolean };

function withSettleLag(connection: Connection, mode: Mode): Connection {
  let calls = 0;
  return {
    push: (batch) => connection.push(batch),
    async pull(cursor, limit) {
      calls += 1;
      if (calls > 50) throw new Error("pull вызван " + calls + " раз, курсор " + cursor);
      const page = await connection.pull(cursor, limit);
      const settled = page.changes.filter((entry) => entry.seq <= mode.settledUpTo);
      const last = settled.length > 0 ? settled[settled.length - 1].seq : cursor;
      const more = page.hasMore || settled.length < page.changes.length;
      return { changes: settled, cursor: last, hasMore: mode.withShift ? more && last > cursor : more };
    },
  };
}

const MODES: Mode[] = [
  { name: "«ещё есть» по всем строкам", settledUpTo: 0, withShift: false },
  { name: "«ещё есть» вместе со сдвигом", settledUpTo: 0, withShift: true },
  { name: "прятать нечего", settledUpTo: 99, withShift: false },
];

async function main(): Promise<void> {
  for (const mode of MODES) {
    const server = createServer();
    const phone = createClient("A", withSettleLag(server.as("u1"), mode));
    const set = phone.journal.newId();
    phone.journal.change(set, { exercise: "жим", weightKg: 80, reps: 5 });
    phone.journal.change(set, { weightKg: 82.5 });
    try {
      await phone.sync();
      console.log(mode.name + ": цикл кончился, курсор", phone.cursor);
    } catch (error) {
      console.log(mode.name + ":", (error as Error).message);
    }
  }
}

main();

// apps/api/src/sync/sync.repository.ts и sync.service.ts боевого проекта автора

// Запас времени, после которого изменение считается «устоявшимся»: за него
// курсор догона двигать безопасно. Должен быть НЕ МЕНЬШЕ удвоенной
// длительности самой долгой пишущей транзакции
const SETTLE_LAG_SECONDS = 5;
const isSettled = sql<boolean>`${messages.updatedAt} <= now() - ${settleLag}`;

// seq последней устоявшейся строки, а не +1: сравнение на сервере строгое (gt),
// так что повторный вызов с этим курсором начнёт ровно со следующей строки
const nextCursor = [...messagePage.page, ...chatPage.page].reduce<number | null>(
  (settled, row) =>
    row.settled && row.seq <= boundary && (settled === null || row.seq > settled)
      ? row.seq
      : settled,
  null,
);

return {
  items: messagePage.page.map(toMessageOutput),
  chats: chatPage.page.map(toSyncChatOutput),
  nextCursor,
  // «есть ещё И курсор сдвинулся»: страница целиком из неустоявшихся строк
  // подтверждать нечего, и клиент, крутящий цикл по hasMore, повторял бы её вечно
  hasMore: (messagePage.more || chatPage.more) && nextCursor !== null,
};

// apps/mobile/src/shared/api/catch-up.ts того же проекта
// nextCursor: null — это отказ подтверждать что-либо: ни одна строка страницы
// ещё не «устоялась». seq выдаётся в НАЧАЛЕ транзакции, а строка становится
// видимой при коммите, поэтому за seq свежей строки может прятаться транзакция
// в полёте с МЕНЬШИМ seq
for (let page = 0; page < port.maxPages; page += 1) {
  const result = await port.pull(cursor);
  for (const message of result.items) port.apply(message);
  for (const chat of result.chats) port.applyChat(chat);

  // подтверждать нечего — хвост доберёт следующий догон, курсор остаётся прежним
  if (result.nextCursor === null) return 'deferred';
  // курсор обязан расти (сервер отдаёт seq > cursor); страховка от вращения на месте
  if (result.nextCursor <= cursor) return 'deferred';

  cursor = result.nextCursor;
  await port.writeCursor(cursor);
  if (!result.hasMore) return 'done';
}
// страницы кончились по нашему потолку, а не по данным сервера
return 'capped';