Цикл синхронизации: отправить, забрать, не задвоить

type Fields = Record<string, string | number>;
type Row = Fields & { id: string };
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
interface Store {
  write(row: Row): void;
  get(id: string): Row | undefined;
  list(): Row[];
}
interface Journal {
  newId(): string;
  change(rowId: string, fields: Fields): void;
  outbox(): Op[];
}
interface Server {
  rows: Map<string, Row>;
  journal: (Op & { seq: number })[];
  net: { offline: boolean; loseResponse: boolean };
  push(batch: Op[]): { verdicts: Verdict[] };
  pull(cursor: number, limit: number): { changes: (Op & { seq: number })[]; cursor: number; hasMore: boolean };
}
declare function createStore(): Store;
declare function createJournal(device: string, store: Store): Journal;
declare function createServer(): Server;

function createClient(device: string, server: Server) {
  const store = createStore();
  const journal = createJournal(device, store);
  const sent = new Set<string>();
  let cursor = 0;
  // в очереди то, на что ещё не пришёл вердикт
  const queue = () => journal.outbox().filter((op) => !sent.has(op.opId));
  return {
    store,
    journal,
    queue,
    sync() {
      const batch = queue();
      if (batch.length > 0) server.push(batch).verdicts.forEach((v) => sent.add(v.opId));
      let hasMore = true;
      while (hasMore) {
        const page = server.pull(cursor, 100);
        // пришедшее записывается тем же способом, каким пишет экран
        page.changes.forEach((entry) => journal.change(entry.rowId, entry.fields));
        cursor = page.cursor;
        hasMore = page.hasMore;
      }
    },
  };
}

const server = createServer();
const phone = createClient("A", server);
const web = createClient("B", server);
const report = (who: string, c: { journal: Journal; queue(): Op[] }) =>
  console.log(who + ": в журнале " + c.journal.outbox().length + ", в очереди " + c.queue().length);

const set = phone.journal.newId();
phone.journal.change(set, { exercise: "жим", weightKg: 80, reps: 5 });
phone.journal.change(set, { weightKg: 82.5 });
phone.sync();
report("телефон после первой", phone);
web.sync();
console.log("на вебе:", JSON.stringify(web.store.list()));
phone.sync();
web.sync();
report("телефон после второй", phone);
report("веб после второй", web);
console.log("у сервера записей:", server.journal.length, "| строк:", server.rows.size);

type Fields = Record<string, string | number>;
type Row = Fields & { id: string };
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
interface Store {
  write(row: Row): void;
  get(id: string): Row | undefined;
  list(): Row[];
}
interface Journal {
  newId(): string;
  change(rowId: string, fields: Fields): void;
  outbox(): Op[];
}
interface Server {
  rows: Map<string, Row>;
  journal: (Op & { seq: number })[];
  net: { offline: boolean; loseResponse: boolean };
  push(batch: Op[]): { verdicts: Verdict[] };
  pull(cursor: number, limit: number): { changes: (Op & { seq: number })[]; cursor: number; hasMore: boolean };
}
declare function createStore(): Store;
declare function createJournal(device: string, store: Store): Journal;
declare function createServer(): Server;
declare function apply(row: Row | undefined, op: Op): Row;

function createClient(device: string, server: Server) {
  const store = createStore();
  const journal = createJournal(device, store);
  const sent = new Set<string>();
  let cursor = 0;
  const queue = () => journal.outbox().filter((op) => !sent.has(op.opId));
  return {
    store,
    journal,
    queue,
    sync() {
      const batch = queue();
      if (batch.length > 0) server.push(batch).verdicts.forEach((v) => sent.add(v.opId));
      let hasMore = true;
      while (hasMore) {
        const page = server.pull(cursor, 100);
        for (const entry of page.changes) {
          // эхо: opId начинается с имени своего устройства — эта операция уже применена
          if (entry.opId.startsWith(device + ":")) continue;
          // чужое ложится в хранилище мимо журнала и теми же полями, что на сервере
          store.write(apply(store.get(entry.rowId), entry));
        }
        cursor = page.cursor;
        hasMore = page.hasMore;
      }
    },
  };
}

