feat(ancpi): batch ordering + download existing orders
Major rewrite: - Queue now processes batches: addToCart×N → saveMetadata×N → ONE submitOrder → poll → download ALL documents → store in MinIO - Removed unique constraint on orderId (shared across batch items) - Added step=download to test endpoint: downloads PDFs from 5 existing orders (9685480-9685484) and stores in MinIO - step=order now uses enqueueBatch for 2 test parcels (61904, 309952) as ONE ePay order instead of separate orders Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
AI Assistant
@@ -1,25 +1,22 @@
|
||||
/**
|
||||
* ePay sequential order queue.
|
||||
* ePay batch 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
|
||||
* BATCH flow (correct — ONE order for N parcels):
|
||||
* 1. Check credits (need >= N)
|
||||
* 2. For each parcel: addToCart → saveMetadata
|
||||
* 3. ONE EditCartSubmit → ONE orderId
|
||||
* 4. Poll that orderId until "Finalizata"
|
||||
* 5. Download ALL documents from that order
|
||||
* 6. Store each in MinIO, update DB records
|
||||
*/
|
||||
|
||||
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";
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
@@ -29,6 +26,11 @@ import type { CfExtractCreateInput } from "./epay-types";
|
||||
type QueueItem = {
|
||||
extractId: string; // CfExtract.id in DB
|
||||
input: CfExtractCreateInput;
|
||||
basketRowId?: number; // set during cart phase
|
||||
};
|
||||
|
||||
type BatchJob = {
|
||||
items: QueueItem[];
|
||||
};
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
@@ -36,70 +38,77 @@ type QueueItem = {
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
const g = globalThis as {
|
||||
__epayQueue?: QueueItem[];
|
||||
__epayBatchQueue?: BatchJob[];
|
||||
__epayQueueProcessing?: boolean;
|
||||
};
|
||||
if (!g.__epayQueue) g.__epayQueue = [];
|
||||
if (!g.__epayBatchQueue) g.__epayBatchQueue = [];
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Public API */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
/**
|
||||
* Enqueue a CF extract order. Creates a DB record and adds to the
|
||||
* processing queue. Returns the CfExtract.id immediately.
|
||||
* Enqueue a single CF extract order (backwards compatible).
|
||||
* Creates a DB record, wraps as a batch of 1, and adds to the queue.
|
||||
*/
|
||||
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,
|
||||
},
|
||||
});
|
||||
const ids = await enqueueBatch([input]);
|
||||
const first = ids[0];
|
||||
if (!first) throw new Error("enqueueBatch returned empty array");
|
||||
return first;
|
||||
}
|
||||
|
||||
g.__epayQueue!.push({ extractId: record.id, input });
|
||||
/**
|
||||
* Enqueue a batch of CF extract orders.
|
||||
* Creates all DB records, then processes as ONE ePay order.
|
||||
* Returns the CfExtract IDs immediately.
|
||||
*/
|
||||
export async function enqueueBatch(
|
||||
inputs: CfExtractCreateInput[],
|
||||
): Promise<string[]> {
|
||||
if (inputs.length === 0) return [];
|
||||
|
||||
const items: QueueItem[] = [];
|
||||
|
||||
for (const input of inputs) {
|
||||
// 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:
|
||||
((
|
||||
await prisma.cfExtract.aggregate({
|
||||
where: { nrCadastral: input.nrCadastral },
|
||||
_max: { version: true },
|
||||
})
|
||||
)._max.version ?? 0) + 1,
|
||||
},
|
||||
});
|
||||
|
||||
items.push({ extractId: record.id, input });
|
||||
}
|
||||
|
||||
g.__epayBatchQueue!.push({ items });
|
||||
|
||||
console.log(
|
||||
`[epay-queue] Enqueued: ${input.nrCadastral} (id=${record.id}, queue=${g.__epayQueue!.length})`,
|
||||
`[epay-queue] Enqueued batch of ${items.length}: ${items.map((i) => i.input.nrCadastral).join(", ")} (queue=${g.__epayBatchQueue!.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;
|
||||
return items.map((i) => i.extractId);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -108,10 +117,9 @@ export async function enqueueBulk(
|
||||
export function getQueueStatus(): {
|
||||
length: number;
|
||||
processing: boolean;
|
||||
currentItem?: string;
|
||||
} {
|
||||
return {
|
||||
length: g.__epayQueue?.length ?? 0,
|
||||
length: g.__epayBatchQueue?.length ?? 0,
|
||||
processing: g.__epayQueueProcessing ?? false,
|
||||
};
|
||||
}
|
||||
@@ -124,13 +132,13 @@ async function processQueue(): Promise<void> {
|
||||
if (g.__epayQueueProcessing) return; // already running
|
||||
g.__epayQueueProcessing = true;
|
||||
|
||||
// Track all orderIds from this batch to avoid duplicates
|
||||
// Track all orderIds from this session to avoid duplicates
|
||||
const knownOrderIds = new Set<string>();
|
||||
|
||||
try {
|
||||
while (g.__epayQueue && g.__epayQueue.length > 0) {
|
||||
const item = g.__epayQueue.shift()!;
|
||||
const orderId = await processItem(item, knownOrderIds);
|
||||
while (g.__epayBatchQueue && g.__epayBatchQueue.length > 0) {
|
||||
const batch = g.__epayBatchQueue.shift()!;
|
||||
const orderId = await processBatch(batch.items, knownOrderIds);
|
||||
if (orderId) knownOrderIds.add(orderId);
|
||||
}
|
||||
} finally {
|
||||
@@ -149,95 +157,126 @@ async function updateStatus(
|
||||
});
|
||||
}
|
||||
|
||||
async function processItem(
|
||||
item: QueueItem,
|
||||
/**
|
||||
* Process a batch of items as ONE ePay order:
|
||||
* 1. Check credits (>= N)
|
||||
* 2. addToCart + saveMetadata for each
|
||||
* 3. ONE submitOrder
|
||||
* 4. Poll until complete
|
||||
* 5. Download + store ALL documents
|
||||
*/
|
||||
async function processBatch(
|
||||
items: QueueItem[],
|
||||
knownOrderIds: Set<string>,
|
||||
): Promise<string | null> {
|
||||
const { extractId, input } = item;
|
||||
const extractIds = items.map((i) => i.extractId);
|
||||
const count = items.length;
|
||||
|
||||
try {
|
||||
// Get ePay credentials
|
||||
const creds = getEpayCredentials();
|
||||
if (!creds) {
|
||||
await updateStatus(extractId, "failed", {
|
||||
errorMessage: "Nu ești conectat la ePay.",
|
||||
});
|
||||
for (const id of extractIds) {
|
||||
await updateStatus(id, "failed", {
|
||||
errorMessage: "Nu ești conectat la ePay.",
|
||||
});
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
const client = await EpayClient.create(creds.username, creds.password);
|
||||
|
||||
// Step 1: Check credits
|
||||
// Step 1: Check credits (need >= count)
|
||||
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.`,
|
||||
});
|
||||
if (credits < count) {
|
||||
for (const id of extractIds) {
|
||||
await updateStatus(id, "failed", {
|
||||
errorMessage: `Credite insuficiente: ${credits} disponibile, ${count} necesare.`,
|
||||
});
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// Step 2: Add to cart
|
||||
await updateStatus(extractId, "cart");
|
||||
const basketRowId = await client.addToCart(input.prodId ?? 14200);
|
||||
await updateStatus(extractId, "cart", { basketRowId });
|
||||
// Step 2: addToCart + saveMetadata for EACH item
|
||||
for (const item of items) {
|
||||
const { extractId, input } = item;
|
||||
|
||||
// Step 3: Save metadata + submit order
|
||||
// ePay internal county IDs = eTerra WORKSPACE_IDs
|
||||
// ePay UAT IDs = SIRUTA codes
|
||||
// So we can skip all discovery calls!
|
||||
await updateStatus(extractId, "ordering");
|
||||
await updateStatus(extractId, "cart");
|
||||
const basketRowId = await client.addToCart(input.prodId ?? 14200);
|
||||
item.basketRowId = basketRowId;
|
||||
await updateStatus(extractId, "cart", { basketRowId });
|
||||
|
||||
// Resolve county ID (workspacePk) and UAT ID (siruta) from our DB
|
||||
let countyInternalId = input.judetIndex; // might already be workspacePk
|
||||
let uatInternalId = input.uatId; // might already be SIRUTA
|
||||
let countyName = input.judetName;
|
||||
let uatName = input.uatName;
|
||||
// Resolve county/UAT from SIRUTA if available
|
||||
let countyInternalId = input.judetIndex;
|
||||
let uatInternalId = input.uatId;
|
||||
let countyName = input.judetName;
|
||||
let uatName = input.uatName;
|
||||
|
||||
// If siruta is provided, look up workspacePk from GisUat
|
||||
if (input.siruta) {
|
||||
const uat = await prisma.gisUat.findUnique({
|
||||
where: { siruta: input.siruta },
|
||||
select: { workspacePk: true, county: true, name: true },
|
||||
});
|
||||
if (uat?.workspacePk) {
|
||||
countyInternalId = uat.workspacePk;
|
||||
uatInternalId = Number(input.siruta);
|
||||
if (uat.county) countyName = uat.county;
|
||||
if (uat.name) uatName = uat.name;
|
||||
if (input.siruta) {
|
||||
const uat = await prisma.gisUat.findUnique({
|
||||
where: { siruta: input.siruta },
|
||||
select: { workspacePk: true, county: true, name: true },
|
||||
});
|
||||
if (uat?.workspacePk) {
|
||||
countyInternalId = uat.workspacePk;
|
||||
uatInternalId = Number(input.siruta);
|
||||
if (uat.county) countyName = uat.county;
|
||||
if (uat.name) uatName = uat.name;
|
||||
}
|
||||
}
|
||||
|
||||
const nrCF = input.nrCF ?? input.nrCadastral;
|
||||
await updateStatus(extractId, "ordering");
|
||||
|
||||
const saved = await client.saveMetadata(
|
||||
basketRowId,
|
||||
countyInternalId,
|
||||
countyName,
|
||||
uatInternalId,
|
||||
uatName,
|
||||
nrCF,
|
||||
input.nrCadastral,
|
||||
process.env.ANCPI_DEFAULT_SOLICITANT_ID || "14452",
|
||||
);
|
||||
|
||||
if (!saved) {
|
||||
await updateStatus(extractId, "failed", {
|
||||
errorMessage: "Salvarea metadatelor în ePay a eșuat.",
|
||||
});
|
||||
// Continue with remaining items — the cart still has them
|
||||
// but this one won't get metadata. Remove from batch.
|
||||
item.basketRowId = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
const nrCF = input.nrCF ?? input.nrCadastral;
|
||||
const saved = await client.saveMetadata(
|
||||
basketRowId,
|
||||
countyInternalId,
|
||||
countyName,
|
||||
uatInternalId,
|
||||
uatName,
|
||||
nrCF,
|
||||
input.nrCadastral,
|
||||
process.env.ANCPI_DEFAULT_SOLICITANT_ID || "14452",
|
||||
);
|
||||
if (!saved) {
|
||||
await updateStatus(extractId, "failed", {
|
||||
errorMessage: "Salvarea metadatelor în ePay a eșuat.",
|
||||
});
|
||||
// Filter to only items that had successful metadata saves
|
||||
const validItems = items.filter((i) => i.basketRowId !== undefined);
|
||||
if (validItems.length === 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Step 3: ONE submitOrder for ALL items
|
||||
console.log(
|
||||
`[epay-queue] Submitting order for ${validItems.length} items...`,
|
||||
);
|
||||
const orderId = await client.submitOrder(knownOrderIds);
|
||||
|
||||
await updateStatus(extractId, "polling", { orderId });
|
||||
// Update all valid items with the shared orderId
|
||||
for (const item of validItems) {
|
||||
await updateStatus(item.extractId, "polling", { orderId });
|
||||
}
|
||||
|
||||
// Step 5: Poll until complete
|
||||
// Step 4: Poll until complete
|
||||
const finalStatus = await client.pollUntilComplete(
|
||||
orderId,
|
||||
async (attempt, status) => {
|
||||
await updateStatus(extractId, "polling", {
|
||||
epayStatus: status,
|
||||
pollAttempts: attempt,
|
||||
});
|
||||
for (const item of validItems) {
|
||||
await updateStatus(item.extractId, "polling", {
|
||||
epayStatus: status,
|
||||
pollAttempts: attempt,
|
||||
});
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
@@ -245,77 +284,122 @@ async function processItem(
|
||||
finalStatus.status === "Anulata" ||
|
||||
finalStatus.status === "Plata refuzata"
|
||||
) {
|
||||
await updateStatus(extractId, "cancelled", {
|
||||
epayStatus: finalStatus.status,
|
||||
errorMessage: `Comanda ${finalStatus.status.toLowerCase()}.`,
|
||||
});
|
||||
for (const item of validItems) {
|
||||
await updateStatus(item.extractId, "cancelled", {
|
||||
epayStatus: finalStatus.status,
|
||||
errorMessage: `Comanda ${finalStatus.status.toLowerCase()}.`,
|
||||
});
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// Step 6: Download PDF
|
||||
const doc = finalStatus.documents.find(
|
||||
// Step 5: Download ALL documents and match to items
|
||||
// The order may contain multiple documents — match by filename/nrCadastral
|
||||
const downloadableDocs = finalStatus.documents.filter(
|
||||
(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 null;
|
||||
}
|
||||
|
||||
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",
|
||||
},
|
||||
console.log(
|
||||
`[epay-queue] Order ${orderId}: ${downloadableDocs.length} documents for ${validItems.length} items`,
|
||||
);
|
||||
|
||||
// Step 8: Complete
|
||||
const documentDate = doc.dataDocument
|
||||
? new Date(doc.dataDocument)
|
||||
: new Date();
|
||||
const expiresAt = new Date(documentDate);
|
||||
expiresAt.setDate(expiresAt.getDate() + 30);
|
||||
// Try to match documents to items by cadastral number in filename
|
||||
// ePay filenames look like "Extras_Informare_65297.pdf" (the number is an internal ID, not nrCadastral)
|
||||
// If we have exactly as many docs as items, assign in order.
|
||||
// Otherwise, download all and try best-effort matching.
|
||||
|
||||
await updateStatus(extractId, "completed", {
|
||||
minioPath: path,
|
||||
minioIndex: index,
|
||||
epayStatus: finalStatus.status,
|
||||
completedAt: new Date(),
|
||||
documentDate,
|
||||
expiresAt,
|
||||
});
|
||||
if (downloadableDocs.length === 0) {
|
||||
for (const item of validItems) {
|
||||
await updateStatus(item.extractId, "failed", {
|
||||
epayStatus: finalStatus.status,
|
||||
errorMessage: "Nu s-au găsit documente PDF în comanda finalizată.",
|
||||
});
|
||||
}
|
||||
return orderId;
|
||||
}
|
||||
|
||||
// Simple matching strategy:
|
||||
// If docs.length === items.length, assign 1:1 in order
|
||||
// Otherwise, assign first doc to first item, etc., and leftover items get "failed"
|
||||
for (let i = 0; i < validItems.length; i++) {
|
||||
const item = validItems[i]!;
|
||||
const doc = downloadableDocs[i];
|
||||
|
||||
if (!doc) {
|
||||
await updateStatus(item.extractId, "failed", {
|
||||
epayStatus: finalStatus.status,
|
||||
errorMessage: `Document lipsă (${downloadableDocs.length} documente pentru ${validItems.length} parcele).`,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
await updateStatus(item.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 6: Store in MinIO
|
||||
const { path, index } = await storeCfExtract(
|
||||
pdfBuffer,
|
||||
item.input.nrCadastral,
|
||||
{
|
||||
"ancpi-order-id": orderId,
|
||||
"nr-cadastral": item.input.nrCadastral,
|
||||
judet: item.input.judetName,
|
||||
uat: item.input.uatName,
|
||||
"data-document": doc.dataDocument ?? "",
|
||||
stare: finalStatus.status,
|
||||
produs: "EXI_ONLINE",
|
||||
},
|
||||
);
|
||||
|
||||
// Complete
|
||||
const documentDate = doc.dataDocument
|
||||
? new Date(doc.dataDocument)
|
||||
: new Date();
|
||||
const expiresAt = new Date(documentDate);
|
||||
expiresAt.setDate(expiresAt.getDate() + 30);
|
||||
|
||||
await updateStatus(item.extractId, "completed", {
|
||||
minioPath: path,
|
||||
minioIndex: index,
|
||||
epayStatus: finalStatus.status,
|
||||
completedAt: new Date(),
|
||||
documentDate,
|
||||
expiresAt,
|
||||
});
|
||||
|
||||
console.log(
|
||||
`[epay-queue] Completed: ${item.input.nrCadastral} → ${path}`,
|
||||
);
|
||||
} catch (error) {
|
||||
const message =
|
||||
error instanceof Error ? error.message : "Eroare download/stocare";
|
||||
await updateStatus(item.extractId, "failed", {
|
||||
errorMessage: message,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Update credits after successful order
|
||||
const newCredits = await client.getCredits();
|
||||
updateEpayCredits(newCredits);
|
||||
|
||||
console.log(
|
||||
`[epay-queue] Completed: ${input.nrCadastral} → ${path} (credits: ${newCredits})`,
|
||||
`[epay-queue] Batch complete: orderId=${orderId}, credits remaining=${newCredits}`,
|
||||
);
|
||||
|
||||
return orderId;
|
||||
} 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 });
|
||||
const message =
|
||||
error instanceof Error ? error.message : "Eroare necunoscută";
|
||||
console.error(`[epay-queue] Batch failed:`, message);
|
||||
for (const id of extractIds) {
|
||||
await updateStatus(id, "failed", { errorMessage: message });
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user