backend/src/modules/investments/operations/infrastructure/brokerage-operations.repository.ts

import {
  BadRequestException,
  Injectable,
  NotFoundException,
} from '@nestjs/common';
import { createHash, randomUUID } from 'node:crypto';
import { isDeepStrictEqual } from 'node:util';

import type { Prisma } from '../../../../../generated/prisma/client';
import type {
  TbankOperationItem,
  TbankOperationsByCursorResponse,
} from '../../../../integrations/tbank/tbank.api';

import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
import { isTbankProtocolValue, TbankOperationState } from '../../../../integrations/tbank/tbank-protocol';
import { PrismaService } from '../../../../prisma/prisma.service';
import {
  dateOnlyBounds,
  DateOnlyRange,
  moscowDate,
} from '../../investment-calendar';
import { OperationErrorCode } from '../domain/operation-error-code';
import {
  type OperationInterval,
  operationIntervals,
  operationIntervalsOverlap,
  operationsSyncPriority,
  uncoveredOperationIntervals,
} from '../domain/operations-sync-plan';
import {
  operationCategory,
  operationTimestamp,
} from './brokerage-operation.mapper';

const CURSOR_LOOKAHEAD_ROWS = 1;
const EMPTY_INTERVAL_COUNT = 0;
const SINGLE_WORK_GROUP = 1;
const INITIAL_PAGE_COUNT = 0;
const PAGE_INCREMENT = 1;
const EXPECTED_UPDATED_WORK_ITEMS = 1;

export type { OperationInterval } from '../domain/operations-sync-plan';

export type OperationAttempt = {
  attemptId: string;
  id: string;
  revision: string;
  userId: string;
};
type OperationsIntent = {
  atomic: boolean;
  atomicIntervals?: string[];
  intervals: string[];
  screenIntervals?: string[];
};
type PageProgress = {
  complete: boolean;
  cursor: string;
  pages: number;
  seen: string[];
};
const progressItem = '';

@Injectable()
export class BrokerageOperationsRepository {
  constructor(private readonly prisma: PrismaService) {}

