Skip to content

fix(admin,etl): CORS preflight fix + ETL retry/failure ops + schema alignment #2409

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
andrew-bierman merged 1 commit into from
May 12, 2026
Merged

fix(admin,etl): CORS preflight fix + ETL retry/failure ops + schema alignment #2409

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions apps/admin/components/edit-catalog-dialog.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ export function EditCatalogDialog({ item }: EditCatalogDialogProps) {
: null;
const weightRaw = fd.get('weight')?.toString().trim();
const weight = weightRaw ? Number(weightRaw) : undefined;
const weightUnit = fd.get('weightUnit')?.toString().trim() || item.weightUnit;
const weightUnit = fd.get('weightUnit')?.toString().trim() || item.weightUnit || undefined;
const priceRaw = fd.get('price')?.toString().trim();
const price = priceRaw ? Number(priceRaw) : null;

Expand Down Expand Up @@ -108,7 +108,7 @@ export function EditCatalogDialog({ item }: EditCatalogDialogProps) {
<Input
id="weightUnit"
name="weightUnit"
defaultValue={item.weightUnit}
defaultValue={item.weightUnit ?? ''}
placeholder="g"
/>
</div>
Expand Down
62 changes: 9 additions & 53 deletions apps/admin/lib/api.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,9 @@ import type { App } from '@packrat/api';
import type {
ActiveUsersSchema,
ActivityPointSchema,
AdminCatalogItemSchema,
AdminPackItemSchema,
AdminUserItemSchema,
BrandRowSchema,
BreakdownItemSchema,
CatalogOverviewSchema,
Expand Down Expand Up @@ -69,17 +72,7 @@ export async function getStats(): Promise<AdminStats> {

// ─── Users ────────────────────────────────────────────────────────────────────

export interface AdminUser {
id: number;
email: string;
firstName: string | null;
lastName: string | null;
role: string | null;
emailVerified: boolean | null;
avatarUrl: string | null;
createdAt: string | null;
updatedAt: string | null;
}
export type AdminUser = Static<typeof AdminUserItemSchema>;

export interface PaginatedResponse<T> {
data: T[];
Expand All @@ -103,7 +96,7 @@ export async function getUsers({
query: { limit, offset, q, includeDeleted: includeDeleted ? 'true' : undefined },
});
if (error) throwOnError(error);
return unwrap(data, 'users') as unknown as PaginatedResponse<AdminUser>; // safe-cast: Eden Treaty infers wide union; runtime shape matches PaginatedResponse<T>
return unwrap(data, 'users');
}

export async function deleteUser(id: number): Promise<{ success: boolean }> {
Expand All @@ -129,19 +122,7 @@ export async function restoreUser(id: number): Promise<{ success: boolean }> {

// ─── Packs ────────────────────────────────────────────────────────────────────

export interface AdminPack {
id: string;
name: string;
description: string | null;
category: string;
isPublic: boolean | null;
isAIGenerated: boolean;
tags: string[] | null;
image: string | null;
createdAt: string | null;
updatedAt: string | null;
userEmail: string | null;
}
export type AdminPack = Static<typeof AdminPackItemSchema>;

export async function getPacks({
limit = 100,
Expand All @@ -158,7 +139,7 @@ export async function getPacks({
query: { limit, offset, q, includeDeleted: includeDeleted ? 'true' : undefined },
});
if (error) throwOnError(error);
return unwrap(data, 'packs') as unknown as PaginatedResponse<AdminPack>; // safe-cast: Eden Treaty infers wide union; runtime shape matches PaginatedResponse<T>
return unwrap(data, 'packs');
}

export async function deletePack(id: string): Promise<{ success: boolean }> {
Expand All @@ -169,32 +150,7 @@ export async function deletePack(id: string): Promise<{ success: boolean }> {

// ─── Catalog Items ────────────────────────────────────────────────────────────

export interface AdminCatalogItem {
id: number;
name: string;
description: string | null;
categories: string[] | null;
brand: string | null;
model: string | null;
sku: string | null;
price: number | null;
currency: string | null;
weight: number | null;
weightUnit: string;
availability: string | null;
ratingValue: number | null;
reviewCount: number | null;
color: string | null;
size: string | null;
material: string | null;
seller: string | null;
productUrl: string | null;
images: string[] | null;
variants: Array<{ attribute: string; values: string[] }> | null;
techs: Record<string, string> | null;
links: Array<{ title: string; url: string }> | null;
createdAt: string | null;
}
export type AdminCatalogItem = Static<typeof AdminCatalogItemSchema>;

export interface UpdateCatalogItemInput {
name?: string;
Expand All @@ -219,7 +175,7 @@ export async function getCatalogItems({
query: { limit, offset, q },
});
if (error) throwOnError(error);
return unwrap(data, 'catalog') as unknown as PaginatedResponse<AdminCatalogItem>; // safe-cast: Eden Treaty infers wide union; runtime shape matches PaginatedResponse<T>
return unwrap(data, 'catalog');
}

export async function deleteCatalogItem(id: number): Promise<{ success: boolean }> {
Expand Down
191 changes: 189 additions & 2 deletions packages/api/src/routes/admin/analytics/catalog.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
import { createDb } from '@packrat/api/db';
import { catalogItems, etlJobs } from '@packrat/api/db/schema';
import { catalogItems, etlJobs, invalidItemLogs } from '@packrat/api/db/schema';
import {
AdminErrorResponses,
BrandRowSchema,
CatalogOverviewSchema,
EtlResponseSchema,
PriceBucketSchema,
} from '@packrat/api/schemas/admin';
import { and, avg, count, desc, gt, isNotNull, max, min, sql } from 'drizzle-orm';
import { queueCatalogETL } from '@packrat/api/services/etl/queue';
import { getEnv } from '@packrat/api/utils/env-validation';
import { and, avg, count, desc, eq, gt, isNotNull, lt, max, min, sql } from 'drizzle-orm';
import { Elysia, status, t } from 'elysia';
import { z } from 'zod';

Expand Down Expand Up @@ -257,4 +259,189 @@ export const catalogAnalyticsRoutes = new Elysia({ prefix: '/catalog' })
}
},
{ detail: { tags: ['Admin'], summary: 'Embedding coverage' } },
)

// ─── ETL failure summary ──────────────────────────────────────────────────────

.get(
'/etl/failure-summary',
async ({ query }) => {
const db = createDb();
const { limit = 20 } = query;

try {
const rows = await db.execute<{ field: string; reason: string; count: number }>(
sql`
SELECT
err->>'field' AS field,
err->>'reason' AS reason,
COUNT(*)::int AS count
FROM ${invalidItemLogs},
jsonb_array_elements(${invalidItemLogs.errors}) AS err
GROUP BY err->>'field', err->>'reason'
ORDER BY count DESC
LIMIT ${limit}
`,
);

const [total] = await db.select({ n: count() }).from(invalidItemLogs);
Comment on lines +273 to +287
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win

Run aggregation and count in parallel.

The db.execute and db.select are independent and currently run sequentially, doubling latency for what's essentially a dashboard read. Wrap them in Promise.all. Also note the rest of this file consistently attaches a response TypeBox schema (see /overview, /brands, /etl) — this endpoint and the three below ship without one, so OpenAPI/Eden Treaty consumers get unknown payloads.

♻️ Proposed parallelization 
-        const rows = await db.execute<{ field: string; reason: string; count: number }>(
-          sql`
-            SELECT
-              err->>'field'  AS field,
-              err->>'reason' AS reason,
-              COUNT(*)::int  AS count
-            FROM ${invalidItemLogs},
-                 jsonb_array_elements(${invalidItemLogs.errors}) AS err
-            GROUP BY err->>'field', err->>'reason'
-            ORDER BY count DESC
-            LIMIT ${limit}
-          `,
-        );
-
-        const [total] = await db.select({ n: count() }).from(invalidItemLogs);
+        const [rows, totals] = await Promise.all([
+          db.execute<{ field: string; reason: string; count: number }>(sql`
+            SELECT
+              err->>'field'  AS field,
+              err->>'reason' AS reason,
+              COUNT(*)::int  AS count
+            FROM ${invalidItemLogs},
+                 jsonb_array_elements(${invalidItemLogs.errors}) AS err
+            GROUP BY err->>'field', err->>'reason'
+            ORDER BY count DESC
+            LIMIT ${limit}
+          `),
+          db.select({ n: count() }).from(invalidItemLogs),
+        ]);
+        const total = totals[0];
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/api/src/routes/admin/analytics/catalog.ts` around lines 273 - 287,
The two independent DB calls (db.execute<{ field: string; reason: string; count:
number }>(...) and db.select({ n: count() }).from(invalidItemLogs)) should be
executed in parallel to reduce latency: run them with Promise.all and
destructure the results into rows and total; keep the same query text and types
(referencing db.execute, db.select and invalidItemLogs). Additionally, add the
missing response TypeBox schema used elsewhere to this route (and the three
subsequent endpoints) so OpenAPI/Eden Treaty consumers get a typed payload
(match the pattern used in /overview, /brands, /etl and attach the same response
schema variable to this handler).


return {
topErrors: rows.rows.map((r) => ({
field: r.field,
reason: r.reason,
count: r.count,
})),
totalInvalidItems: total?.n ?? 0,
};
} catch (error) {
console.error('ETL failure summary error:', error);
return status(500, {
error: 'Failed to fetch failure summary',
code: 'ETL_FAILURE_SUMMARY_ERROR',
});
}
},
{
query: z.object({
limit: z.coerce.number().int().min(1).max(100).optional().default(20),
}),
detail: { tags: ['Admin'], summary: 'Top ETL validation failure patterns' },
},
)

// ─── Per-job failure drill-down ───────────────────────────────────────────────

.get(
'/etl/:jobId/failures',
async ({ params, query }) => {
const db = createDb();
const { limit = 50 } = query;

try {
const samples = await db
.select()
.from(invalidItemLogs)
.where(eq(invalidItemLogs.jobId, params.jobId))
.orderBy(invalidItemLogs.rowIndex)
.limit(limit);
Comment on lines +322 to +327

const breakdown = await db.execute<{ field: string; reason: string; count: number }>(
sql`
SELECT
err->>'field' AS field,
err->>'reason' AS reason,
COUNT(*)::int AS count
FROM ${invalidItemLogs},
jsonb_array_elements(${invalidItemLogs.errors}) AS err
WHERE ${invalidItemLogs.jobId} = ${params.jobId}
GROUP BY err->>'field', err->>'reason'
ORDER BY count DESC
`,
);
Comment on lines +322 to +341
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial | ⚡ Quick win

Parallelize the two job-scoped queries.

Same shape as the previous endpoint — samples and breakdown are independent and should run concurrently via Promise.all. Also consider tightening jobId validation to z.string().uuid() on Line 366 so malformed IDs short-circuit before hitting the DB (especially relevant because the raw sql query interpolates ${params.jobId} — parameterized via Drizzle, so not an injection issue, but a wasted round-trip).

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/api/src/routes/admin/analytics/catalog.ts` around lines 322 - 341,
Run the two independent DB calls (the select that assigns samples and the sql
execute that assigns breakdown) in parallel using Promise.all to avoid serial
round-trips: kick off
db.select().from(invalidItemLogs).where(eq(invalidItemLogs.jobId,
params.jobId)).orderBy(...).limit(limit) and db.execute(...) together and await
Promise.all to assign samples and breakdown from the results; also tighten the
jobId validation to z.string().uuid() so params.jobId is validated as a UUID
before any DB interpolation (affecting the code paths that call
samples/breakdown and the schema that currently accepts a plain string).


return {
jobId: params.jobId,
errorBreakdown: breakdown.rows.map((r) => ({
field: r.field,
reason: r.reason,
count: r.count,
})),
samples: samples.map((s) => ({
rowIndex: s.rowIndex,
errors: s.errors,
rawData: s.rawData,
})),
totalShown: samples.length,
};
} catch (error) {
console.error('ETL job failures error:', error);
return status(500, {
error: 'Failed to fetch job failures',
code: 'ETL_JOB_FAILURES_ERROR',
});
}
},
{
params: z.object({ jobId: z.string() }),
query: z.object({
limit: z.coerce.number().int().min(1).max(200).optional().default(50),
}),
detail: { tags: ['Admin'], summary: 'Validation failures for a specific ETL job' },
},
)

// ─── Reset stuck jobs ─────────────────────────────────────────────────────────

.post(
'/etl/reset-stuck',
async () => {
const db = createDb();

try {
// Jobs stuck in 'running' for more than 30 minutes are considered stalled
const stuckCutoff = new Date(Date.now() - 30 * 60 * 1000);

const reset = await db
.update(etlJobs)
.set({ status: 'failed', completedAt: new Date() })
.where(and(eq(etlJobs.status, 'running'), lt(etlJobs.startedAt, stuckCutoff)))
.returning();

return { reset: reset.length, ids: reset.map((r) => r.id) };
} catch (error) {
console.error('ETL reset stuck error:', error);
return status(500, { error: 'Failed to reset stuck jobs', code: 'ETL_RESET_STUCK_ERROR' });
}
},
{ detail: { tags: ['Admin'], summary: 'Mark stuck running ETL jobs as failed' } },
)
Comment on lines +376 to +398
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Find the ETL queue consumer and check how it updates etlJobs status.
fd -t f -e ts -e js | xargs rg -nP -C5 '\betlJobs\b' -g '!**/admin/**' 2>/dev/null
echo '---'
rg -nP -C3 "queueCatalogETL|ETL_QUEUE" --type=ts

Repository: PackRat-AI/PackRat

Length of output: 34809

🏁 Script executed:

# Check processCatalogEtl function entry and queue batch processor for status validation
rg -A10 'async function processCatalogETL' packages/api/src/services/etl/processCatalogEtl.ts
rg -A15 'export async function processQueueBatch' packages/api/src/services/etl/queue.ts

Repository: PackRat-AI/PackRat

Length of output: 710

The ETL worker does not validate job status before processing or during writes — creating TOCTOU and race-condition risks.

The worker (processCatalogETL) starts processing without checking the job's current status, and its error handler unconditionally updates the row with no status guard. This creates two real failure modes:

  1. TOCTOU race on job row: If reset-stuck marks a job failed, the worker continues processing and can still write its own status/completedAt updates, overwriting the reset. At minimum, the error handler should include a WHERE status = 'running' guard, or the worker should validate the job is still 'running' before committing writes.

  2. Concurrent processors from race: An operator triggering retry while reset-stuck is executing can spawn two workers processing the same object key, resulting in duplicate catalog rows.

Additionally, the 30-minute stuck threshold is hardcoded in the endpoint; make it a query parameter (with a sane minimum) or environment variable so operators can adjust detection without redeploying.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/api/src/routes/admin/analytics/catalog.ts` around lines 376 - 398,
The endpoint that marks ETL jobs as failed and the worker lack status guards and
a configurable cutoff: update the '/etl/reset-stuck' handler to accept a query
parameter (or read an environment variable) for the stuck threshold with a
sensible minimum, and use that value instead of the hardcoded 30 minutes; in the
DB update (currently using etlJobs.update in that handler) ensure the WHERE
includes status = 'running' to avoid changing non-running rows; in the worker
(processCatalogETL) add a pre-processing validation that reloads the job row and
aborts if status !== 'running', and change the worker's error/cleanup write
logic to include a conditional WHERE (e.g., WHERE id = jobId AND status =
'running') so it never overwrites a reset or retry decision; these changes
together prevent TOCTOU and concurrent-processor races.


// ─── Retry a failed job ───────────────────────────────────────────────────────

.post(
'/etl/:jobId/retry',
async ({ params }) => {
const db = createDb();

try {
const [original] = await db
.select()
.from(etlJobs)
.where(eq(etlJobs.id, params.jobId))
.limit(1);

if (!original) return status(404, { error: 'ETL job not found' });
if (original.status === 'running')
return status(409, {
error: 'Job is still running — wait for it to complete or reset stuck jobs first',
});

const newJobId = crypto.randomUUID();
const objectKey = `v2/${original.source}/${original.filename}`;
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Confirm whether etlJobs schema already stores an object key, and where the original v2/<source>/<filename> convention is established.
fd -t f -e ts | xargs rg -nP -C3 'etlJobs\s*=|object[Kk]ey|v2/\$\{' 2>/dev/null | head -200

Repository: PackRat-AI/PackRat

Length of output: 9305

🏁 Script executed:

sed -n '497,530p' packages/api/src/db/schema.ts

Repository: PackRat-AI/PackRat

Length of output: 1139

🏁 Script executed:

rg -n 'etlJobs.*insert|db\.insert.*etlJobs' packages/api/src -A 5 -B 2

Repository: PackRat-AI/PackRat

Length of output: 1365

🏁 Script executed:

rg -n 'objectKey.*=|fileName.*=' packages/api/src/routes/upload.ts -B 5 -A 2

Repository: PackRat-AI/PackRat

Length of output: 362

🏁 Script executed:

sed -n '150,190p' packages/api/src/routes/catalog/index.ts

Repository: PackRat-AI/PackRat

Length of output: 959

🏁 Script executed:

sed -n '19,100p' packages/api/src/routes/upload.ts

Repository: PackRat-AI/PackRat

Length of output: 1978

🏁 Script executed:

rg -n 'chunks.*=|objectKey.*=' packages/admin/lib -A 2 -B 2

Repository: PackRat-AI/PackRat

Length of output: 121

🏁 Script executed:

sed -n '140,200p' packages/api/src/routes/catalog/index.ts

Repository: PackRat-AI/PackRat

Length of output: 1573

🏁 Script executed:

rg -n 'v2/' packages/api --type ts | head -20

Repository: PackRat-AI/PackRat

Length of output: 287

🏁 Script executed:

rg -n 'catalogETLSchema' packages/api/src -A 10

Repository: PackRat-AI/PackRat

Length of output: 1694

🏁 Script executed:

rg -n 'queueCatalogETL|CatalogETL' packages/api/src/routes -B 3 -A 3 | head -50

Repository: PackRat-AI/PackRat

Length of output: 2577

Add objectKey field to etlJobs table to persist the original R2 key on first queue.

The retry endpoint reconstructs the key as v2/${source}/${filename}, but that assumes all uploads follow this prefix shape. Since the initial ETL queueing accepts arbitrary chunks (object keys) from the client and never stores them, the retry has no way to know what the original keys were. If an upload used a different naming convention or path structure, the retry silently queues non-existent keys and returns 200—masking the failure.

Either: (a) add an objectKey column to etlJobs and persist the full keys during initial queue, then retrieve and re-enqueue them during retry, or (b) validate the reconstructed key exists in R2 before enqueueing and return 404 if missing.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/api/src/routes/admin/analytics/catalog.ts` at line 421, Persist the
original R2 key instead of reconstructing it: add an objectKey column to the
etlJobs table and, in the code path that creates/queues the initial ETL job
(where original and chunks are accepted and etlJobs rows are inserted), store
the full object key (use the existing original or chunks value) into
etlJobs.objectKey; then update the retry logic (which currently builds
`v2/${source}/${filename}`) to read and re-enqueue using etlJobs.objectKey.
Alternatively, if you prefer not to alter schema, change the retry handler to
call R2 to validate the reconstructed key before enqueueing and return 404 if
the R2 object does not exist; ensure the variable objectKey and the etlJobs
lookup are updated accordingly so retries use the persisted key or validated
key.

const env = getEnv();

if (!env.ETL_QUEUE) return status(400, { error: 'ETL_QUEUE is not configured' });

await db.insert(etlJobs).values({
id: newJobId,
status: 'running',
source: original.source,
filename: original.filename,
scraperRevision: original.scraperRevision,
startedAt: new Date(),
});
Comment on lines +408 to +433
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical | ⚡ Quick win

Retry accepts any non-running status and can orphan a running row on enqueue failure.

Two correctness problems in this handler:

  1. Status not constrained to failed. Line 415 only rejects running, so a completed job can be "retried" — re-queuing the same object key will re-ingest every row and likely duplicate catalog items (or churn embeddings). The PR description explicitly scopes this to failed jobs; enforce it.
  2. Non-atomic insert-then-enqueue. The new etlJobs row is committed at Line 426 before queueCatalogETL at Line 435. If the enqueue throws (queue unavailable, network blip), you've left a brand-new running row that will never progress — exactly the kind of stuck job /etl/reset-stuck was added to clean up. Enqueue first (or wrap in try/catch that rolls back the insert / marks the new row failed).

Also worth noting: there's no idempotency on retries — clicking the button twice creates two parallel runs of the same file. A failed_at-based uniqueness check or a short-lived advisory lock on (source, filename) would harden this.

🛡️ Proposed fix 
-        if (!original) return status(404, { error: 'ETL job not found' });
-        if (original.status === 'running')
-          return status(409, {
-            error: 'Job is still running — wait for it to complete or reset stuck jobs first',
-          });
+        if (!original) return status(404, { error: 'ETL job not found' });
+        if (original.status !== 'failed') {
+          return status(409, {
+            error: `Only failed jobs can be retried (current status: ${original.status})`,
+          });
+        }
 
         const newJobId = crypto.randomUUID();
         const objectKey = `v2/${original.source}/${original.filename}`;
         const env = getEnv();
 
         if (!env.ETL_QUEUE) return status(400, { error: 'ETL_QUEUE is not configured' });
 
         await db.insert(etlJobs).values({
           id: newJobId,
           status: 'running',
           source: original.source,
           filename: original.filename,
           scraperRevision: original.scraperRevision,
           startedAt: new Date(),
         });
-
-        await queueCatalogETL({ queue: env.ETL_QUEUE, objectKeys: [objectKey], jobId: newJobId });
+
+        try {
+          await queueCatalogETL({
+            queue: env.ETL_QUEUE,
+            objectKeys: [objectKey],
+            jobId: newJobId,
+          });
+        } catch (enqueueErr) {
+          await db
+            .update(etlJobs)
+            .set({ status: 'failed', completedAt: new Date() })
+            .where(eq(etlJobs.id, newJobId));
+          throw enqueueErr;
+        }
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/api/src/routes/admin/analytics/catalog.ts` around lines 408 - 433,
Change the retry handler to only allow retries when original.status === 'failed'
(replace the current check that only rejects 'running'), and make the enqueue +
DB insert atomic: call queueCatalogETL(objectKey, newJobId, ...) before
committing the new row or wrap the insert (db.insert(etlJobs).values(...)) in a
try/catch that rolls back the insert or updates the newly inserted row back to
'failed' if queueCatalogETL throws; ensure you still validate env.ETL_QUEUE
before attempting enqueue. Additionally, harden against duplicate parallel
retries by checking for an existing recent/pending job for the same
(original.source, original.filename) (e.g., a failed_at uniqueness check or
short-lived advisory lock) before creating newJobId.


await queueCatalogETL({ queue: env.ETL_QUEUE, objectKeys: [objectKey], jobId: newJobId });

Comment on lines +420 to +436
return { success: true, newJobId, objectKey };
} catch (error) {
console.error('ETL retry error:', error);
return status(500, { error: 'Failed to retry ETL job', code: 'ETL_RETRY_ERROR' });
}
},
{
params: z.object({ jobId: z.string() }),
detail: { tags: ['Admin'], summary: 'Retry a failed ETL job' },
},
Comment on lines +264 to +446
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win

No response TypeBox schemas on any of the four new routes.

Every other route in this file declares response: { 200: ..., ...AdminErrorResponses }, which is what feeds OpenAPI and the Eden Treaty types consumed by apps/admin/lib/api.ts (the very thing this PR is tightening with Static<> derivations). Without response schemas here, the admin client will see these payloads as unknown and you'll lose the type-safety win for the new endpoints. Add schemas to packages/api/src/schemas/admin and wire them in, mirroring EtlResponseSchema.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/api/src/routes/admin/analytics/catalog.ts` around lines 264 - 446,
The four new routes ('/etl/failure-summary', '/etl/:jobId/failures',
'/etl/reset-stuck', '/etl/:jobId/retry') are missing response schemas so the
admin client sees their payloads as unknown; add appropriate response TypeBox
schemas in packages/api/src/schemas/admin (mirror EtlResponseSchema structure)
for each route's success shape (e.g., Top error list + total for
failure-summary, jobId/errorBreakdown/samples for per-job failures, reset
count/ids for reset-stuck, and { success, newJobId, objectKey } for retry),
export them (e.g., EtlFailureSummaryResponseSchema,
EtlJobFailuresResponseSchema, EtlResetStuckResponseSchema,
EtlRetryResponseSchema) and wire them into each route's options under response:
{ 200: <schema>, ...AdminErrorResponses } so OpenAPI and the Admin Static<>
types pick up the correct types; use the route-local symbols like
invalidItemLogs, etlJobs, queueCatalogETL, createDb to locate the handlers to
update.

);
30 changes: 24 additions & 6 deletions packages/api/src/schemas/admin.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,9 +33,9 @@ export const AdminUserItemSchema = t.Object({
lastName: t.Nullable(t.String()),
role: t.Nullable(t.String()),
emailVerified: t.Nullable(t.Boolean()),
avatarUrl: t.Nullable(t.String()),
createdAt: t.Nullable(t.String()),
lastActiveAt: t.Nullable(t.String()),
deletedAt: t.Nullable(t.String()),
updatedAt: t.Nullable(t.String()),
});

// ─── Packs ────────────────────────────────────────────────────────────────────
Expand All @@ -46,9 +46,11 @@ export const AdminPackItemSchema = t.Object({
description: t.Nullable(t.String()),
category: t.String(),
isPublic: t.Nullable(t.Boolean()),
deleted: t.Boolean(),
deletedAt: t.Nullable(t.String()),
isAIGenerated: t.Boolean(),
tags: t.Nullable(t.Array(t.String())),
image: t.Nullable(t.String()),
createdAt: t.Nullable(t.String()),
updatedAt: t.Nullable(t.String()),
userEmail: t.Nullable(t.String()),
});

Expand All @@ -57,11 +59,27 @@ export const AdminPackItemSchema = t.Object({
export const AdminCatalogItemSchema = t.Object({
id: t.Number(),
name: t.String(),
description: t.Nullable(t.String()),
categories: t.Nullable(t.Array(t.String())),
brand: t.Nullable(t.String()),
model: t.Nullable(t.String()),
sku: t.String(),
price: t.Nullable(t.Number()),
weight: t.Number(),
weightUnit: t.String(),
currency: t.Nullable(t.String()),
weight: t.Nullable(t.Number()),
weightUnit: t.Nullable(t.String()),
availability: t.Nullable(t.String()),
ratingValue: t.Nullable(t.Number()),
reviewCount: t.Nullable(t.Number()),
color: t.Nullable(t.String()),
size: t.Nullable(t.String()),
material: t.Nullable(t.String()),
seller: t.Nullable(t.String()),
productUrl: t.String(),
images: t.Nullable(t.Array(t.String())),
variants: t.Nullable(t.Array(t.Object({ attribute: t.String(), values: t.Array(t.String()) }))),
techs: t.Nullable(t.Record(t.String(), t.String())),
links: t.Nullable(t.Array(t.Object({ title: t.String(), url: t.String() }))),
createdAt: t.Nullable(t.String()),
});

Expand Down
Loading