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

import { StructuredLogger } from '../../../common/logging/structured-logger.service';
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 {
  HOURLY_CAUSE_INTERRUPTED,
  type HourlyCause,
  type HourlyChild,
  type HourlySection,
  TbankSyncRepository,
} from './tbank-sync.repository';

const HOUR_MS = 3_600_000;
const TICK_MS = 60_000;
const WORK_POLL_MS = 100;
const WORK_POLL_BACKOFF_FACTOR = 2;
const WORK_POLL_MAX_MS = 5_000;
// ponytail: fixed 15-minute step wait bound (900_000 ms), no config knob;
// make it configurable if slow steps become regular.
const WORK_WAIT_BOUND_MS = 900_000;
// An await that outlived WORK_WAIT_BOUND_MS. The underlying call keeps
// running; its late durable result is fenced like any other late publication.
const BOUND_EXPIRED = Symbol('hourly-bound-expired');
const FIRST_DAY_OF_MONTH = 1;
const NO_MILLISECONDS = 0;

type DispatchStep = (child: HourlyChild) => Promise<void>;
type Accounts = Awaited<ReturnType<TbankAccessService['local']>>['accounts'];

// 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();
  // Attempt ids of walks currently executing in this process; a running row
  // outside this set lost its settle and is closed by recovery.
  private readonly active = new Set<string>();
  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,
    private readonly logger: StructuredLogger,
  ) {}

  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) {
    await this.recoverAbandoned(userId);
    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;
  }

  // A running hourly row of this process that has no active walk lost its
  // settle (suppressed persistence error). Close it from its durable step
  // results; rows of other processes are closed by the reservation restart
  // branch. Rides the existing tick: no new timers or sanitors.
  private async recoverAbandoned(userId: string) {
    const rows = await this.repository.runningHourly(userId);
    for (const row of rows) {
      if (row.processRunId !== this.processRunId) continue;
      if (this.active.has(row.attemptId)) continue;
      await this.repository
        .settleHourly(
          {
            userId,
            revision: row.revision,
            id: row.id,
            attemptId: row.attemptId,
          },
          HOURLY_CAUSE_INTERRUPTED,
        )
        .catch(() => undefined);
    }
  }

  // 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. The attempt outcome is derived from durable
  // step rows by the repository, never from walk-local state.
  private async walk(attempt: TbankSyncAttempt) {
    this.active.add(attempt.attemptId);
    try {
      const { userId, revision } = attempt;
      const children: HourlyChild[] = [];
      // Required sections whose dispatch failed never get a durable step
      // row; they are recorded as problems so the attempt cannot succeed.
      const dispatchProblems: HourlySection[] = [];
      const dispatch: DispatchStep = async (child) => {
        children.push(child);
        await this.repository.recordHourlyChild(attempt, child);
      };
      const accounts = await this.refreshMembership(
        attempt,
        dispatch,
        dispatchProblems,
      );
      if (!accounts) return;
      await this.refreshAccounts(
        userId,
        revision,
        accounts,
        dispatch,
        dispatchProblems,
      );
      await this.waitWorks(userId, children.map((child) => child.id));
      for (const account of accounts) {
        const pending = this.events
          .get(userId, account.accountId)
          .catch(() => undefined);
        await this.withBound(pending);
      }
      await this.settle(attempt, null, dispatchProblems);
    } catch {
      await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED);
    } finally {
      this.active.delete(attempt.attemptId);
    }
  }

  // Returns the current accounts, or null when the attempt is already
  // settled (invalid token or a replaced connection revision).
  private async refreshMembership(
    attempt: TbankSyncAttempt,
    dispatch: DispatchStep,
    dispatchProblems: HourlySection[],
  ) {
    const { userId, revision } = attempt;
    const member = await this.withBound(
      this.membership.request(userId, revision, true),
    );
    if (member && member !== BOUND_EXPIRED) {
      await dispatch({ id: member.id, section: 'membership' });
      const done = await this.withBound(member.done);
      if (
        done !== BOUND_EXPIRED &&
        (await this.repository.workError(userId, member.id)) ===
        TbankErrorCode.InvalidToken
      ) {
        await this.settle(attempt);
        return null;
      }
    } else {
      // No durable step id (rejected request or an expired wait): record the
      // observable problem; a late row stays fenced by its durable attempt id.
      dispatchProblems.push({
        section: 'membership',
        errorCode: TbankErrorCode.Unavailable,
      });
    }
    const current = await this.withBound(this.access.local(userId));
    if (current === BOUND_EXPIRED || current.revision !== revision) {
      await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED);
      return null;
    }
    return current.accounts;
  }

  // Races a step await against the same per-step bound the row wait uses, so
  // a dispatch that never resolves cannot hold the attempt running forever.
  // The call itself is not cancelled; a rejection landing after the expiry is
  // swallowed here because the expiry already accounts the step as a problem.
  private withBound<T>(step: Promise<T>): Promise<T | typeof BOUND_EXPIRED> {
    step.catch(() => undefined);
    return new Promise<T | typeof BOUND_EXPIRED>((resolve, reject) => {
      const timer = setTimeout(
        () => resolve(BOUND_EXPIRED),
        WORK_WAIT_BOUND_MS,
      );
      step.then(
        (value) => {
          clearTimeout(timer);
          resolve(value);
        },
        (error: unknown) => {
          clearTimeout(timer);
          reject(error instanceof Error ? error : new Error(String(error)));
        },
      );
    });
  }

  private async refreshAccounts(
    userId: string,
    revision: string,
    accounts: Accounts,
    dispatch: DispatchStep,
    dispatchProblems: HourlySection[],
  ) {
    for (const account of accounts) {
      await this.dispatchPortfolio(
        userId,
        revision,
        account.accountId,
        dispatch,
        dispatchProblems,
      );
    }
    for (const account of accounts) {
      await this.dispatchOperations(
        userId,
        revision,
        account.accountId,
        dispatch,
        dispatchProblems,
      );
    }
    for (const account of accounts) {
      await this.dispatchEvents(
        userId,
        account.accountId,
        dispatch,
        dispatchProblems,
      );
    }
  }

  private async dispatchPortfolio(
    userId: string,
    revision: string,
    accountId: string,
    dispatch: DispatchStep,
    dispatchProblems: HourlySection[],
  ) {
    try {
      const work = await this.withBound(
        this.portfolio.request(userId, revision, accountId, true),
      );
      if (work && work !== BOUND_EXPIRED) {
        await dispatch({ accountId, id: work.id, section: 'portfolio' });
        return;
      }
    } catch {
      // A failed dispatch has no durable step row; record the problem.
    }
    dispatchProblems.push({
      accountId,
      errorCode: TbankErrorCode.Unavailable,
      section: 'portfolio',
    });
  }

  private async dispatchOperations(
    userId: string,
    revision: string,
    accountId: string,
    dispatch: DispatchStep,
    dispatchProblems: HourlySection[],
  ) {
    try {
      const id = await this.withBound(
        this.operations.request(
          userId,
          [
            {
              accountId,
              from: monthStartOf(new Date()),
              to: new Date(),
            },
          ],
          true,
          'screen',
          revision,
        ),
      );
      if (id && id !== BOUND_EXPIRED) {
        await dispatch({ accountId, id, section: 'operations' });
        return;
      }
    } catch {
      // A failed dispatch has no durable step row; record the problem.
    }
    dispatchProblems.push({
      accountId,
      errorCode: TbankErrorCode.Unavailable,
      section: 'operations',
    });
  }

  private async dispatchEvents(
    userId: string,
    accountId: string,
    dispatch: DispatchStep,
    dispatchProblems: HourlySection[],
  ) {
    try {
      const ids = await this.withBound(this.events.refresh(userId, accountId));
      if (ids !== BOUND_EXPIRED) {
        for (const id of ids) {
          await dispatch({ accountId, id, section: 'events' });
        }
        return;
      }
    } catch {
      // A failed dispatch has no durable step row; record the problem.
    }
    dispatchProblems.push({
      accountId,
      errorCode: TbankErrorCode.Unavailable,
      section: 'events',
    });
  }

  private async settle(
    attempt: TbankSyncAttempt,
    cause: HourlyCause | null = null,
    dispatchProblems: HourlySection[] = [],
  ) {
    // A suppressed settle error is not lost: the walk leaves the active set
    // and the next tick closes the attempt from its durable step results.
    await this.repository
      .settleHourly(attempt, cause, dispatchProblems)
      .catch(() => {
        this.logger.event('error', 'tbank.hourly.settle_failed', {
          userId: attempt.userId,
        });
        return undefined;
      });
  }

  // Drops settled and missing rows from the wait set and expires waits that
  // passed the per-step bound.
  private advanceWaiting(
    waiting: Map<string, number>,
    rows: { id: string; state: string }[],
  ) {
    for (const row of rows) {
      if (row.state !== 'running') waiting.delete(row.id);
    }
    for (const id of [...waiting.keys()]) {
      if (!rows.some((row) => row.id === id)) {
        waiting.delete(id);
        continue;
      }
      const startedAt = waiting.get(id) ?? NO_MILLISECONDS;
      if (Date.now() - startedAt >= WORK_WAIT_BOUND_MS) waiting.delete(id);
    }
  }

  // Waits for every dispatched step to leave the running state. Each wait is
  // bounded so a stuck step cannot hold the attempt running forever; the
  // bound does not cancel the upstream call, and late results stay in the
  // step row where the attempt fence decides whether they count.
  private async waitWorks(userId: string, ids: string[]) {
    const waiting = new Map(ids.map((id) => [id, Date.now()]));
    let pollMs = WORK_POLL_MS;
    while (waiting.size) {
      const rows = await this.repository.works(userId, [...waiting.keys()]);
      this.advanceWaiting(waiting, rows);
      if (waiting.size) {
        await new Promise((resolve) => setTimeout(resolve, pollMs));
        pollMs = Math.min(pollMs * WORK_POLL_BACKOFF_FACTOR, WORK_POLL_MAX_MS);
      }
    }
  }
}

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