const server = createServer();
const phone = createClient("A", server);
const web = createClient("B", server);
const report = (who: string, c: { journal: Journal; queue(): Op[] }) =>
  console.log(who + ": в журнале " + c.journal.outbox().length + ", в очереди " + c.queue().length);

const set = phone.journal.newId();
phone.journal.change(set, { exercise: "жим", weightKg: 80, reps: 5 });
phone.journal.change(set, { weightKg: 82.5 });
phone.sync();
web.sync();
phone.sync();
web.sync();
report("телефон", phone);
report("веб", web);
console.log("на вебе:", JSON.stringify(web.store.list()));
console.log("у сервера записей:", server.journal.length, "| строк:", server.rows.size);
phone.journal.change(set, { reps: 6 });
report("телефон после новой записи", phone);

type Fields = Record<string, string | number>;
type Row = Fields & { id: string };
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
interface Store {
  write(row: Row): void;
  get(id: string): Row | undefined;
  list(): Row[];
}
interface Journal {
  newId(): string;
  change(rowId: string, fields: Fields): void;
  outbox(): Op[];
}
interface Server {
  rows: Map<string, Row>;
  journal: (Op & { seq: number })[];
  net: { offline: boolean; loseResponse: boolean };
  push(batch: Op[]): { verdicts: Verdict[] };
  pull(cursor: number, limit: number): { changes: (Op & { seq: number })[]; cursor: number; hasMore: boolean };
}
declare function createStore(): Store;
declare function createJournal(device: string, store: Store): Journal;
declare function createServer(): Server;
declare function apply(row: Row | undefined, op: Op): Row;
interface Client {
  store: Store;
  journal: Journal;
  queue(): Op[];
  readonly cursor: number;
  sync(): void;
}
declare function createClient(device: string, server: Server): Client;

const server = createServer();
const web = createClient("B", server);
const first = web.journal.newId();
web.journal.change(first, { exercise: "жим", weightKg: 80, reps: 5 });
const second = web.journal.newId();
web.journal.change(second, { exercise: "присед", weightKg: 100, reps: 5 });
web.sync();

// «процесс упал» изображает исключение: посреди применения страницы
let crashAt = "";
function applyChange(store: Store, entry: Op) {
  if (entry.opId === crashAt) {
    crashAt = "";
    throw new Error("процесс упал");
  }
  store.write(apply(store.get(entry.rowId), entry));
}

function syncOnce(store: Store, state: { cursor: number }, order: "до" | "после") {
  const page = server.pull(state.cursor, 100);
  if (order === "до") state.cursor = page.cursor;
  page.changes.forEach((entry) => applyChange(store, entry));
  if (order === "после") state.cursor = page.cursor;
}

for (const order of ["до", "после"] as const) {
  const store = createStore();
  const state = { cursor: 0 };
  crashAt = "B:2";
  try {
    syncOnce(store, state, order);
  } catch (error) {
    console.log("курсор " + order + " применения:", (error as Error).message, "| курсор", state.cursor);
  }
  // приложение открыли снова: синхронизация идёт с сохранённого курсора
  syncOnce(store, state, order);
  console.log("курсор " + order + " применения: на экране подходов " + store.list().length + " | курсор " + state.cursor);
}

type Fields = Record<string, string | number>;
type Row = Fields & { id: string };
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
interface Store {
  write(row: Row): void;
  get(id: string): Row | undefined;
  list(): Row[];
}
interface Journal {
  newId(): string;
  change(rowId: string, fields: Fields): void;
  outbox(): Op[];
}
interface Server {
  rows: Map<string, Row>;
  journal: (Op & { seq: number })[];
  net: { offline: boolean; loseResponse: boolean };
  push(batch: Op[]): { verdicts: Verdict[] };
  pull(cursor: number, limit: number): { changes: (Op & { seq: number })[]; cursor: number; hasMore: boolean };
}
declare function createStore(): Store;
declare function createJournal(device: string, store: Store): Journal;
declare function createServer(): Server;
declare function apply(row: Row | undefined, op: Op): Row;
interface Client {
  store: Store;
  journal: Journal;
  queue(): Op[];
  readonly cursor: number;
  sync(): void;
}
declare function createClient(device: string, server: Server): Client;

