backend/src/modules/investments/tbank-sync/tbank-sync.repository.ts

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

import type { Prisma } from '../../../../generated/prisma/client';
import type { TbankSyncAttempt } from '../../../integrations/tbank/tbank.api';
import type { InvestmentAttemptDto } from '../investment-resource.dto';
import type { InvestmentScreenDto } from './tbank-sync.dto';

import { claimSyncWork, lockConnectionGeneration } from '../../../integrations/tbank/tbank-sync-claim';
import { PrismaService } from '../../../prisma/prisma.service';
import { toJson } from '../../../prisma/to-json';
export type ScreenIntent = { now: string; scope: InvestmentScreenDto };

// Json columns are written only by this repository; a shape check turns corruption into an explicit 500.
export function screenIntent(value: Prisma.JsonValue): ScreenIntent {
  const intent = value as null | Partial<ScreenIntent>;
  if (typeof intent?.now !== 'string' || typeof intent.scope?.screen !== 'string') {
    throw new Error('Corrupted screen sync intent');
  }
  return intent as ScreenIntent;
}
export type ScreenProgress = {
  attempts?: Record<string, string>;
  children: string[];
  dispatched: boolean;
};
// Progress columns are written only by this repository; a shape check turns
// corruption into an explicit 500. Legacy rows predate optional fields, so
// only the fields every writer has always written are required.
function validAttemptMap(attempts: unknown): boolean {
  return (
    attempts === undefined ||
    (typeof attempts === 'object' &&
      attempts !== null &&
      !Array.isArray(attempts) &&
      Object.values(attempts).every((id) => typeof id === 'string'))
  );
}

export function screenProgress(value: Prisma.JsonValue): ScreenProgress {
  const progress = value as null | Partial<ScreenProgress>;
  if (
    typeof progress?.dispatched !== 'boolean' ||
    !Array.isArray(progress.children) ||
    !progress.children.every((child) => typeof child === 'string') ||
    !validAttemptMap(progress.attempts)
  ) {
    throw new Error('Corrupted screen sync progress');
  }
  return progress as ScreenProgress;
}
export type HourlySection = {
  accountId?: string;
  errorCode: null | string;
  section: 'events' | 'membership' | 'operations' | 'portfolio';
};
export type HourlyChild = {
  accountId?: string;
  id: string;
  section: HourlySection['section'];
};
export const HOURLY_CAUSE_DEADLINE = 'DEADLINE';
export const HOURLY_CAUSE_INTERRUPTED = 'INTERRUPTED';
export type HourlyCause = typeof HOURLY_CAUSE_INTERRUPTED;
type StepOutcome = { problem?: HourlySection; success: boolean };
type StepRow = { attemptId: string; errorCode: null | string; state: string };
export type HourlyProgress = {
  attempts?: Record<string, string>;
  children?: HourlyChild[];
  result?: 'failed' | 'partial' | 'succeeded';
  sections: HourlySection[];
};
// Same tolerance rule as screenProgress: required fields are the ones every
// hourly writer has always written; optional collections may be absent.
export function hourlyProgress(value: Prisma.JsonValue): HourlyProgress {
  const progress: unknown = value;
  if (typeof progress !== 'object' || progress === null) {
    throw new Error('Corrupted hourly sync progress');
  }
  const row = progress as Record<string, unknown>;
  const validChildren =
    row.children === undefined ||
    (Array.isArray(row.children) &&
      row.children.every((child) => {
        if (typeof child !== 'object' || child === null) return false;
        const entry = child as Record<string, unknown>;
        return typeof entry.id === 'string' && typeof entry.section === 'string';
      }));
  const validSections =
    row.sections === undefined ||
    (Array.isArray(row.sections) &&
      row.sections.every((section) => {
        if (typeof section !== 'object' || section === null) return false;
        const entry = section as Record<string, unknown>;
        return (
          typeof entry.section === 'string' &&
          (typeof entry.errorCode === 'string' || entry.errorCode === null)
        );
      }));
  if (
    !validSections ||
    !validAttemptMap(row.attempts) ||
    !validChildren
  ) {
    throw new Error('Corrupted hourly sync progress');
  }
  return progress as HourlyProgress;
}
type HourlyAttempt = {
  finishedAt: Date | null;
  sections: HourlySection[];
  startedAt: Date;
  state: 'cancelled' | 'failed' | 'partial' | 'running' | 'succeeded';
};

