Чужой идентификатор: владение в push и pull

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

// слияние по полям и полный порядок правок — из части V, целиком на подложке
declare function merge(row: Fields | undefined, op: Op, mark: Marks): Row;

// строка итога: поля, отметки по полям и номер последней правки
type Stored = { fields: Fields; marks: Marks; seq: number };

// проверяется строка после применения операции, а не поля самой операции
function checkRow(row: Row): string | null {
  const missing = ["exercise", "weightKg", "reps"].filter((field) => !(field in row));
  if (missing.length > 0) return "нет полей: " + missing.join(", ");
  return Number(row.reps) > 100 ? "повторов больше сотни" : null;
}

// Приём пачки из прошлой статьи. Изменилось одно: чем сервер отпирает строку
// итога — одним row_id или парой «владелец и row_id»
function createServer(keyOf: (userId: string, rowId: string) => string) {
  const rows = new Map<string, Stored>();
  const heads = new Map<string, number>();
  return {
    push(userId: string, batch: Op[]): Verdict[] {
      return batch.map((op) => {
        const key = keyOf(userId, op.rowId);
        const stored = rows.get(key);
        const before = stored?.marks ?? {};
        const marks: Marks = { ...before };
        const row = merge(stored?.fields, op, marks);
        const moved = JSON.stringify(marks) !== JSON.stringify(before);
        const reason = stored && !moved ? null : checkRow(row);
        if (reason) return { opId: op.opId, status: "rejected", reason };
        if (moved) {
          const seq = (heads.get(userId) ?? 0) + 1;
          heads.set(userId, seq);
          const { id, ...fields } = row;
          rows.set(key, { fields, marks, seq });
        }
        return { opId: op.opId, status: "applied" };
      });
    },
    show(userId: string, rowId: string): string {
      const stored = rows.get(keyOf(userId, rowId));
      return stored ? JSON.stringify(stored.fields) + " · номер " + stored.seq : "строки нет";
    },
    count(): number {
      return rows.size;
    },
  };
}

// Аня записала подход: имя строки «A-1» придумал её телефон
const ANN: Op[] = [{ opId: "A:1", rowId: "A-1", fields: { exercise: "жим", weightKg: 80, reps: 5 }, at: 1 }];
// Боря присылает чужое имя строки: сначала правку веса, потом вставку целиком
const BOB: Op[] = [
  { opId: "T:1", rowId: "A-1", fields: { weightKg: 5 }, at: 2 },
  { opId: "T:2", rowId: "A-1", fields: { exercise: "тяга", weightKg: 60, reps: 10 }, at: 3 },
];

const KEYS: [string, (userId: string, rowId: string) => string][] = [
  ["ключ — один row_id:", (userId, rowId) => rowId],
  ["ключ — пара (user_id, row_id):", (userId, rowId) => userId + "|" + rowId],
];

for (const [name, keyOf] of KEYS) {
  const server = createServer(keyOf);
  server.push("аня", ANN);
  const verdicts = server.push("боря", BOB);
  console.log(name);
  console.log("  вердикты Боре:", verdicts.map((v) => v.opId + " " + v.status + (v.reason ? " · " + v.reason : "")).join(" | "));
  console.log("  строка Ани:", server.show("аня", "A-1"));
  console.log("  строка Бори:", server.show("боря", "A-1"));
  console.log("  строк у сервера:", server.count());
}

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

declare function merge(row: Fields | undefined, op: Op, mark: Marks): Row;

type Stored = { fields: Fields; marks: Marks };
const rows = new Map<string, Stored>();
const keyOf = (userId: string, rowId: string): string => userId + "|" + rowId;

const SOMEONE_ELSE = "строка другого пользователя";

function checkRow(row: Row): string | null {
  const missing = ["exercise", "weightKg", "reps"].filter((field) => !(field in row));
  if (missing.length > 0) return "нет полей: " + missing.join(", ");
  return Number(row.reps) > 100 ? "повторов больше сотни" : null;
}

// Честный приём: строку ищет ключ пары, и чужая строка им просто не находится
function honest(userId: string, op: Op): Verdict {
  const stored = rows.get(keyOf(userId, op.rowId));
  const marks: Marks = { ...(stored?.marks ?? {}) };
  const reason = checkRow(merge(stored?.fields, op, marks));
  return reason ? { opId: op.opId, status: "rejected", reason } : { opId: op.opId, status: "applied" };
}