  async readPage(input: {
    accountId: string;
    categories: string[];
    cursor?: { accountId: string; at: string; binding: string; id: string };
    filterKey: string;
    from: Date;
    instrumentUid?: string;
    limit?: number;
    periods?: (openedAt: Date | undefined) => DateOnlyRange[];
    revision: string;
    to: Date;
    userId: string;
  }) {
    return this.prisma.$transaction(async (tx) => {
      const connection = await tx.tbankConnection.findUnique({
        where: {
          userId_revision: { userId: input.userId, revision: input.revision },
          invalidToken: false,
        },
        select: {
          membershipAt: true,
          membershipRevision: true,
          accounts: {
            where:
              input.accountId === 'all' ? {} : { accountId: input.accountId },
            orderBy: { accountId: 'asc' },
            select: {
              accountId: true,
              name: true,
              openedOn: true,
              operationsRevision: true,
            },
          },
        },
      });
      if (!connection) return null;
      if (
        connection.membershipAt &&
        input.accountId !== 'all' &&
        !connection.accounts.length
      ) {
        throw new NotFoundException();
      }
      const accounts = connection.accounts;
      // eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings.
      const oldest = accounts.map((account) => account.openedOn).sort()[0];
      const openedAt =
        oldest && accounts.every((account) => account.openedOn)
          ? new Date(`${oldest}T00:00:00+03:00`)
          : undefined;
      const periods = input.periods?.(openedAt).map((period) => {
        const now = new Date();
        const bounds = dateOnlyBounds(period, now);
        return {
          ...bounds,
          key: `${period.from}/${period.to}`,
          movingEnd: period.to >= moscowDate(now),
        };
      });
      const binding = createHash('sha256')
        .update(
          JSON.stringify([
            input.userId,
            input.revision,
            input.accountId === 'all' ? connection.membershipRevision : null,
            accounts.map((account) => [
              account.accountId,
              account.operationsRevision,
            ]),
            input.filterKey,
          ]),
        )
        .digest('base64url');
      if (input.cursor && input.cursor.binding !== binding) {
        throw new BadRequestException({ code: OperationErrorCode.InvalidOperationsCursor });
      }
      const scope = {
        userId: input.userId,
        revision: input.revision,
        accountId: { in: accounts.map((account) => account.accountId) },
      };
      const coverage = await tx.tbankOperationCoverage.findMany({
        where: scope,
      });
      const work = await tx.tbankSyncWork.findMany({
        where: {
          userId: input.userId,
          revision: input.revision,
          scopeKey: { startsWith: 'operations:' },
          state: { in: ['running', 'failed'] },
        },
        select: {
          id: true,
          state: true,
          errorCode: true,
          intent: true,
          priority: true,
        },
      });
      const membershipWork = await tx.tbankSyncWork.findUnique({
        where: {
          userId_revision_scopeKey: {
            userId: input.userId,
            revision: input.revision,
            scopeKey: 'membership',
          },
        },
        select: { id: true, state: true, errorCode: true },
      });
      const after = input.cursor;
      const rows = await tx.tbankOperation.findMany({
        where: {
          ...scope,
          ...(periods
            ? {
                OR: periods.map((period) => ({
                  occurredAt: { gte: period.from, lt: period.to },
                })),
              }
            : { occurredAt: { gte: input.from, lt: input.to } }),
          ...(input.instrumentUid
            ? { instrumentUid: input.instrumentUid }
            : {}),
          ...(input.categories.length
            ? { category: { in: input.categories } }
            : {}),
          ...(after
            ? {
                OR: [
                  { occurredAt: { lt: new Date(after.at) } },
                  {
                    occurredAt: new Date(after.at),
                    accountId: { gt: after.accountId },
                  },
                  {
                    occurredAt: new Date(after.at),
                    accountId: after.accountId,
                    brokerOperationId: { gt: after.id },
                  },
                ],
              }
            : {}),
        },
        orderBy: [
          { occurredAt: 'desc' },
          { accountId: 'asc' },
          { brokerOperationId: 'asc' },
        ],
        ...(input.limit === undefined ? {} : { take: input.limit + CURSOR_LOOKAHEAD_ROWS }),
      });
      return {
        connection,
        periods,
        membershipWork,
        binding,
        rows,
        coverage: coverage.map(({ instrumentPredicate, ...row }) => ({
          ...row,
          instrumentUid: instrumentPredicate || undefined,
        })),
        work: work.map(({ intent, ...row }) => ({
          ...row,
          intervals: (intent as { intervals: string[] }).intervals.map(
            decodeInterval,
          ),
        })),
      };
    });
  }

  async coverage(userId: string, revision: string, accountIds: string[]) {
    const rows = await this.prisma.tbankOperationCoverage.findMany({
      where: {
        userId,
        revision,
        accountId: { in: accountIds },
        account: { connection: { invalidToken: false } },
      },
    });
    return rows.map(({ instrumentPredicate, ...row }) => ({
      ...row,
      instrumentUid: instrumentPredicate || undefined,
    }));
  }