export function hourlyAttempt(
  hourly: HourlyAttempt | null,
  section: HourlySection['section'] | readonly HourlySection['section'][],
  accountId?: string,
): InvestmentAttemptDto {
  if (!hourly) {
    return { state: 'none', startedAt: null, finishedAt: null, problems: [] };
  }
  const sections = Array.isArray(section) ? section : [section];
  const problems = hourly.sections
    .filter(
      (entry) =>
        sections.includes(entry.section) &&
        (accountId === undefined || entry.accountId === accountId),
    )
    .map((entry) => ({
      section: entry.section,
      accountId: entry.accountId ?? null,
      errorCode: entry.errorCode,
    }));
  return {
    state: hourly.state === 'cancelled' ? 'failed' : hourly.state,
    startedAt: hourly.startedAt.toISOString(),
    finishedAt: hourly.finishedAt?.toISOString() ?? null,
    problems,
  };
}

@Injectable()
export class TbankSyncRepository {
  constructor(private readonly prisma: PrismaService) {}
  reserve(
    userId: string,
    revision: string,
    scope: InvestmentScreenDto,
    runId: string,
  ) {
    return this.prisma.$transaction(async (tx) => {
      const valid = await lockConnectionGeneration(tx, { userId, revision });
      if (!valid) return null;
      const scopeKey = `screen:${JSON.stringify(scope)}`;
      const key = { userId, revision, scopeKey };
      const previous = await tx.tbankSyncWork.findUnique({
        where: { userId_revision_scopeKey: key },
      });
      if (previous?.state === 'running' && previous.processRunId === runId) {
        const progress = screenProgress(previous.progress);
        const children = progress.children ?? [];
        if (
          !progress.dispatched ||
          (await tx.tbankSyncWork.count({
            where: { userId, revision, id: { in: children }, state: 'running' },
          }))
        ) {
          return { work: previous, launch: false };
        }
      }
      const data = {
        attemptId: randomUUID(),
        processRunId: runId,
        state: 'running',
        priority: 0,
        intent: toJson({ scope, now: new Date().toISOString() }),
        progress: { children: [], dispatched: false },
        errorCode: null,
      };
      const work = await tx.tbankSyncWork.upsert({
        where: { userId_revision_scopeKey: key },
        create: { ...key, ...data, id: randomUUID() },
        update: { ...data, id: randomUUID() },
      });
      return { work, launch: true };
    });
  }

  read(userId: string, revision: string, id: string) {
    return this.prisma.$transaction(async (tx) => {
      const work = await tx.tbankSyncWork.findFirst({
        where: { userId, revision, id },
      });
      if (!work) return null;
      const progress = screenProgress(work.progress);
      const children = await tx.tbankSyncWork.findMany({
        where: { userId, revision, id: { in: progress.children ?? [] } },
        select: { id: true, attemptId: true, state: true, errorCode: true },
      });
      return {
        work,
        children: children.map((child) =>
          progress.attempts?.[child.id] &&
          progress.attempts[child.id] !== child.attemptId
            ? { ...child, state: 'cancelled', errorCode: null }
            : child,
        ),
        progress,
      };
    });
  }

  async progress(
    attempt: TbankSyncAttempt,
    children: string[],
    dispatched: boolean,
    errorCode: null | string = null,
  ) {
    return this.prisma.$transaction(async (tx) => {
      const previous = await tx.tbankSyncWork.findFirst({ where: attempt });
      if (!previous) return { count: 0 };
      const old = screenProgress(previous.progress);
      const rows = await tx.tbankSyncWork.findMany({
        where: {
          userId: attempt.userId,
          revision: attempt.revision,
          id: { in: children },
        },
        select: { id: true, attemptId: true },
      });
      const attempts = {
        ...Object.fromEntries(rows.map((row) => [row.id, row.attemptId])),
        ...old.attempts,
      };
      const claimed = await claimSyncWork(tx, attempt, {
        progress: { children: [...new Set(children)], attempts, dispatched },
        ...(errorCode ? { state: 'failed', errorCode } : {}),
      });
      return { count: claimed };
    });
  }