const server = createServer();
const phone = createClient("A", server);
const web = createClient("B", server);

// подход записан на телефоне и уехал на сервер, веб-версия его забрала
const set = phone.journal.newId();
phone.journal.change(set, { exercise: "жим", weightKg: 80, reps: 5 });
phone.sync();
web.sync();
// дома поправили повторы, в зале без сети — вес
web.journal.change(set, { reps: 12 });
web.sync();
phone.journal.change(set, { weightKg: 82.5 });

// ровно та страница, которую телефон сейчас заберёт
const incoming = server.pull(phone.cursor, 100).changes;
console.log("придёт с сервера:", incoming.map((e) => e.opId + " " + JSON.stringify(e.fields)).join(", "));

function receive(store: Store, entry: Op, how: "строкой" | "полями") {
  if (how === "строкой") store.write({ id: entry.rowId, ...entry.fields });
  else store.write(apply(store.get(entry.rowId), entry));
}

for (const how of ["строкой", "полями"] as const) {
  const screen = createStore();
  // то, что сейчас на экране телефона: правка веса применена, но не отправлена
  phone.store.list().forEach((row) => screen.write(row));
  incoming.forEach((entry) => receive(screen, entry, how));
  console.log(how + ":", JSON.stringify(screen.list()), "| не отправлено", phone.queue().length);
}

phone.sync();
console.log("после синхронизации у сервера:", JSON.stringify([...server.rows.values()]));

type Fields = Record<string, string | number>;
type Row = Fields & { id: string };
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
interface Store {
  write(row: Row): void;
  get(id: string): Row | undefined;
  list(): Row[];
  remove(id: string): void;
}
interface Journal {
  newId(): string;
  change(rowId: string, fields: Fields): void;
  outbox(): Op[];
}
interface Server {
  rows: Map<string, Row>;
  journal: (Op & { seq: number })[];
  net: { offline: boolean; loseResponse: boolean };
  push(batch: Op[]): { verdicts: Verdict[] };
  pull(cursor: number, limit: number): { changes: (Op & { seq: number })[]; cursor: number; hasMore: boolean };
}
declare function createStore(): Store;
declare function createJournal(device: string, store: Store): Journal;
declare function createServer(): Server;
declare function fold(log: Op[]): Map<string, Row>;

const server = createServer();
const store = createStore();
const journal = createJournal("A", store);
const dropped = new Set<string>();

const good = journal.newId();
journal.change(good, { exercise: "жим", weightKg: 80, reps: 5 });
const bad = journal.newId();
journal.change(bad, { exercise: "присед", weightKg: 100, reps: 800 });
journal.change(bad, { weightKg: 50 });

const verdicts = server.push(journal.outbox()).verdicts;
console.log("вердикты:", verdicts.map((v) => v.opId + " " + v.status).join(", "));
console.log("просто убрать из очереди:", JSON.stringify(store.list()));

function reject(opId: string) {
  const failed = journal.outbox().find((op) => op.opId === opId);
  if (!failed) return;
  dropped.add(opId);
  const rowOps = journal.outbox().filter((op) => op.rowId === failed.rowId);
  // отвергли вставку — строки нет ни у сервера, ни у нас: уходят все её операции
  if (rowOps[0].opId === opId) rowOps.forEach((op) => dropped.add(op.opId));
  const rest = rowOps.filter((op) => !dropped.has(op.opId));
  if (rest.length === 0) store.remove(failed.rowId);
  else store.write(fold(rest).get(failed.rowId)!);
}