  async request(
    userId: string,
    revision: string,
    intervals: OperationInterval[],
    processRunId: string,
    atomic: boolean,
    runningAttempts: ReadonlySet<string>,
    priority: number,
  ) {
    if (!intervals.length) throw new Error('Empty operations request');
    const scopeKey =
      'operations:' +
      createHash('sha256')
        // eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings.
        .update(JSON.stringify([...new Set(intervals.map(intervalKey))].sort()))
        .digest('hex');
    return this.prisma.$transaction(async (tx) => {
      const locked = await tx.tbankConnection.updateMany({
        where: {
          userId,
          revision,
          invalidToken: false,
          membershipAt: { not: null },
        },
        data: { membershipRevision: { increment: 0 } },
      });
      if (!locked.count) return null;
      const accounts = await tx.tbankAccount.findMany({
        where: { userId, revision },
        select: { accountId: true, openedOn: true },
      });
      if (
        intervals.some(
          (interval) =>
            !accounts.some(
              (account) => account.accountId === interval.accountId,
            ),
        )
      ) {
        throw new NotFoundException();
      }
      if (priority > operationsSyncPriority.screen) {
        for (const account of accounts.filter((account) =>
          intervals.some(
            (interval) => interval.accountId === account.accountId,
          ),
        )) {
          await tx.tbankAccount.update({
            where: {
              userId_revision_accountId: {
                userId,
                revision,
                accountId: account.accountId,
              },
            },
            data: {
              historyFrom: new Date(
                account.openedOn
                  ? `${account.openedOn}T00:00:00Z`
                  : '1970-01-01T00:00:00Z',
              ),
              historyBoundary: account.openedOn ? 'known' : 'unknown',
            },
          });
        }
      }
      const coverage = await tx.tbankOperationCoverage.findMany({
        where: {
          userId,
          revision,
          accountId: { in: intervals.map((interval) => interval.accountId) },
        },
      });
      const publicationAtomic =
        atomic ||
        intervals.every(
          (interval) =>
            uncoveredOperationIntervals(
              [interval],
              coverage
                .filter(
                  (row) =>
                    row.accountId === interval.accountId &&
                    (!row.instrumentPredicate ||
                      row.instrumentPredicate === interval.instrumentUid),
                )
                .map((row) => ({
                  accountId: row.accountId,
                  from: row.from,
                  to: row.to,
                  instrumentUid: interval.instrumentUid,
                })),
            ).length === EMPTY_INTERVAL_COUNT,
        );
      const active = await tx.tbankSyncWork.findMany({
        where: {
          userId,
          revision,
          state: 'running',
          scopeKey: { startsWith: 'operations:' },
        },
        orderBy: [{ createdAt: 'asc' }, { id: 'asc' }],
      });
      const touching = active.filter((work) => {
        const pending = (work.intent as { intervals: string[] }).intervals.map(
          decodeInterval,
        );
        return pending.some((existing) =>
          intervals.some((incoming) =>
            operationIntervalsOverlap(existing, incoming),
          ),
        );
      });
      const previous =
        touching[0] ??
        (await tx.tbankSyncWork.findUnique({
          where: { userId_revision_scopeKey: { userId, revision, scopeKey } },
        }));
      const identity = {
        userId,
        revision,
        scopeKey: previous?.scopeKey ?? scopeKey,
      };
      const previousIntent = previous?.intent as OperationsIntent | undefined;
      const upgradeGroup =
        publicationAtomic && previousIntent && !previousIntent.atomic;
      if (
        touching.length <= SINGLE_WORK_GROUP &&
        previous?.state === 'running' &&
        previous.processRunId === processRunId &&
        runningAttempts.has(previous.attemptId) &&
        !upgradeGroup
      ) {
        const pending = previousIntent?.intervals.map(decodeInterval) ?? [];
        const additions =
          previous.scopeKey === scopeKey
            ? []
            : uncoveredOperationIntervals(intervals, pending);
        const work = await tx.tbankSyncWork.update({
          where: { id: previous.id },
          data: {
            intent: {
              atomic: previousIntent?.atomic ?? publicationAtomic,
              atomicIntervals: previousIntent?.atomic
                ? [
                    ...new Set([
                      ...(previousIntent.atomicIntervals ??
                        previousIntent.intervals),
                      ...additions.map(intervalKey),
                    ]),
                  ]
                : [],
              screenIntervals: [
                ...new Set([
                  ...(previousIntent?.screenIntervals ??
                    (previous.priority === operationsSyncPriority.screen
                      ? (previousIntent?.intervals ?? [])
                      : [])),
                  ...(priority === operationsSyncPriority.screen
                    ? [...pending, ...additions]
                        .filter((item) =>
                          intervals.some((target) =>
                            operationIntervalsOverlap(item, target),
                          ),
                        )
                        .map(intervalKey)
                    : []),
                ]),
              ],
              intervals: (priority < previous.priority
                ? [
                    ...pending.filter((item) =>
                      intervals.some((target) =>
                        operationIntervalsOverlap(item, target),
                      ),
                    ),
                    ...additions,
                    ...pending.filter(
                      (item) =>
                        !intervals.some((target) =>
                          operationIntervalsOverlap(item, target),
                        ),
                    ),
                  ]
                : [...pending, ...additions]
              ).map(intervalKey),
            },
            priority: Math.min(priority, previous.priority),
          },
        });
        return { work, launch: false };
      }
      const recovering =
        previous &&
        previous.state !== 'complete' &&
        previous.state !== 'superseded';
      const retained = recovering
        ? (previousIntent?.intervals.map(decodeInterval) ?? [])
        : [];
      // A bridge between two active groups must fence both producers before
      // either can publish a subset of the newly combined result.
      for (const joined of touching) {
        if (joined.id === previous?.id) continue;
        retained.push(
          ...(joined.intent as { intervals: string[] }).intervals.map(
            decodeInterval,
          ),
        );
        await tx.tbankSyncWork.update({
          where: { id: joined.id },
          data: {
            state: 'superseded',
            progress: { successorId: previous?.id ?? '' },
          },
        });
        await tx.tbankSyncStage.deleteMany({ where: { workId: joined.id } });
      }
      if (previous) {
        await tx.tbankSyncStage.deleteMany({ where: { workId: previous.id } });
      }
      const targets =
        recovering && !upgradeGroup && touching.length <= SINGLE_WORK_GROUP
          ? previous?.scopeKey === scopeKey
            ? retained
            : [...retained, ...uncoveredOperationIntervals(intervals, retained)]
          : operationIntervals([...retained, ...intervals]);
      const atomicTargets = [
        ...(publicationAtomic ? intervals : []),
        ...(recovering && previousIntent?.atomic
          ? (previousIntent.atomicIntervals ?? previousIntent.intervals).map(
              decodeInterval,
            )
          : []),
        ...touching.flatMap((work) => {
          const joined = work.intent as OperationsIntent;
          return joined.atomic
            ? (joined.atomicIntervals ?? joined.intervals).map(decodeInterval)
            : [];
        }),
      ];
      const intent = {
        intervals: targets.map(intervalKey),
        atomicIntervals: targets
          .filter((target) =>
            atomicTargets.some((atomic) =>
              operationIntervalsOverlap(target, atomic),
            ),
          )
          .map(intervalKey),
        screenIntervals: targets
          .filter(
            (target) =>
              (priority === operationsSyncPriority.screen &&
                intervals.some((incoming) =>
                  operationIntervalsOverlap(target, incoming),
                )) ||
                (
                  previousIntent?.screenIntervals ??
                  (previous?.priority === operationsSyncPriority.screen
                    ? (previousIntent?.intervals ?? [])
                    : [])
                ).some((key) =>
                  operationIntervalsOverlap(target, decodeInterval(key)),
                ),
          )
          .map(intervalKey),
        atomic: atomicTargets.length > EMPTY_INTERVAL_COUNT,
      };
      const data = {
        intent,
        state: 'running',
        attemptId: randomUUID(),
        processRunId,
        priority: Math.min(
          priority,
          previous?.priority ?? priority,
          ...touching.map((work) => work.priority),
        ),
        progress: {},
        errorCode: null,
      };
      const work = await tx.tbankSyncWork.upsert({
        where: { userId_revision_scopeKey: identity },
        create: { ...identity, ...data, id: randomUUID() },
        update: data,
      });
      return { work, launch: true };
    });
  }