  // The timer only proposes work; this reservation is the durable fence. The
  // attempt start (createdAt) is the basis of the next hourly term, including
  // failed and interrupted attempts. After downtime the next tick is due at
  // once, so no catch-up queue forms.
  reserveHourly(userId: string, runId: string, hourMs: number) {
    return this.prisma.$transaction(async (tx) => {
      const connection = await tx.tbankConnection.findFirst({
        where: { userId, invalidToken: false },
        select: { revision: true },
      });
      if (!connection) return null;
      const key = {
        userId,
        revision: connection.revision,
        scopeKey: 'hourly',
      };
      const previous = await tx.tbankSyncWork.findUnique({
        where: { userId_revision_scopeKey: key },
      });
      const now = new Date();
      if (previous) {
        // Terms are hourMs steps from the attempt start; a walk still running
        // at a term consumes it. A dead process's attempt has no known end, so
        // it consumes only its own term.
        const started = previous.createdAt.valueOf();
        const ended =
          previous.state === 'running' ? started : previous.updatedAt.valueOf();
        const dueAt =
          started +
          Math.max(hourMs, Math.ceil((ended - started) / hourMs) * hourMs);
        const due = now.valueOf() >= dueAt;
        if (previous.state === 'running') {
          if (previous.processRunId === runId) return null;
          // A dead process cannot finish its attempt; close it from its
          // actual step results, but its start stays the basis of the next
          // term.
          await this.closeHourly(tx, previous, 'INTERRUPTED', []);
          if (!due) return null;
        } else if (!due) return null;
      }
      const data = {
        attemptId: randomUUID(),
        processRunId: runId,
        state: 'running',
        priority: 0,
        intent: { source: 'hourly' },
        progress: { sections: [] },
        errorCode: null,
      };
      const work = await tx.tbankSyncWork.upsert({
        where: { userId_revision_scopeKey: key },
        create: { ...key, ...data, id: randomUUID() },
        update: { ...data, createdAt: now },
      });
      return { work, launch: true };
    });
  }

  private classifyStep(
    child: HourlyChild,
    row: StepRow | undefined,
    recorded: string | undefined,
    cause: HourlyCause | null,
  ): StepOutcome {
    if (row && row.attemptId === recorded && row.state === 'complete') {
      return { success: true };
    }
    const stale = !row || (recorded !== undefined && row.attemptId !== recorded);
    let errorCode: null | string = null;
    if (!stale && row.state === 'failed') errorCode = row.errorCode;
    else if (!stale && cause === HOURLY_CAUSE_INTERRUPTED) {
      errorCode = HOURLY_CAUSE_INTERRUPTED;
    } else if (!stale) errorCode = HOURLY_CAUSE_DEADLINE;
    return {
      problem: { accountId: child.accountId, errorCode, section: child.section },
      success: false,
    };
  }

  private deriveResult(
    cause: HourlyCause | null,
    sections: HourlySection[],
    successes: number,
  ): 'failed' | 'partial' | 'succeeded' {
    if (!cause && !sections.length) return 'succeeded';
    return successes ? 'partial' : 'failed';
  }

