feat(parcel-sync): add ANCPI ePay CF extract ordering backend
Foundation (Phase 1): - CfExtract Prisma model with version tracking, expiry, MinIO path - epay-types.ts: all ePay API response types - epay-counties.ts: WORKSPACE_ID → ePay county index mapping (42 counties) - epay-storage.ts: MinIO helpers (bucket, naming, upload, download) - docker-compose.yml: ANCPI env vars ePay Client (Phase 2): - epay-client.ts: full HTTP client (login, credits, cart, search estate, submit order, poll status, download PDF) with cookie jar + auto-relogin - epay-session-store.ts: separate session from eTerra Queue + API (Phase 3): - epay-queue.ts: sequential FIFO queue (global cart constraint), 10-step workflow per order with DB status updates at each step - POST /api/ancpi/session: connect/disconnect - POST /api/ancpi/order: create single or bulk orders - GET /api/ancpi/orders: list all extracts - GET /api/ancpi/credits: live credit balance - GET /api/ancpi/download: stream PDF from MinIO Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
AI Assistant
@@ -0,0 +1,302 @@
|
||||
/**
|
||||
* ePay sequential order queue.
|
||||
*
|
||||
* ePay has a GLOBAL cart per account — only one order can be processed
|
||||
* at a time. This queue ensures sequential execution.
|
||||
*
|
||||
* Flow per item:
|
||||
* 1. Check credits
|
||||
* 2. Add to cart (prodId=14200)
|
||||
* 3. Search estate on ePay
|
||||
* 4. Submit order (EditCartSubmit)
|
||||
* 5. Poll status (15s × 40 = max 10 min)
|
||||
* 6. Download PDF
|
||||
* 7. Store in MinIO
|
||||
* 8. Update DB record
|
||||
*/
|
||||
|
||||
import { prisma } from "@/core/storage/prisma";
|
||||
import { EpayClient } from "./epay-client";
|
||||
import { getEpayCredentials, updateEpayCredits } from "./epay-session-store";
|
||||
import { storeCfExtract } from "./epay-storage";
|
||||
import { resolveEpayCountyIndex } from "./epay-counties";
|
||||
import type { CfExtractCreateInput } from "./epay-types";
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Types */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
type QueueItem = {
|
||||
extractId: string; // CfExtract.id in DB
|
||||
input: CfExtractCreateInput;
|
||||
};
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Global singleton queue */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
const g = globalThis as {
|
||||
__epayQueue?: QueueItem[];
|
||||
__epayQueueProcessing?: boolean;
|
||||
};
|
||||
if (!g.__epayQueue) g.__epayQueue = [];
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Public API */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
/**
|
||||
* Enqueue a CF extract order. Creates a DB record and adds to the
|
||||
* processing queue. Returns the CfExtract.id immediately.
|
||||
*/
|
||||
export async function enqueueOrder(
|
||||
input: CfExtractCreateInput,
|
||||
): Promise<string> {
|
||||
// Create DB record in "queued" status
|
||||
const record = await prisma.cfExtract.create({
|
||||
data: {
|
||||
nrCadastral: input.nrCadastral,
|
||||
nrCF: input.nrCF ?? input.nrCadastral,
|
||||
siruta: input.siruta,
|
||||
judetIndex: input.judetIndex,
|
||||
judetName: input.judetName,
|
||||
uatId: input.uatId,
|
||||
uatName: input.uatName,
|
||||
gisFeatureId: input.gisFeatureId,
|
||||
prodId: input.prodId ?? 14200,
|
||||
status: "queued",
|
||||
// Version: find max version for this cadastral number and increment
|
||||
version:
|
||||
((
|
||||
await prisma.cfExtract.aggregate({
|
||||
where: { nrCadastral: input.nrCadastral },
|
||||
_max: { version: true },
|
||||
})
|
||||
)._max.version ?? 0) + 1,
|
||||
},
|
||||
});
|
||||
|
||||
g.__epayQueue!.push({ extractId: record.id, input });
|
||||
|
||||
console.log(
|
||||
`[epay-queue] Enqueued: ${input.nrCadastral} (id=${record.id}, queue=${g.__epayQueue!.length})`,
|
||||
);
|
||||
|
||||
// Start processing if not already running
|
||||
void processQueue();
|
||||
|
||||
return record.id;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue multiple orders at once (bulk).
|
||||
*/
|
||||
export async function enqueueBulk(
|
||||
inputs: CfExtractCreateInput[],
|
||||
): Promise<string[]> {
|
||||
const ids: string[] = [];
|
||||
for (const input of inputs) {
|
||||
const id = await enqueueOrder(input);
|
||||
ids.push(id);
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get queue status for UI display.
|
||||
*/
|
||||
export function getQueueStatus(): {
|
||||
length: number;
|
||||
processing: boolean;
|
||||
currentItem?: string;
|
||||
} {
|
||||
return {
|
||||
length: g.__epayQueue?.length ?? 0,
|
||||
processing: g.__epayQueueProcessing ?? false,
|
||||
};
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Queue Processor */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
async function processQueue(): Promise<void> {
|
||||
if (g.__epayQueueProcessing) return; // already running
|
||||
g.__epayQueueProcessing = true;
|
||||
|
||||
try {
|
||||
while (g.__epayQueue && g.__epayQueue.length > 0) {
|
||||
const item = g.__epayQueue.shift()!;
|
||||
await processItem(item);
|
||||
}
|
||||
} finally {
|
||||
g.__epayQueueProcessing = false;
|
||||
}
|
||||
}
|
||||
|
||||
async function updateStatus(
|
||||
id: string,
|
||||
status: string,
|
||||
extra?: Record<string, unknown>,
|
||||
): Promise<void> {
|
||||
await prisma.cfExtract.update({
|
||||
where: { id },
|
||||
data: { status, ...extra },
|
||||
});
|
||||
}
|
||||
|
||||
async function processItem(item: QueueItem): Promise<void> {
|
||||
const { extractId, input } = item;
|
||||
|
||||
try {
|
||||
// Get ePay credentials
|
||||
const creds = getEpayCredentials();
|
||||
if (!creds) {
|
||||
await updateStatus(extractId, "failed", {
|
||||
errorMessage: "Nu ești conectat la ePay.",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const client = await EpayClient.create(creds.username, creds.password);
|
||||
|
||||
// Step 1: Check credits
|
||||
const credits = await client.getCredits();
|
||||
updateEpayCredits(credits);
|
||||
if (credits < 1) {
|
||||
await updateStatus(extractId, "failed", {
|
||||
errorMessage: `Credite insuficiente: ${credits}. Reîncărcați contul pe epay.ancpi.ro.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 2: Add to cart
|
||||
await updateStatus(extractId, "cart");
|
||||
const basketRowId = await client.addToCart(input.prodId ?? 14200);
|
||||
await updateStatus(extractId, "cart", { basketRowId });
|
||||
|
||||
// Step 3: Search estate on ePay
|
||||
await updateStatus(extractId, "searching");
|
||||
const results = await client.searchEstate(
|
||||
input.nrCadastral,
|
||||
input.judetIndex,
|
||||
input.uatId,
|
||||
);
|
||||
|
||||
if (results.length === 0) {
|
||||
await updateStatus(extractId, "failed", {
|
||||
errorMessage: `Imobilul ${input.nrCadastral} nu a fost găsit în baza ANCPI.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const estate = results[0]!;
|
||||
await updateStatus(extractId, "searching", {
|
||||
immovableId: estate.immovableId,
|
||||
immovableType: estate.immovableTypeCode,
|
||||
measuredArea: estate.measureadArea,
|
||||
legalArea: estate.legalArea,
|
||||
address: estate.address,
|
||||
});
|
||||
|
||||
// Step 4: Submit order
|
||||
await updateStatus(extractId, "ordering");
|
||||
const nrCF = input.nrCF ?? estate.electronicIdentifier ?? input.nrCadastral;
|
||||
const orderId = await client.submitOrder({
|
||||
basketRowId,
|
||||
judetIndex: input.judetIndex,
|
||||
uatId: input.uatId,
|
||||
nrCF,
|
||||
nrCadastral: input.nrCadastral,
|
||||
solicitantId:
|
||||
process.env.ANCPI_DEFAULT_SOLICITANT_ID || "14452",
|
||||
});
|
||||
|
||||
await updateStatus(extractId, "polling", { orderId });
|
||||
|
||||
// Step 5: Poll until complete
|
||||
const finalStatus = await client.pollUntilComplete(
|
||||
orderId,
|
||||
async (attempt, status) => {
|
||||
await updateStatus(extractId, "polling", {
|
||||
epayStatus: status,
|
||||
pollAttempts: attempt,
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
if (
|
||||
finalStatus.status === "Anulata" ||
|
||||
finalStatus.status === "Plata refuzata"
|
||||
) {
|
||||
await updateStatus(extractId, "cancelled", {
|
||||
epayStatus: finalStatus.status,
|
||||
errorMessage: `Comanda ${finalStatus.status.toLowerCase()}.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 6: Download PDF
|
||||
const doc = finalStatus.documents.find(
|
||||
(d) => d.downloadValabil && d.contentType === "application/pdf",
|
||||
);
|
||||
if (!doc) {
|
||||
await updateStatus(extractId, "failed", {
|
||||
epayStatus: finalStatus.status,
|
||||
errorMessage: "Nu s-a găsit documentul PDF în comanda finalizată.",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
await updateStatus(extractId, "downloading", {
|
||||
idDocument: doc.idDocument,
|
||||
documentName: doc.nume,
|
||||
documentDate: doc.dataDocument ? new Date(doc.dataDocument) : null,
|
||||
});
|
||||
|
||||
const pdfBuffer = await client.downloadDocument(doc.idDocument, 4);
|
||||
|
||||
// Step 7: Store in MinIO
|
||||
const { path, index } = await storeCfExtract(
|
||||
pdfBuffer,
|
||||
input.nrCadastral,
|
||||
{
|
||||
"ancpi-order-id": orderId,
|
||||
"nr-cadastral": input.nrCadastral,
|
||||
judet: input.judetName,
|
||||
uat: input.uatName,
|
||||
"data-document": doc.dataDocument ?? "",
|
||||
stare: finalStatus.status,
|
||||
produs: "EXI_ONLINE",
|
||||
},
|
||||
);
|
||||
|
||||
// Step 8: Complete
|
||||
const documentDate = doc.dataDocument
|
||||
? new Date(doc.dataDocument)
|
||||
: new Date();
|
||||
const expiresAt = new Date(documentDate);
|
||||
expiresAt.setDate(expiresAt.getDate() + 30);
|
||||
|
||||
await updateStatus(extractId, "completed", {
|
||||
minioPath: path,
|
||||
minioIndex: index,
|
||||
epayStatus: finalStatus.status,
|
||||
completedAt: new Date(),
|
||||
documentDate,
|
||||
expiresAt,
|
||||
});
|
||||
|
||||
// Update credits after successful order
|
||||
const newCredits = await client.getCredits();
|
||||
updateEpayCredits(newCredits);
|
||||
|
||||
console.log(
|
||||
`[epay-queue] Completed: ${input.nrCadastral} → ${path} (credits: ${newCredits})`,
|
||||
);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : "Eroare necunoscută";
|
||||
console.error(`[epay-queue] Failed: ${input.nrCadastral}:`, message);
|
||||
await updateStatus(extractId, "failed", { errorMessage: message });
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user