// Заботливый: сначала ищет строку по одному row_id и объясняет, чья она
function helpful(userId: string, op: Op): Verdict {
  const owner = [...rows.keys()].find((key) => key.endsWith("|" + op.rowId));
  if (owner !== undefined && owner !== keyOf(userId, op.rowId)) {
    return { opId: op.opId, status: "rejected", reason: SOMEONE_ELSE };
  }
  return honest(userId, op);
}

// у Ани три подхода: имена строк её телефон выдаёт подряд
for (const rowId of ["A-1", "A-2", "A-3"]) {
  const marks: Marks = {};
  const op: Op = { opId: "A:" + rowId, rowId, fields: { exercise: "жим", weightKg: 80, reps: 5 }, at: 1 };
  const { id, ...fields } = merge(undefined, op, marks);
  rows.set(keyOf("аня", rowId), { fields, marks });
}

const NAMES = ["A-1", "A-2", "A-3", "A-4", "A-5", "A-6"];

// Боря перебирает имена строк правкой веса и смотрит только на ответы
function probe(name: string, decide: (userId: string, op: Op) => Verdict): void {
  const answers = NAMES.map((rowId, i) => decide("боря", { opId: "T:" + (i + 1), rowId, fields: { weightKg: 5 }, at: 10 }));
  const kinds = [...new Set(answers.map((v) => v.reason ?? v.status))];
  console.log(name);
  console.log("  разных ответов:", kinds.length, "—", kinds.join(" / "));
  console.log("  карта по ответам:", answers.map((v, i) => NAMES[i] + (v.reason === SOMEONE_ELSE ? " занято" : " —")).join(", "));
}

probe("заботливый сервер:", helpful);
probe("честный сервер:", honest);

type Fields = Record<string, string | number | boolean>;
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Marks = Record<string, { at: number; opId: string }>;
type Row = Fields & { id: string };

declare function merge(row: Fields | undefined, op: Op, mark: Marks): Row;

type Stored = { userId: string; rowId: string; fields: Fields; marks: Marks; seq: number };
const rows: Stored[] = [];
const heads = new Map<string, number>();

// приём сведён к принятой операции: отказы разобраны выше
function write(userId: string, op: Op): void {
  const found = rows.find((row) => row.userId === userId && row.rowId === op.rowId);
  const marks: Marks = { ...(found?.marks ?? {}) };
  const { id, ...fields } = merge(found?.fields, op, marks);
  const seq = (heads.get(userId) ?? 0) + 1;
  heads.set(userId, seq);
  if (found) {
    found.fields = fields;
    found.marks = marks;
    found.seq = seq;
  } else {
    rows.push({ userId, rowId: op.rowId, fields, marks, seq });
  }
}

// условие видимости: у одного сервера владелец в нём есть, у другого нет
type Where = (row: Stored, userId: string) => boolean;
const withoutOwner: Where = () => true;
const withOwner: Where = (row, userId) => row.userId === userId;

function pull(where: Where, userId: string, cursor: number, limit: number) {
  const mine = rows.filter((row) => where(row, userId) && row.seq > cursor).sort((a, b) => a.seq - b.seq);
  const page = mine.slice(0, limit);
  const last = page.length > 0 ? page[page.length - 1].seq : cursor;
  return { page, cursor: last, hasMore: mine.some((row) => row.seq > last) };
}

// цикл клиента части V: крутить, пока сервер говорит «ещё есть»
function catchUp(where: Where, userId: string, limit: number) {
  const got: string[] = [];
  let cursor = 0;
  let hasMore = true;
  let calls = 0;
  while (hasMore && calls < 20) {
    calls += 1;
    const page = pull(where, userId, cursor, limit);
    for (const row of page.page) got.push(row.userId + "/" + row.rowId);
    cursor = page.cursor;
    hasMore = page.hasMore;
  }
  return { got, calls, cursor };
}

