inport fix
This commit is contained in:
openhands committed
1 parent
6a67fb6e83
commit
ebd2ff131c
5 files changed
+387
-411
No files matched your search
@@ -277,82 +277,14 @@ export function autoPriceFurni(classname: string): {
|
||||
return { credits: 3, points: 0, pointsType: 0 };
|
||||
}
|
||||
|
||||
/**
|
||||
* Allocate the next `catalog_items.id` and run the INSERT through a
|
||||
* process-level async mutex so concurrent imports in the same Node process
|
||||
* cannot race on MAX(id). A DB-level lock would require pinning a single
|
||||
* connection for the whole operation; the in-process mutex is enough
|
||||
* for the single-server deployment and avoids pool-connection gymnastics.
|
||||
*/
|
||||
let catalogNextId: number | null = null;
|
||||
let catalogIdLastUsed = 0;
|
||||
let catalogIdSeedChain: Promise<void> = Promise.resolve();
|
||||
const CATALOG_ID_REFRESH_MS = 60_000;
|
||||
|
||||
// In-memory cache for catalog page IDs — avoids N+1 DB lookups during batch imports
|
||||
// where many items belong to the same category.
|
||||
let parentIdCache: number | null = null;
|
||||
let categoryPageCache: Map<string, number> | null = null;
|
||||
let categoryCacheTimestamp = 0;
|
||||
const CATEGORY_CACHE_TTL_MS = 5 * 60_000; // 5 minutes
|
||||
|
||||
/** Drop the in-process catalog page cache. Call after bulk operations
|
||||
* (e.g. the PUT re-organize route) to force a fresh DB lookup. */
|
||||
export function resetCatalogPageCache(): void {
|
||||
parentIdCache = null;
|
||||
categoryPageCache = null;
|
||||
categoryCacheTimestamp = 0;
|
||||
}
|
||||
|
||||
/** Allocate a unique catalog_items id. Seeded once with MAX(id)+1, then handed
|
||||
* out from an in-process counter so concurrent imports don't serialize on a
|
||||
* SELECT MAX + INSERT round-trip per item. Re-seeds when idle so rows added
|
||||
* externally don't collide. */
|
||||
export async function allocateCatalogItemId<T>(
|
||||
insertFn: (nextId: number) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const now = Date.now();
|
||||
if (
|
||||
catalogNextId === null ||
|
||||
now - catalogIdLastUsed > CATALOG_ID_REFRESH_MS
|
||||
) {
|
||||
let settle!: () => void;
|
||||
const prev = catalogIdSeedChain;
|
||||
catalogIdSeedChain = new Promise<void>((r) => {
|
||||
settle = r;
|
||||
});
|
||||
await prev.catch(() => {});
|
||||
try {
|
||||
if (
|
||||
catalogNextId === null ||
|
||||
Date.now() - catalogIdLastUsed > CATALOG_ID_REFRESH_MS
|
||||
) {
|
||||
const [maxIdResult] = (await db.execute(sql`
|
||||
SELECT MAX(id) as maxId FROM catalog_items
|
||||
`)) as unknown as [Array<{ maxId: number | bigint | null }>, unknown];
|
||||
catalogNextId = Number(maxIdResult[0]?.maxId ?? 0) + 1;
|
||||
}
|
||||
} finally {
|
||||
settle();
|
||||
}
|
||||
}
|
||||
catalogIdLastUsed = Date.now();
|
||||
const nextId = catalogNextId;
|
||||
catalogNextId += 1;
|
||||
return await insertFn(nextId);
|
||||
}
|
||||
|
||||
/** Test seam: drop the cached id counter between tests. */
|
||||
export function __resetCatalogIdAllocatorForTests(): void {
|
||||
catalogNextId = null;
|
||||
catalogIdLastUsed = 0;
|
||||
catalogIdSeedChain = Promise.resolve();
|
||||
}
|
||||
|
||||
/** Get or create a category sub-page under the imported parent page.
|
||||
* Caches category page IDs in-memory for CATEGORY_CACHE_TTL_MS to avoid
|
||||
* redundant DB lookups during batch imports where many items map to the same
|
||||
* category. */
|
||||
let categoryPageCache: Map<string, number> | null = null;
|
||||
let categoryCacheTimestamp = 0;
|
||||
const CATEGORY_CACHE_TTL_MS = 5 * 60_000; // 5 minutes
|
||||
|
||||
export async function getOrCreateCategoryPage(
|
||||
classname: string,
|
||||
itemType: string,
|
||||
@@ -414,6 +346,71 @@ export async function getOrCreateCategoryPage(
|
||||
return created.id;
|
||||
}
|
||||
|
||||
export function resetCatalogPageCache(): void {
|
||||
parentIdCache = null;
|
||||
categoryPageCache = null;
|
||||
categoryCacheTimestamp = 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Allocate the next `catalog_items.id` and run the INSERT through a
|
||||
* process-level async mutex so concurrent imports in the same Node process
|
||||
* cannot race on MAX(id). A DB-level lock would require pinning a single
|
||||
* connection for the whole operation; the in-process mutex is enough
|
||||
* for the single-server deployment and avoids pool-connection gymnastics.
|
||||
*/
|
||||
let catalogNextId: number | null = null;
|
||||
let catalogIdLastUsed = 0;
|
||||
let catalogIdSeedChain: Promise<void> = Promise.resolve();
|
||||
const CATALOG_ID_REFRESH_MS = 60_000;
|
||||
|
||||
let parentIdCache: number | null = null;
|
||||
|
||||
/** Allocate a unique catalog_items id. Seeded once with MAX(id)+1, then handed
|
||||
* out from an in-process counter so concurrent imports don't serialize on a
|
||||
* SELECT MAX + INSERT round-trip per item. Re-seeds when idle so rows added
|
||||
* externally don't collide. */
|
||||
export async function allocateCatalogItemId<T>(
|
||||
insertFn: (nextId: number) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const now = Date.now();
|
||||
if (
|
||||
catalogNextId === null ||
|
||||
now - catalogIdLastUsed > CATALOG_ID_REFRESH_MS
|
||||
) {
|
||||
let settle!: () => void;
|
||||
const prev = catalogIdSeedChain;
|
||||
catalogIdSeedChain = new Promise<void>((r) => {
|
||||
settle = r;
|
||||
});
|
||||
await prev.catch(() => {});
|
||||
try {
|
||||
if (
|
||||
catalogNextId === null ||
|
||||
Date.now() - catalogIdLastUsed > CATALOG_ID_REFRESH_MS
|
||||
) {
|
||||
const [maxIdResult] = (await db.execute(sql`
|
||||
SELECT MAX(id) as maxId FROM catalog_items
|
||||
`)) as unknown as [Array<{ maxId: number | bigint | null }>, unknown];
|
||||
catalogNextId = Number(maxIdResult[0]?.maxId ?? 0) + 1;
|
||||
}
|
||||
} finally {
|
||||
settle();
|
||||
}
|
||||
}
|
||||
catalogIdLastUsed = Date.now();
|
||||
const nextId = catalogNextId;
|
||||
catalogNextId += 1;
|
||||
return await insertFn(nextId);
|
||||
}
|
||||
|
||||
/** Test seam: drop the cached id counter between tests. */
|
||||
export function __resetCatalogIdAllocatorForTests(): void {
|
||||
catalogNextId = null;
|
||||
catalogIdLastUsed = 0;
|
||||
catalogIdSeedChain = Promise.resolve();
|
||||
}
|
||||
|
||||
// ── Import Single Furni ─────────────────────────────────────────────
|
||||
|
||||
export async function importSingleFurni(params: {
|
||||
@@ -833,9 +830,6 @@ export async function importSingleFurni(params: {
|
||||
const categoryPageId = await getOrCreateCategoryPage(classname, itemType);
|
||||
const pageIdStr = String(categoryPageId);
|
||||
const price = autoPriceFurni(classname);
|
||||
// catalog_items.id has no AUTO_INCREMENT — serialize ID allocation
|
||||
// across concurrent imports via a MySQL named lock to avoid MAX(id)
|
||||
// TOCTOU that would cause duplicate-key failures.
|
||||
catalogItemId = await allocateCatalogItemId(async (nextCatalogId) => {
|
||||
await db.execute(sql`
|
||||
INSERT INTO catalog_items (id, page_id, item_ids, catalog_name, cost_credits, cost_points, points_type, amount, order_number, offer_id, extradata)
|
||||
|
||||
Reference in new issue
Block a user