  async pending(attempt: OperationAttempt) {
    const work = await this.prisma.tbankSyncWork.findFirst({
      where: {
        ...attempt,
        state: 'running',
        connection: { invalidToken: false },
      },
      select: { intent: true, priority: true },
    });
    if (!work) return null;
    const intent = work.intent as OperationsIntent;
    const markers = await this.prisma.tbankSyncStage.findMany({
      where: {
        workId: attempt.id,
        attemptId: attempt.attemptId,
        itemKey: progressItem,
      },
      select: { sourceKey: true, payload: true },
    });
    const progress = new Map(
      markers.map((marker) => [
        marker.sourceKey,
        marker.payload as PageProgress,
      ]),
    );
    const group = intent.atomic
      ? (intent.atomicIntervals ?? intent.intervals)
      : intent.intervals;
    const nextKey = group.find((key) => !progress.get(key)?.complete);
    return {
      priority: work.priority,
      atomic: intent.atomic,
      intervals: group.map(decodeInterval),
      next: nextKey
        ? {
            interval: decodeInterval(nextKey),
            cursor: progress.get(nextKey)?.cursor || undefined,
          }
        : null,
    };
  }

  async observe(userId: string, revision: string, id: string) {
    let target = id;
    const seen = new Set<string>();
    while (!seen.has(target)) {
      seen.add(target);
      const work = await this.prisma.tbankSyncWork.findFirst({
        where: {
          id: target,
          userId,
          revision,
          scopeKey: { startsWith: 'operations:' },
        },
        select: {
          id: true,
          state: true,
          progress: true,
          errorCode: true,
          updatedAt: true,
        },
      });
      if (!work || work.state !== 'superseded') {
        return work ? { ...work, id } : null;
      }
      target = (work.progress as { successorId: string }).successorId;
    }
    throw new Error('Cyclic operations work successor');
  }