const SET: Fields = { exercise: "жим", weightKg: 80, reps: 5 };
write("аня", { opId: "A:1", rowId: "A-1", fields: SET, at: 1 });
write("аня", { opId: "A:2", rowId: "A-2", fields: SET, at: 2 });
write("аня", { opId: "A:3", rowId: "A-3", fields: SET, at: 3 });
// две строки Бори записал его планшет, а забирает их телефон
write("боря", { opId: "T:1", rowId: "B-1", fields: SET, at: 4 });
write("боря", { opId: "T:2", rowId: "B-2", fields: SET, at: 5 });

console.log("номер 1 у сервера носят:", rows.filter((row) => row.seq === 1).map((row) => row.userId + "/" + row.rowId).join(" и "));
for (const [name, where] of [["без владельца в условии:", withoutOwner], ["владелец в условии:", withOwner]] as [string, Where][]) {
  const run = catchUp(where, "боря", 1);
  console.log(name, run.got.join(", "), "| заходов", run.calls, "| курсор", run.cursor);
  console.log("  чужих:", run.got.filter((name2) => name2.startsWith("аня/")).length, "| своих:", run.got.filter((name2) => name2.startsWith("боря/")).length, "из 2");
}

import { bigint, index, jsonb, pgTable, primaryKey, text, uuid } from 'drizzle-orm/pg-core';

export const syncRows = pgTable(
  'sync_rows',
  {
    userId: uuid('user_id')
      .notNull()
      .references(() => users.id, { onDelete: 'cascade' }),
    // имя строки придумал телефон, поэтому ключом оно работает только в паре с владельцем
    rowId: text('row_id').notNull(),
    fields: jsonb('fields').$type<Fields>().notNull(),
    // отметки [at, opId] по полям: слияние сравнивает поле с полем, а не строку со строкой
    marks: jsonb('marks').$type<Marks>().notNull(),
    // номер последней правки строки; mode: 'number' приводит bigint к числу
    // на границе чтения — драйвер отдаёт его строкой
    seq: bigint('seq', { mode: 'number' }).notNull(),
  },
  (table) => [
    primaryKey({ columns: [table.userId, table.rowId] }),
    // выдача отбирает страницу по владельцу и номеру сразу
    index('sync_rows_user_seq_idx').on(table.userId, table.seq),
  ],
);

// packages/db/src/schema/chats.ts боевого проекта автора, выдержка

export const messages = pgTable(
  'messages',
  {
    // монотонный id = курсор keyset-пагинации истории; identity, чтобы приложение
    // не могло подставить своё значение и сломать порядок
    id: bigintNum('id').generatedAlwaysAsIdentity().primaryKey(),
    chatId: uuid('chat_id').notNull().references(() => chats.id, { onDelete: 'cascade' }),
    // денормализация: sync.pull забирает всё новое по пользователю одним запросом без JOIN
    userId: uuid('user_id').notNull().references(() => users.id, { onDelete: 'cascade' }),
    // клиентский nanoid: делает повтор отправки из outbox идемпотентным
    clientMid: text('client_mid'),
    seq: bigintNum('seq').notNull().default(sql`nextval('message_seq')`),
  },
  (t) => [
    index('messages_user_seq_idx').on(t.userId, t.seq),
    // дедуп ретраев исходящей очереди: на дубль сервер отвечает существующим сообщением
    uniqueIndex('messages_chat_client_mid_uq')
      .on(t.chatId, t.clientMid)
      .where(sql`${t.clientMid} is not null`),
  ],
);

// apps/api/src/chat/chat.repository.ts: условие владения диалогом, вынесенное в функцию
export function ownedLiveChat(chatId: string, userId: string): SQL | undefined {
  return and(eq(chats.id, chatId), eq(chats.userId, userId), isNull(chats.deletedAt));
}

// apps/api/src/message/message.service.ts, начало отправки сообщения
const context = await this.repo.findChatContext(identity.userId, input.chatId);
if (!context) throw domainError('CHAT_NOT_FOUND');

// ШАГ 1. Идемпотентность — раньше всего остального: повтор из очереди клиента
// не должен ни списать денег, ни разбудить модель, ни завести второе сообщение
const duplicate = await this.repo.findByClientMid(input.chatId, input.clientMid);
if (duplicate) return this.resultForExisting(identity.userId, duplicate);