backend/src/modules/investments/operations/application/brokerage-operations-sync.service.ts

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

import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
import { MILLISECONDS_PER_SECOND } from '../../../../common/time';
import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
import { TbankOperationState } from '../../../../integrations/tbank/tbank-protocol';
import { TbankAccessService, TbankClient } from '../../../../integrations/tbank/tbank.api';
import {
  type OperationInterval,
  operationIntervals,
  operationsSyncPriority,
  planOperationRead,
} from '../domain/operations-sync-plan';
import {
  BrokerageOperationsRepository,
  type OperationAttempt,
} from '../infrastructure/brokerage-operations.repository';

const UPSTREAM_RANGE_MARGIN_SECONDS = 1;

const operationsPageSize = 1000;

@Injectable()
export class BrokerageOperationsSyncService {
  private readonly processRunId = randomUUID();
  private readonly runningAttempts = new Set<string>();
  private readonly ready = new Map<
    string,
    { attempt: OperationAttempt; priority: number }
  >();

  private draining = false;
  // Only short DB reservations/publications are serialized, never RPCs.
  private reservations: Promise<unknown> = Promise.resolve();

  constructor(
    private readonly repository: BrokerageOperationsRepository,
    private readonly access: TbankAccessService,
    private readonly tbank: TbankClient,
    private readonly logger: StructuredLogger,
  ) {}

  async readPlan(
    userId: string,
    intervals: OperationInterval[],
    options: {
      force?: boolean;
      movingEnd?: boolean;
      observe?: boolean;
      revision?: string;
    } = {},
  ) {
    const connection = await this.access.local(userId);
    if (
      !connection.membershipAt ||
      (options.revision && options.revision !== connection.revision)
    ) {
      return { plan: null, handle: null };
    }
    if (
      intervals.some(
        (interval) =>
          !connection.accounts.some(
            (account) => account.accountId === interval.accountId,
          ),
      )
    ) {
      throw new NotFoundException();
    }
    const coverage = await this.repository.coverage(
      userId,
      connection.revision,
      intervals.map((interval) => interval.accountId),
    );
    const now = new Date();
    const movingTo = Math.max(
      ...intervals.map((interval) => interval.to.valueOf()),
    );
    const targets = intervals.map((interval) =>
      options.movingEnd && interval.to.valueOf() === movingTo
        ? { ...interval, to: now }
        : interval,
    );
    const plan = planOperationRead(targets, coverage, now, options);
    const complete = plan.intervals.every(
      (interval) => interval.availability === 'complete',
    );
    const handle =
      !options.observe && plan.load.length
        ? await this.request(
            userId,
            plan.load,
            complete || Boolean(options.force),
            'screen',
            connection.revision,
          )
        : null;
    return { plan, handle };
  }

  async request(
    userId: string,
    intervals: OperationInterval[],
    atomic: boolean,
    priority: keyof typeof operationsSyncPriority = 'screen',
    revision?: string,
  ): Promise<null | string> {
    const connection = await this.access.local(userId);
    if (
      connection.membershipAt &&
      intervals.some(
        (interval) =>
          !connection.accounts.some(
            (account) => account.accountId === interval.accountId,
          ),
      )
    ) {
      throw new NotFoundException();
    }
    if (revision && revision !== connection.revision) return null;
    const cutoff = new Date();
    const bounded = operationIntervals(intervals)
      .map((interval) => ({
        ...interval,
        to: new Date(Math.min(interval.to.valueOf(), cutoff.valueOf())),
      }))
      .filter((interval) => interval.from < interval.to);
    if (!bounded.length) return null;
    const reserve = this.reservations.then(async () => {
      const requested = await this.repository.request(
        userId,
        connection.revision,
        operationIntervals(bounded),
        this.processRunId,
        atomic,
        this.runningAttempts,
        // eslint-disable-next-line security/detect-object-injection -- The priority belongs to the internal priority union.
        operationsSyncPriority[priority],
      );
      if (!requested) return null;
      const { id, attemptId, revision } = requested.work;
      this.runningAttempts.add(attemptId);
      this.ready.set(attemptId, {
        attempt: { id, attemptId, userId, revision },
        priority: requested.work.priority,
      });
      if (!this.draining) {
        this.draining = true;
        // Startup and observation never enter this path.
        setImmediate(() => {
          void this.drain();
        });
      }
      return id;
    });
    this.reservations = reserve.catch(() => undefined);
    return reserve;
  }