  async fail(attempt: OperationAttempt, invalidToken: boolean) {
    return this.prisma.$transaction(async (tx) => {
      if (!(await claim(tx, attempt))) return false;
      await tx.tbankSyncWork.update({
        where: { id: attempt.id },
        data: {
          state: 'failed',
          errorCode: invalidToken ? TbankErrorCode.InvalidToken : TbankErrorCode.Unavailable,
        },
      });
      if (invalidToken) {
        await tx.tbankConnection.update({
          where: {
            userId_revision: {
              userId: attempt.userId,
              revision: attempt.revision,
            },
          },
          data: { invalidToken: true },
        });
      }
      return true;
    });
  }

  async stagePage(
    attempt: OperationAttempt,
    interval: OperationInterval,
    cursor: string | undefined,
    response: TbankOperationsByCursorResponse,
  ): Promise<boolean> {
    const sourceKey = intervalKey(interval);
    return this.prisma.$transaction(async (tx) => {
      if (!(await claim(tx, attempt))) return false;
      const stage = {
        workId: attempt.id,
        attemptId: attempt.attemptId,
        sourceKey,
      };
      const marker = await tx.tbankSyncStage.findUnique({
        where: {
          workId_attemptId_sourceKey_itemKey: {
            ...stage,
            itemKey: progressItem,
          },
        },
      });
      const progress = marker?.payload as PageProgress | undefined;
      if (progress?.complete || (progress?.cursor ?? '') !== (cursor ?? '')) {
        throw new Error('Unexpected operations page');
      }
      if (response.hasNext === undefined) {
        throw new Error('Missing operations pagination state');
      }
      const nextCursor = response.nextCursor ?? '';
      if (
        response.hasNext &&
        (!nextCursor ||
          nextCursor === cursor ||
          progress?.seen.includes(nextCursor))
      ) {
        throw new Error('Invalid operations continuation');
      }
      for (const operation of response.items ?? []) {
        if (!operation.id?.trim()) {
          throw new Error('Missing broker operation identity');
        }
        const occurredAt = operationTimestamp(operation.date);
        if (
          operation.date?.seconds === undefined ||
          !Number.isFinite(occurredAt.valueOf())
        ) {
          throw new Error('Invalid operation timestamp');
        }
        if (
          occurredAt < interval.from ||
          occurredAt >= interval.to ||
          (interval.instrumentUid &&
            operation.instrumentUid !== interval.instrumentUid)
        ) {
          continue;
        }
        const payload = JSON.parse(
          JSON.stringify({ ...operation, cursor: undefined }),
        ) as Prisma.InputJsonObject;
        // Provider pagination position is not part of a financial row.
        const identity = { ...stage, itemKey: operation.id };
        const previous = await tx.tbankSyncStage.findUnique({
          where: { workId_attemptId_sourceKey_itemKey: identity },
        });
        if (previous && !isDeepStrictEqual(previous.payload, payload)) {
          throw new Error('Conflicting broker operation identity');
        }
        if (!previous) {
          await tx.tbankSyncStage.create({ data: { ...identity, payload } });
        }
      }
      const payload: PageProgress = {
        complete: !response.hasNext,
        cursor: nextCursor,
        seen: [
          ...(progress?.seen ?? []),
          ...(response.hasNext ? [nextCursor] : []),
        ],
        pages: (progress?.pages ?? INITIAL_PAGE_COUNT) + PAGE_INCREMENT,
      };
      await tx.tbankSyncStage.upsert({
        where: {
          workId_attemptId_sourceKey_itemKey: {
            ...stage,
            itemKey: progressItem,
          },
        },
        create: { ...stage, itemKey: progressItem, payload },
        update: { payload },
      });
      await tx.tbankSyncWork.update({
        where: { id: attempt.id },
        data: { progress: { sourceKey, ...payload } },
      });
      return true;
    });
  }