verdicts.filter((v) => v.status === "rejected").forEach((v) => reject(v.opId));
console.log("с пересборкой:", JSON.stringify(store.list()));
const settled = new Set(verdicts.map((v) => v.opId));
const queue = journal.outbox().filter((op) => !settled.has(op.opId) && !dropped.has(op.opId));
console.log("в очереди:", queue.length, "| у сервера строк:", server.rows.size);
console.log("причина отказа:", verdicts.find((v) => v.status === "rejected")?.reason);

type Fields = Record<string, string | number>;
type Row = Fields & { id: string };
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
interface Store {
  write(row: Row): void;
  get(id: string): Row | undefined;
  list(): Row[];
  remove(id: string): void;
  subscribe(listener: () => void): () => void;
}
interface Journal {
  newId(): string;
  change(rowId: string, fields: Fields): void;
  outbox(): Op[];
}
interface Server {
  rows: Map<string, Row>;
  journal: (Op & { seq: number })[];
  net: { offline: boolean; loseResponse: boolean };
  push(batch: Op[]): { verdicts: Verdict[] };
  pull(cursor: number, limit: number): { changes: (Op & { seq: number })[]; cursor: number; hasMore: boolean };
}
declare function createStore(): Store;
declare function createJournal(device: string, store: Store): Journal;
declare function createServer(): Server;
interface Client {
  store: Store;
  journal: Journal;
  queue(): Op[];
  sync(): Verdict[];
}
declare function createClient(device: string, server: Server): Client;
const { View, Text, Pressable } = ReactNative;

const server = createServer();
const client = createClient("A", server);

function Screen() {
  const rows = React.useSyncExternalStore(client.store.subscribe, client.store.list);
  const [note, setNote] = React.useState("");
  const pending = client.queue().length;
  const add = (reps: number) => {
    setNote("");
    client.journal.change(client.journal.newId(), { exercise: "жим", weightKg: 80, reps });
  };
  const sync = () => {
    try {
      const rejected = client.sync();
      setNote(rejected.length > 0 ? "отвергнуто: " + rejected.length + " · " + rejected[0].reason : "");
    } catch (error) {
      setNote((error as Error).message);
    }
  };
  const label = note !== "" ? note : pending > 0 ? "ждут отправки: " + pending : "сохранено на телефоне";
  console.log("индикатор:", label, "| подходов:", rows.length);
  return (
    <View style={{ gap: 6 }}>
      <Text testID="label">{label}</Text>
      <Pressable testID="add" onPress={() => add(5)}><Text>+ подход</Text></Pressable>
      <Pressable testID="add-bad" onPress={() => add(800)}><Text>+ подход с опечаткой</Text></Pressable>
      <Pressable testID="sync" onPress={sync}><Text>синхронизировать</Text></Pressable>
    </View>
  );
}

mount(<Screen />);
setTimeout(() => fireEvent.press("add"), 50);
setTimeout(() => { server.net.offline = true; fireEvent.press("sync"); }, 100);
setTimeout(() => { server.net.offline = false; fireEvent.press("sync"); }, 150);
setTimeout(() => fireEvent.press("add-bad"), 200);
setTimeout(() => fireEvent.press("sync"), 250);

// sync-loop.ts — цикл в приложении
const BACKOFF = [1000, 2000, 4000, 8000, 15000, 30000, 60000];
let attempt = 0;
let timer: ReturnType<typeof setTimeout> | null = null;

export function schedule(delay = 0) {
  if (timer) clearTimeout(timer);
  timer = setTimeout(run, delay);
}

async function run() {
  timer = null;
  try {
    await sync();
    // получилось — лестница отсрочек сбрасывается
    attempt = 0;
    // очередь могла подрасти, пока пачка была в пути
    if (client.queue().length > 0) schedule(0);
  } catch {
    // сетевая ошибка: следующая попытка позже, очередь не тронута
    schedule(BACKOFF[Math.min(attempt, BACKOFF.length - 1)]);
    attempt += 1;
  }
}