  async observe(userId: string, id: string) {
    const connection = await this.access.local(userId);
    const work = await this.repository.observe(userId, connection.revision, id);
    if (!work) throw new NotFoundException();
    return work;
  }

  private async drain(): Promise<void> {
    try {
      while (this.ready.size) {
        const selected = [...this.ready.values()].sort(
          (a, b) => a.priority - b.priority,
        )[0];
        const { attempt } = selected;
        this.ready.delete(attempt.attemptId);
        let pending = false;
        try {
          pending = await this.step(attempt);
        } catch {
          this.logger.event(
            'error',
            'tbank.operations.sync.persistence_failed',
            { userId: attempt.userId, code: TbankErrorCode.Unavailable },
          );
          // The step could not even record its failure; close the attempt
          // durably so it does not stay running. Best effort — an open row
          // is re-reserved by the next request.
          await this.repository
            .fail(attempt, false)
            .catch(() => undefined);
        }
        const next = pending ? await this.repository.pending(attempt) : null;
        if (next) {
          const joined = this.ready.get(attempt.attemptId);
          this.ready.set(attempt.attemptId, {
            attempt,
            priority: Math.min(
              next.priority,
              joined?.priority ?? next.priority,
            ),
          });
        } else {
          this.ready.delete(attempt.attemptId);
          this.runningAttempts.delete(attempt.attemptId);
        }
      }
    } finally {
      this.draining = false;
    }
  }

  private async step(attempt: OperationAttempt): Promise<boolean> {
    try {
      const plan = await this.repository.pending(attempt);
      if (!plan) return false;
      if (!plan.next) {
        const publish = this.reservations.then(async () => {
          const current = await this.repository.pending(attempt);
          if (!current) return false;
          if (current.next) return true;
          const published = await this.repository.publish(
            attempt,
            current.intervals,
            new Date(),
            true,
          );
          return published && Boolean(await this.repository.pending(attempt));
        });
        this.reservations = publish.catch(() => undefined);
        return await publish;
      }
      const { interval, cursor } = plan.next;
      const token = await this.access.tokenForAttempt(attempt);
      if (!token) return false;
      const response = await this.tbank.getOperationsByCursor(token, {
        accountId: interval.accountId,
        // Upstream inclusivity is not assumed; storage selects exact [from,to).
        from: {
          seconds: String(
            Math.floor(interval.from.valueOf() / MILLISECONDS_PER_SECOND) - UPSTREAM_RANGE_MARGIN_SECONDS,
          ),
        },
        to: {
          seconds: String(
            Math.ceil(interval.to.valueOf() / MILLISECONDS_PER_SECOND) + UPSTREAM_RANGE_MARGIN_SECONDS,
          ),
        },
        ...(interval.instrumentUid
          ? { instrumentId: interval.instrumentUid }
          : {}),
        cursor,
        limit: operationsPageSize,
        state: TbankOperationState.Executed,
      });
      if (
        !(await this.repository.stagePage(attempt, interval, cursor, response))
      ) {
        return false;
      }
      if (!plan.atomic && !response.hasNext) {
        if (
          !(await this.repository.publish(
            attempt,
            [interval],
            new Date(),
            true,
          ))
        ) {
          return false;
        }
      }
      // One page per turn: newly requested screen work can overtake backfill.
      // TbankClient alone owns the bounded RPC queue and retries.
      return true;
    } catch (error) {
      await this.repository.fail(
        attempt,
        error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken),
      );
      return false;
    }
  }
}