  // A complete shown scope passes every account/month/comparison here together.
  // Initial missing history may pass one completed interval at a time.
  async publish(
    attempt: OperationAttempt,
    intervals: OperationInterval[],
    fetchedAt: Date,
    advanceWork = false,
  ): Promise<boolean> {
    if (!intervals.length) {
      throw new Error('Empty operations publication group');
    }
    return this.prisma.$transaction(
      async (tx) => {
        if (!(await claim(tx, attempt))) return false;
        const changedAccounts = new Set<string>();
        const groupRows = new Map<string, Prisma.JsonValue>();
        for (const interval of intervals) {
          const sourceKey = intervalKey(interval);
          const stage = {
            workId: attempt.id,
            attemptId: attempt.attemptId,
            sourceKey,
          };
          const marker = await tx.tbankSyncStage.findUnique({
            where: {
              workId_attemptId_sourceKey_itemKey: {
                ...stage,
                itemKey: progressItem,
              },
            },
          });
          if (!(marker?.payload as PageProgress | undefined)?.complete) {
            throw new Error('Incomplete operations publication group');
          }
          const staged = await tx.tbankSyncStage.findMany({
            where: { ...stage, itemKey: { not: progressItem } },
          });
          for (const row of staged) {
            const identity = JSON.stringify([interval.accountId, row.itemKey]);
            const previous = groupRows.get(identity);
            if (
              previous !== undefined &&
              !isDeepStrictEqual(previous, row.payload)
            ) {
              throw new Error(
                'Conflicting broker operation identity across intervals',
              );
            }
            groupRows.set(identity, row.payload);
          }
          const rows = staged.filter((row) => {
            const state = (row.payload as TbankOperationItem).state;
            return !state || isTbankProtocolValue(state, TbankOperationState.Executed);
          });
          const scope = {
            userId: attempt.userId,
            revision: attempt.revision,
            accountId: interval.accountId,
          };
          const existing = await tx.tbankOperation.findMany({
            where: {
              ...scope,
              occurredAt: { gte: interval.from, lt: interval.to },
              ...(interval.instrumentUid
                ? { instrumentUid: interval.instrumentUid }
                : {}),
            },
          });
          const incoming = new Set(rows.map((row) => row.itemKey));
          const removed = existing.filter(
            (row) => !incoming.has(row.brokerOperationId),
          );
          for (const row of removed) {
            await tx.tbankOperation.delete({
              where: {
                userId_revision_accountId_brokerOperationId: {
                  ...scope,
                  brokerOperationId: row.brokerOperationId,
                },
              },
            });
          }
          if (removed.length) changedAccounts.add(interval.accountId);
          for (const row of rows) {
            const operation = row.payload as TbankOperationItem;
            const identity = { ...scope, brokerOperationId: row.itemKey };
            const previous = await tx.tbankOperation.findUnique({
              where: { userId_revision_accountId_brokerOperationId: identity },
            });
            if (previous && isDeepStrictEqual(previous.payload, operation)) {
              continue;
            }
            const data = {
              occurredAt: operationTimestamp(operation.date),
              instrumentUid: operation.instrumentUid ?? null,
              category: operationCategory(operation.type),
              payload: row.payload as Prisma.InputJsonObject,
            };
            await tx.tbankOperation.upsert({
              where: { userId_revision_accountId_brokerOperationId: identity },
              create: { ...identity, ...data },
              update: data,
            });
            changedAccounts.add(interval.accountId);
          }
          const coverage = {
            ...scope,
            instrumentPredicate: interval.instrumentUid ?? '',
            from: interval.from,
            to: interval.to,
          };
          await tx.tbankOperationCoverage.upsert({
            where: {
              userId_revision_accountId_instrumentPredicate_from_to: coverage,
            },
            create: { ...coverage, fetchedAt, asOf: interval.to },
            update: { fetchedAt, asOf: interval.to },
          });
        }
        for (const accountId of changedAccounts) {
          await tx.tbankAccount.update({
            where: {
              userId_revision_accountId: {
                userId: attempt.userId,
                revision: attempt.revision,
                accountId,
              },
            },
            data: { operationsRevision: { increment: 1 } },
          });
        }
        for (const interval of intervals) {
          await tx.tbankSyncStage.deleteMany({
            where: {
              workId: attempt.id,
              attemptId: attempt.attemptId,
              sourceKey: intervalKey(interval),
            },
          });
        }
        if (advanceWork) {
          const work = await tx.tbankSyncWork.findUniqueOrThrow({
            where: { id: attempt.id },
          });
          const intent = work.intent as OperationsIntent;
          const published = new Set(intervals.map(intervalKey));
          const remaining = intent.intervals.filter(
            (key) => !published.has(key),
          );
          await tx.tbankSyncWork.update({
            where: { id: attempt.id },
            data: {
              intent: {
                ...intent,
                intervals: remaining,
                atomic:
                  intent.atomic &&
                  (intent.atomicIntervals ?? intent.intervals).some(
                    (key) => !published.has(key),
                  ),
                atomicIntervals: (
                  intent.atomicIntervals ?? intent.intervals
                ).filter((key) => !published.has(key)),
              },
              state: remaining.length ? 'running' : 'complete',
              priority: remaining.some((key) =>
                (
                  intent.screenIntervals ??
                  (work.priority === operationsSyncPriority.screen ? intent.intervals : [])
                ).includes(key),
              )
                ? operationsSyncPriority.screen
                : operationsSyncPriority.history,
              progress: { remainingIntervals: remaining.length },
            },
          });
        }
        return true;
      },
      { timeout: 30_000 },
    );
  }
}

function intervalKey(interval: OperationInterval): string {
  // eslint-disable-next-line sonarjs/no-inverted-boolean-check -- Negation also rejects Invalid Date; >= would silently accept NaN.
  if (!(interval.from < interval.to)) {
    throw new Error('Invalid operations interval');
  }
  return JSON.stringify([
    interval.accountId,
    interval.from.toISOString(),
    interval.to.toISOString(),
    interval.instrumentUid ?? '',
  ]);
}

async function claim(
  tx: Prisma.TransactionClient,
  attempt: OperationAttempt,
): Promise<boolean> {
  const result = await tx.tbankSyncWork.updateMany({
    where: {
      ...attempt,
      state: 'running',
      connection: { invalidToken: false },
    },
    data: { updatedAt: new Date() },
  });
  return result.count === EXPECTED_UPDATED_WORK_ITEMS;
}

function decodeInterval(key: string): OperationInterval {
  const [accountId, from, to, instrumentUid] = JSON.parse(key) as string[];
  return {
    accountId,
    from: new Date(from),
    to: new Date(to),
    ...(instrumentUid ? { instrumentUid } : {}),
  };
}