  // The outcome is derived from the durable state of the dispatched steps,
  // never from an in-memory error list that an exception path can empty.
  // `cause` marks a close before normal completion and caps the result at
  // `partial`; `dispatchProblems` records required sections that never got a
  // durable step row because their dispatch failed.
  private async closeHourly(
    tx: Prisma.TransactionClient,
    work: {
      attemptId: string;
      id: string;
      progress: Prisma.JsonValue;
      revision: string;
      userId: string;
    },
    cause: HourlyCause | null,
    dispatchProblems: HourlySection[],
  ) {
    const progress = hourlyProgress(work.progress);
    const children = progress.children ?? [];
    const rows = children.length
      ? await tx.tbankSyncWork.findMany({
          where: {
            userId: work.userId,
            id: { in: children.map((child) => child.id) },
          },
          select: { attemptId: true, errorCode: true, id: true, state: true },
        })
      : [];
    const byId = new Map(rows.map((row) => [row.id, row]));
    const outcomes = children.map((child) =>
      this.classifyStep(
        child,
        byId.get(child.id),
        progress.attempts?.[child.id],
        cause,
      ),
    );
    const sections = outcomes
      .flatMap((outcome) => (outcome.problem ? [outcome.problem] : []))
      .concat(dispatchProblems);
    const successes = outcomes.filter((outcome) => outcome.success).length;
    const result = this.deriveResult(cause, sections, successes);
    return tx.tbankSyncWork.updateMany({
      where: { id: work.id, attemptId: work.attemptId, state: 'running' },
      data: {
        state: result === 'failed' ? 'failed' : 'complete',
        errorCode:
          cause ??
          sections.find((section) => section.errorCode)?.errorCode ??
          null,
        progress: {
          attempts: progress.attempts ?? {},
          children,
          result,
          sections,
        },
      },
    });
  }

  async settleHourly(
    attempt: TbankSyncAttempt,
    cause: HourlyCause | null = null,
    dispatchProblems: HourlySection[] = [],
  ) {
    return this.prisma.$transaction(async (tx) => {
      const work = await tx.tbankSyncWork.findFirst({ where: attempt });
      if (!work || work.state !== 'running') return { count: 0 };
      return this.closeHourly(tx, work, cause, dispatchProblems);
    });
  }

  // Records a dispatched step on the attempt before it is awaited, so a lost
  // walk can still derive its outcome from durable rows.
  async recordHourlyChild(attempt: TbankSyncAttempt, child: HourlyChild) {
    return this.prisma.$transaction(async (tx) => {
      const work = await tx.tbankSyncWork.findFirst({
        where: { ...attempt, state: 'running' },
        select: { progress: true },
      });
      if (!work) return false;
      const progress = hourlyProgress(work.progress);
      const row = await tx.tbankSyncWork.findFirst({
        where: { userId: attempt.userId, id: child.id },
        select: { attemptId: true },
      });
      const updated = await claimSyncWork(tx, attempt, {
        progress: {
          attempts: {
            ...progress.attempts,
            ...(row ? { [child.id]: row.attemptId } : {}),
          },
          children: [...(progress.children ?? []), child],
          sections: progress.sections ?? [],
        },
      }, { withoutConnectionFence: true });
      return Boolean(updated);
    });
  }

  runningHourly(userId: string) {
    return this.prisma.tbankSyncWork.findMany({
      where: { userId, scopeKey: 'hourly', state: 'running' },
      select: { attemptId: true, id: true, processRunId: true, revision: true },
    });
  }

  async hourly(userId: string) {
    const connection = await this.prisma.tbankConnection.findFirst({
      where: { userId },
      select: { revision: true, invalidToken: true },
    });
    if (!connection || connection.invalidToken) return null;
    const work = await this.prisma.tbankSyncWork.findUnique({
      where: {
        userId_revision_scopeKey: {
          userId,
          revision: connection.revision,
          scopeKey: 'hourly',
        },
      },
    });
    if (!work) return null;
    const progress = hourlyProgress(work.progress);
    let state: 'cancelled' | 'running' | HourlyProgress['result'];
    if (work.state === 'running') state = 'running';
    else if (work.state === 'cancelled') state = 'cancelled';
    else state = progress.result ?? 'failed';
    return {
      startedAt: work.createdAt,
      finishedAt: work.state === 'running' ? null : work.updatedAt,
      state,
      sections: progress.sections ?? [],
    };
  }

  works(userId: string, ids: string[]) {
    return this.prisma.tbankSyncWork.findMany({
      where: { userId, id: { in: ids } },
      select: { id: true, state: true, errorCode: true },
    });
  }

  async workError(userId: string, id: string) {
    const work = await this.prisma.tbankSyncWork.findFirst({
      where: { userId, id },
      select: { state: true, errorCode: true },
    });
    return work?.state === 'failed' ? work.errorCode : null;
  }
}