tehran_taj/apps/api/src/modules/accounting-sync/accounting-worker.service.ts

import { Injectable, Logger, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
import { AccountingOutbox, Prisma } from '@prisma/client';
import { randomUUID } from 'crypto';
import { envOr } from '../../common/env';
import { PrismaService } from '../../prisma/prisma.service';
import { ActiveProvider, AccountingConfigService, JobName } from './accounting-config.service';
import { AccountingSyncService, NeedsReviewError } from './accounting-sync.service';
import { AccountingNotSupportedError, AccountingProviderError } from './providers';

/** Scheduled pulls (customers, full balances, full products): the safety net. */
const TICK_MS = 30_000;
/** Outbox and change poll, the real-time path: checked this often. */
const FAST_TICK_MS = 2_000;
/** How often the change feed is polled. */
const CHANGES_POLL_MS = 5_000;
/** Long enough for a full customer import; renewed before every step. */
const LEASE_MS = 10 * 60_000;
const LEASE_KEY = 'accounting.lease';
const OUTBOX_BATCH = 25;
const MAX_ATTEMPTS = 8;

type PullJob = Exclude<JobName, 'outbox'>;

export const PULL_FEATURE: Record<PullJob, 'customers' | 'balances' | 'products'> = {
  customers: 'customers',
  balances: 'balances',
  products: 'products',
};

function toJson(v: unknown): Prisma.InputJsonValue | typeof Prisma.JsonNull {
  if (v === undefined || v === null) return Prisma.JsonNull;
  return JSON.parse(JSON.stringify(v)) as Prisma.InputJsonValue;
}

/**
 * Runs the accounting integration in the background (D27, plan §3.4). Two
 * loops: a fast one (every couple of seconds) for the outbox and the change
 * feed, the real-time path; and the 30-second one for the scheduled pulls that
 * reconcile what the feed cannot see (new customers, edits and deletions,
 * overflow).
 *
 * In-process, no queue dependency. A lease row in the Setting table
 * (`accounting.lease`) keeps a second API instance from running the same jobs
 * at the same time; the in-memory flag keeps one instance's ticks from
 * overlapping. Set ACCOUNTING_WORKER=off to keep an instance out entirely.
 */
@Injectable()
export class AccountingWorkerService implements OnApplicationBootstrap, OnModuleDestroy {
  private readonly logger = new Logger('Accounting');
  private readonly owner = randomUUID();
  private timer: NodeJS.Timeout | null = null;
  private running = false;
  private fastRunning = false;
  private fastTimer: NodeJS.Timeout | null = null;
  private kickTimer: NodeJS.Timeout | null = null;
  private changes: { provider: string; cursor: string | null } = { provider: '', cursor: null };
  private lastChangePollAt = 0;
  private lastChangeWarnAt = 0;
  private forced = new Set<PullJob>();
  private lastForcedCustomersAt = 0;
  /** Shown on the admin screen. */
  readonly realtime: {
    lastPollAt?: string;
    lastOkAt?: string;
    lastError?: string | null;
    lastChangeAt?: string;
    lastSummary?: Record<string, unknown>;
  } = {};
  private recovered = new Set<string>();

  constructor(
    private readonly prisma: PrismaService,
    private readonly config: AccountingConfigService,
    private readonly sync: AccountingSyncService,
  ) {}

  onApplicationBootstrap() {
    if (envOr('ACCOUNTING_WORKER', 'on') === 'off' || process.env.NODE_ENV === 'test') return;
    this.timer = setInterval(() => void this.tick(), TICK_MS);
    this.timer.unref?.();
    this.fastTimer = setInterval(() => void this.fastTick(), FAST_TICK_MS);
    this.fastTimer.unref?.();
  }

  onModuleDestroy() {
    if (this.timer) clearInterval(this.timer);
    if (this.fastTimer) clearInterval(this.fastTimer);
    if (this.kickTimer) clearTimeout(this.kickTimer);
  }

  /**
   * Something was just queued. Hooks run inside the caller's transaction, so
   * wake the outbox a moment later, once it has committed, instead of waiting
   * for the next tick.
   */
  kickSoon() {
    if (this.kickTimer || !this.fastTimer) return;
    this.kickTimer = setTimeout(() => {
      this.kickTimer = null;
      void this.fastTick(true);
    }, 700);
    this.kickTimer.unref?.();
  }

  /** True when this instance holds (or just took) the lease. */
  private async acquireLease(): Promise<boolean> {
    const value = JSON.stringify({ owner: this.owner, until: new Date(Date.now() + LEASE_MS).toISOString() });
    await this.prisma.$executeRaw`
      INSERT INTO "Setting" ("key", "value", "updatedAt") VALUES (${LEASE_KEY}, ${value}::jsonb, now())
      ON CONFLICT ("key") DO NOTHING`;
    const n = await this.prisma.$executeRaw`
      UPDATE "Setting" SET "value" = ${value}::jsonb, "updatedAt" = now()
      WHERE "key" = ${LEASE_KEY}
        AND ("value"->>'owner' = ${this.owner} OR ("value"->>'until')::timestamptz < now())`;
    return n > 0;
  }

  async tick(only?: JobName): Promise<boolean> {
    if (this.running) return false;
    this.running = true;
    try {
      if (!(await this.acquireLease())) return false;
      const a = await this.config.active();
      if (!a) return false;

      if (!this.recovered.has(a.provider.name)) {
        await this.recoverInterrupted(a);
        this.recovered.add(a.provider.name);
      }

      if (only === 'outbox') await this.processOutbox(a);
      for (const job of ['balances', 'customers', 'products'] as PullJob[]) {
        if (only && only !== job) continue;
        if (!a.config.features[PULL_FEATURE[job]]) continue;
        const wasForced = this.forced.delete(job);
        if (!only && !wasForced && !(await this.isDue(job, a))) continue;
        await this.acquireLease();
        await this.runPull(job, a);
      }
      return true;
    } catch (err) {
      this.logger.error(`Accounting tick failed: ${(err as Error).message}`);
      return false;
    } finally {
      this.running = false;
    }
  }

  /**
   * The real-time path, every couple of seconds: send queued documents and
   * edits at once, and poll the software's change feed so a balance or a stock
   * figure changed there is changed here moments later. Never waits behind a
   * long scheduled pull (it has its own guard, apart from tick()).
   */
  async fastTick(force = false): Promise<void> {
    if (this.fastRunning) return;
    this.fastRunning = true;
    try {
      if (!(await this.acquireLease())) return;
      const a = await this.config.active();
      if (!a) return;
      if (!this.recovered.has(a.provider.name)) {
        await this.recoverInterrupted(a);
        this.recovered.add(a.provider.name);
      }
      await this.processOutbox(a);

      if (!force && Date.now() - this.lastChangePollAt < CHANGES_POLL_MS) return;
      if (!a.provider.capabilities.changeFeed || !(a.config.features.balances || a.config.features.products)) return;
      this.lastChangePollAt = Date.now();
      await this.pollChanges(a);
    } catch (err) {
      this.logger.error(`Accounting fast tick failed: ${(err as Error).message}`);
    } finally {
      this.fastRunning = false;
    }
  }

  private async pollChanges(a: ActiveProvider) {
    if (this.changes.provider !== a.provider.name) this.changes = { provider: a.provider.name, cursor: null };
    this.realtime.lastPollAt = new Date().toISOString();
    try {
      const baseline = this.changes.cursor === null;
      const r = await this.sync.pollChanges(a, this.changes.cursor);
      this.changes.cursor = r.cursor;
      this.realtime.lastOkAt = new Date().toISOString();
      this.realtime.lastError = null;
      if (r.balances > 0 || r.stocks > 0) {
        this.realtime.lastChangeAt = this.realtime.lastOkAt;
        this.realtime.lastSummary = { balances: r.balances, stocks: r.stocks };
      }
      if (baseline) return;
      // What the feed cannot describe goes to the scheduled jobs right away.
      if (r.fullBalances) this.forced.add('balances');
      if (r.fullProducts) this.forced.add('products');
      if (r.newAccounts && a.config.features.customers && Date.now() - this.lastForcedCustomersAt > 60_000) {
        this.lastForcedCustomersAt = Date.now();
        this.forced.add('customers');
      }
      if (this.forced.size > 0) void this.tick();
    } catch (err) {
      const msg = this.sync.describeError(err);
      this.realtime.lastError = msg;
      if (Date.now() - this.lastChangeWarnAt > 60_000) {
        this.lastChangeWarnAt = Date.now();
        this.logger.warn(`Accounting change poll failed: ${msg}`);
      }
    }
  }

  /**
   * A sync button (plan §12): runs these pulls now and waits for them. Needs
   * only a connected provider — the feature switches do not apply. Each job
   * reports on its own, so one failing does not hide the others.
   */
  async runPullsNow(jobs: PullJob[]): Promise<Record<string, { ok: boolean; summary?: unknown; error?: string }> | null> {
    if (this.running) return null;
    this.running = true;
    try {
      const a = await this.config.active();
      if (!a) return null;
      const out: Record<string, { ok: boolean; summary?: unknown; error?: string }> = {};
      for (const job of jobs) {
        await this.acquireLease();
        const startedAt = new Date().toISOString();
        await this.config.recordJob(job, { lastRunAt: startedAt });
        try {
          const summary =
            job === 'customers'
              ? await this.sync.pullCustomers(a)
              : job === 'balances'
                ? await this.sync.pullBalances(a)
                : await this.sync.pullProducts(a);
          await this.config.recordJob(job, { lastOkAt: new Date().toISOString(), lastError: null, summary });
          out[job] = { ok: true, summary };
        } catch (err) {
          const error = this.sync.describeError(err);
          await this.config.recordJob(job, { lastError: error });
          out[job] = { ok: false, error };
        }
      }
      return out;
    } finally {
      this.running = false;
    }
  }

  /** «سینک الان» from the admin screen: runs in the background, returns at once. */
  runNow(job: JobName): boolean {
    if (job === 'outbox') {
      if (this.fastRunning) return false;
      void this.fastTick(true);
      return true;
    }
    if (this.running) return false;
    void this.tick(job);
    return true;
  }

  get busy() {
    return this.running;
  }

  private async isDue(job: PullJob, a: ActiveProvider) {
    const minutes = {
      customers: a.config.intervals.customersMinutes,
      balances: a.config.intervals.balancesMinutes,
      products: a.config.intervals.productsMinutes,
    }[job];
    const state = await this.config.getState();
    const last = state.jobs[job]?.lastRunAt;
    return !last || Date.now() - new Date(last).getTime() >= minutes * 60_000;
  }

  private async runPull(job: PullJob, a: ActiveProvider) {
    const startedAt = new Date().toISOString();
    await this.config.recordJob(job, { lastRunAt: startedAt });
    try {
      const summary =
        job === 'customers'
          ? await this.sync.pullCustomers(a)
          : job === 'balances'
            ? await this.sync.pullBalances(a)
            : await this.sync.pullProducts(a);
      await this.config.recordJob(job, { lastOkAt: new Date().toISOString(), lastError: null, summary });
    } catch (err) {
      const msg = this.sync.describeError(err);
      this.logger.warn(`Accounting ${job} pull failed: ${msg}`);
      await this.config.recordJob(job, { lastError: msg });
    }
  }

  /**
   * A job left in SENDING means the process stopped mid-call: its outcome is
   * unknown. Safe kinds go back to the queue; a Zarrin document goes to a
   * person, since sending it again could post it twice.
   */
  private async recoverInterrupted(a: ActiveProvider) {
    const stuck = await this.prisma.accountingOutbox.findMany({ where: { provider: a.provider.name, status: 'SENDING' } });
    for (const job of stuck) {
      const safe = this.sync.isRetrySafe(a.provider.name, job.kind);
      await this.prisma.accountingOutbox.update({
        where: { id: job.id },
        data: safe
          ? { status: 'PENDING', nextAttemptAt: new Date() }
          : {
              status: 'NEEDS_REVIEW',
              lastError: 'سرور هنگام ارسال متوقف شد؛ معلوم نیست سند ثبت شده یا نه. پیش از «تلاش دوباره» در حسابداری بررسی کنید.',
            },
      });
    }
  }

  private async processOutbox(a: ActiveProvider) {
    const jobs = await this.prisma.accountingOutbox.findMany({
      where: { provider: a.provider.name, status: 'PENDING', nextAttemptAt: { lte: new Date() } },
      orderBy: { createdAt: 'asc' },
      take: OUTBOX_BATCH,
    });
    const startedAt = new Date().toISOString();
    let sent = 0;
    let failed = 0;
    for (const job of jobs) {
      await this.acquireLease();
      const ok = await this.runOne(a, job);
      if (ok) sent += 1;
      else failed += 1;
    }
    if (jobs.length > 0) {
      await this.config.recordJob('outbox', { lastRunAt: startedAt, lastOkAt: new Date().toISOString(), summary: { picked: jobs.length, sent, failed } });
    }
  }

  private async runOne(a: ActiveProvider, job: AccountingOutbox): Promise<boolean> {
    const attempts = job.attempts + 1;
    await this.prisma.accountingOutbox.update({ where: { id: job.id }, data: { status: 'SENDING', attempts } });
    const fresh = { ...job, attempts };
    try {
      const res = await this.sync.runJob(a, fresh);
      await this.prisma.accountingOutbox.update({
        where: { id: job.id },
        data: {
          status: 'SENT',
          sentAt: new Date(),
          externalRef: res.externalRef ?? null,
          lastRequest: toJson(res.request),
          lastResponse: toJson(res.response),
          lastError: null,
        },
      });
      return true;
    } catch (err) {
      const msg = this.sync.describeError(err);
      const response = err instanceof AccountingProviderError ? err.response : undefined;
      let data: Prisma.AccountingOutboxUpdateInput;
      if (err instanceof NeedsReviewError || err instanceof AccountingNotSupportedError) {
        data = { status: 'NEEDS_REVIEW', lastError: msg };
      } else if (err instanceof AccountingProviderError && err.ambiguous && !this.sync.isRetrySafe(a.provider.name, job.kind)) {
        data = {
          status: 'NEEDS_REVIEW',
          lastError: `${msg} — ممکن است سند ثبت شده باشد؛ پیش از «تلاش دوباره» در حسابداری بررسی کنید.`,
        };
      } else if (attempts >= MAX_ATTEMPTS) {
        data = { status: 'FAILED', lastError: msg };
      } else {
        const delayMin = Math.min(2 ** (attempts - 1), 60);
        data = { status: 'PENDING', lastError: msg, nextAttemptAt: new Date(Date.now() + delayMin * 60_000) };
      }
      if (response !== undefined) data.lastResponse = toJson(response);
      await this.prisma.accountingOutbox.update({ where: { id: job.id }, data });
      this.logger.warn(`Accounting job ${job.kind} ${job.entityId} → ${String(data.status)}: ${msg}`);
      return false;
    }
  }
}