feat: accounting integration (D27), SMS relay driver (D26), rebuild user apps
- Add Tahesab/Zarrin accounting sync module with admin panel UI - Add SMS relay driver as default provider, refactor SMS into driver pattern - Update customer services, orders, products, and catalog queries - Rebuild user-app and user-app-v2 dist assets - Add Prisma migration for accounting tables - Update docker-compose and env examples - Update parity docs (DECISIONS, ISSUES, PROGRESS) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
s.esmaeili
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
a72fa607a5
commit
19d490e790
@@ -0,0 +1,375 @@
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user