backend/src/modules/investments/tbank-sync/tbank-hourly.scheduler.ts

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

import { TbankErrorCode } from '../../../integrations/tbank/tbank-error-code';
import {
  TbankAccessService,
  TbankMembershipService,
  type TbankSyncAttempt,
} from '../../../integrations/tbank/tbank.api';
import { PrismaService } from '../../../prisma/prisma.service';
import { BrokerageOperationsSyncService } from '../operations/operations.api';
import { PortfolioService } from '../portfolio/portfolio.api';
import { UpcomingEventsService } from '../upcoming-events/upcoming-events.service';
import {
  type HourlySection,
  TbankSyncRepository,
} from './tbank-sync.repository';

const HOUR_MS = 3_600_000;
const TICK_MS = 60_000;
const WORK_POLL_MS = 100;
const FIRST_DAY_OF_MONTH = 1;
const ONE_ATTEMPT = 1;
const NO_ATTEMPTS = 0;

type SettledWork = { errorCode: null | string; id: string; state: string };

// One native timer per process. The timer only proposes work; the durable
// reservation in TbankSyncRepository is the fence that keeps a single
// attempt per connection. The deployment runs a single backend process.
// ponytail: no inter-process lease; add one before scaling past one process.
@Injectable()
export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
  private readonly processRunId = randomUUID();
  private timer: NodeJS.Timeout | null = null;
  private ticking = false;

  constructor(
    private readonly prisma: PrismaService,
    private readonly repository: TbankSyncRepository,
    private readonly access: TbankAccessService,
    private readonly membership: TbankMembershipService,
    private readonly portfolio: PortfolioService,
    private readonly operations: BrokerageOperationsSyncService,
    private readonly events: UpcomingEventsService,
  ) {}

  onModuleInit() {
    this.timer = setInterval(() => void this.tick(), TICK_MS);
  }

  onModuleDestroy() {
    if (this.timer) clearInterval(this.timer);
  }

  async tick() {
    if (this.ticking) return;
    this.ticking = true;
    try {
      const connections = await this.prisma.tbankConnection.findMany({
        where: { invalidToken: false },
        select: { userId: true },
      });
      for (const { userId } of connections) {
        await this.launch(userId);
      }
    } catch {
      // A failed tick only delays the proposal until the next tick.
    } finally {
      this.ticking = false;
    }
  }

  // A fresh connection starts its first walk immediately, regardless of the
  // previous term of the replaced connection.
  private async launch(userId: string) {
    const reserved = await this.repository.reserveHourly(
      userId,
      this.processRunId,
      HOUR_MS,
    );
    if (!reserved) return null;
    const { userId: owner, revision, id, attemptId } = reserved.work;
    setImmediate(() => {
      void this.walk({ userId: owner, revision, id, attemptId });
    });
    return reserved.work;
  }

  // One membership pass, then the owners refresh their own datasets. The
  // walk publishes only safe section results and holds no transaction or
  // upstream call inside one.
  private async walk(attempt: TbankSyncAttempt) {
    const failures: HourlySection[] = [];
    const attempted = { count: NO_ATTEMPTS };
    try {
      const { userId, revision } = attempt;
      await this.refreshMembership(userId, revision, failures, attempted);
      if (failures.some((item) => item.errorCode === TbankErrorCode.InvalidToken)) {
        return void this.settle(attempt, failures, attempted.count);
      }
      const current = await this.access.local(userId);
      if (current.revision !== revision) {
        return void this.settle(attempt, failures, attempted.count);
      }
      await this.refreshAccounts(userId, revision, current.accounts, failures, attempted);
      await this.settle(attempt, failures, attempted.count);
    } catch {
      await this.settle(attempt, failures, attempted.count);
    }
  }

  private async refreshMembership(
    userId: string,
    revision: string,
    failures: HourlySection[],
    attempted: { count: number },
  ) {
    const member = await this.membership.request(userId, revision, true);
    attempted.count += ONE_ATTEMPT;
    if (!member) return;
    await member.done;
    const errorCode = await this.repository.workError(userId, member.id);
    if (errorCode) failures.push({ section: 'membership', errorCode });
  }

  private async refreshAccounts(
    userId: string,
    revision: string,
    accounts: Awaited<ReturnType<TbankAccessService['local']>>['accounts'],
    failures: HourlySection[],
    attempted: { count: number },
  ) {
    for (const account of accounts) {
      await this.refreshPortfolio(userId, revision, account.accountId, failures, attempted);
    }
    for (const account of accounts) {
      await this.refreshOperations(userId, revision, account.accountId, failures, attempted);
    }
    for (const account of accounts) {
      await this.refreshEvents(userId, account.accountId, failures, attempted);
    }
  }

  private async refreshPortfolio(userId: string, revision: string, accountId: string, failures: HourlySection[], attempted: { count: number }) {
    const work = await this.portfolio.request(userId, revision, accountId, true);
    attempted.count += ONE_ATTEMPT;
    if (!work) return;
    await work.done;
    const errorCode = await this.repository.workError(userId, work.id);
    if (errorCode) failures.push({ section: 'portfolio', errorCode, accountId });
  }

  private async refreshOperations(userId: string, revision: string, accountId: string, failures: HourlySection[], attempted: { count: number }) {
    attempted.count += ONE_ATTEMPT;
    try {
      const id = await this.operations.request(userId, [{ accountId, from: monthStartOf(new Date()), to: new Date() }], true, 'screen', revision);
      if (!id) return;
      const errorCode = await this.settledWorkError(userId, [id]);
      if (errorCode) failures.push({ section: 'operations', errorCode, accountId });
    } catch {
      failures.push({ section: 'operations', errorCode: TbankErrorCode.Unavailable, accountId });
    }
  }

  private async refreshEvents(userId: string, accountId: string, failures: HourlySection[], attempted: { count: number }) {
    attempted.count += ONE_ATTEMPT;
    try {
      const ids = await this.events.refresh(userId, accountId);
      const failed = ids.length ? (await this.settledWorks(userId, ids)).find((row) => row.state === 'failed') : undefined;
      if (failed?.errorCode) failures.push({ section: 'events', errorCode: failed.errorCode, accountId });
      else await this.events.get(userId, accountId);
    } catch {
      failures.push({ section: 'events', errorCode: TbankErrorCode.Unavailable, accountId });
    }
  }

  private async settle(
    attempt: TbankSyncAttempt,
    failures: HourlySection[],
    attempted: number,
  ) {
    let result: 'failed' | 'partial' | 'succeeded' = 'succeeded';
    if (failures.length) {
      result = attempted > NO_ATTEMPTS && failures.length >= attempted ? 'failed' : 'partial';
    }
    await this.repository
      .settleHourly(attempt, result, failures)
      .catch(() => undefined);
  }

  private async settledWorkError(userId: string, ids: string[]) {
    const rows = await this.settledWorks(userId, ids);
    return rows.find((row) => row.state === 'failed')?.errorCode ?? null;
  }

  // Polls the durable work rows until they leave the running state; a row
  // deleted with its connection counts as interrupted, not failed.
  private async settledWorks(userId: string, ids: string[]) {
    const settled = new Map<string, SettledWork>();
    const pending = new Set(ids);
    while (pending.size) {
      const rows = await this.repository.works(userId, [...pending]);
      for (const row of rows) {
        if (row.state !== 'running') {
          settled.set(row.id, row);
          pending.delete(row.id);
        }
      }
      for (const id of [...pending]) {
        if (!rows.some((row) => row.id === id)) {
          settled.set(id, { id, state: 'cancelled', errorCode: null });
          pending.delete(id);
        }
      }
      if (pending.size) await new Promise((resolve) => setTimeout(resolve, WORK_POLL_MS));
    }
    return [...settled.values()];
  }
}

function monthStartOf(now: Date) {
  return new Date(
    Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), FIRST_DAY_OF_MONTH),
  );
}