root/sase.tr
1
0
Fork 0
Code Issues 5 Pull Requests 3 Actions 6 Packages Projects Releases Wiki Activity

dev #92

Merged
root merged 2 commits from dev into main 2026-06-03 20:22:06 +03:00
Conversation 0 Commits 2 Files Changed 2 +74 -17
2 changed files with 74 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
apps/api/src/jobs/prefetch-utils.ts
View File

@@ -77,10 +77,15 @@ function currentIstanbulHour(): number {
* captures alive 24/7 so backfill can sweep through the night when the
* upstream is most idle.
*/
/** PL24 scrape window (Europe/Istanbul hours), env-tunable so it can be widened
* without a redeploy (e.g. PREFETCH_PL24_START=8 PREFETCH_PL24_END=22). */
const PL24_WINDOW_START = Number(process.env.PREFETCH_PL24_START) || 9;
const PL24_WINDOW_END = Number(process.env.PREFETCH_PL24_END) || 18;
export function isWithinTimeWindow(source: string): boolean {
if (source !== "pl24") return true;
const h = currentIstanbulHour();
return h >= 9 && h < 18;
return h >= PL24_WINDOW_START && h < PL24_WINDOW_END;
}
/**
@@ -118,13 +123,12 @@ export function msUntilNext9AM(): number {
const minute = get("minute");
const second = get("second");
// If before 09:00 → wait until 09:00 today
// If 18:00+ → wait until 09:00 tomorrow
// Before the window opens → wait until it opens today; otherwise → tomorrow.
let hoursToWait: number;
if (hour < 9) {
hoursToWait = 9 - hour;
if (hour < PL24_WINDOW_START) {
hoursToWait = PL24_WINDOW_START - hour;
} else {
hoursToWait = 24 - hour + 9;
hoursToWait = 24 - hour + PL24_WINDOW_START;
}
const ms = hoursToWait * 3600_000 - minute * 60_000 - second * 1000;

75
apps/api/src/jobs/prefetch-worker.service.ts
View File

@@ -44,11 +44,36 @@ const BACKFILL_SOURCES = ["pl24", "emex", "parts-catalogs"];
/** Redis key holding the rolling rescan cursor (last createdAt seen). */
const BACKFILL_CURSOR_KEY = "prefetch:backfill:cursor";
// ── Throughput knobs (env-tunable so prod can ramp without a redeploy) ──
/**
* Concurrent jobs the worker runs. emex/pl24 fetches parallelise well, so >1
* lets a slow parts-catalogs job stop head-of-line-blocking emex. A user-active
* cooldown still pauses the whole worker, so this only speeds up idle periods.
*/
const WORKER_CONCURRENCY = Number(process.env.PREFETCH_CONCURRENCY) || 3;
/** Global queue rate ceiling (jobs per minute). Was a hard 5/min. */
const WORKER_RATE_MAX = Number(process.env.PREFETCH_RATE_MAX) || 20;
/**
* Per-job pacing for parts-catalogs only (its browser/JWT capture is heavy).
* Set 0 to disable. Other sources are paced by the limiter + cooldown alone.
*/
const PCAT_PACE_MS = Number(process.env.PREFETCH_PCAT_DELAY_MS) || 15_000;
// ── Phase-1 residue exclusion ──
/**
* Skip a zero-parts vehicle once this many backfill attempts have completed with
* still-zero parts (genuine upstream-data gaps — e.g. a VIN in no catalog).
* Re-picking these every wave wastes capacity and starves the Phase-2 rescan.
*/
const NORESULT_MAX_ATTEMPTS = Number(process.env.PREFETCH_NORESULT_MAX) || 2;
/** TTL for the no-result counter — excluded vehicles retry after this, so a later
* catalog fix eventually re-fills them (default 7 days). */
const NORESULT_TTL_S = (Number(process.env.PREFETCH_NORESULT_TTL_DAYS) || 7) * 86_400;
@Injectable()
export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(PrefetchWorkerService.name);
private worker: Worker | null = null;
private pcatJobIndex = 0;
constructor(
@Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
@@ -65,8 +90,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
(job, token) => backfillContext.run(true, () => this.process(job, token)),
{
connection: getBullConnection(),
concurrency: 1,
limiter: { max: 5, duration: 60_000 },
concurrency: WORKER_CONCURRENCY,
limiter: { max: WORKER_RATE_MAX, duration: 60_000 },
},
);
@@ -80,7 +105,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
this.logger.error(`[prefetch] Worker error: ${err.message}`);
});
this.logger.log("[prefetch] Worker started (concurrency=1, 5 jobs/min)");
this.logger.log(
`[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, ${WORKER_RATE_MAX} jobs/min, pcatPace=${PCAT_PACE_MS}ms)`,
);
}
async onModuleDestroy() {
@@ -93,8 +120,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
private async process(job: Job, token?: string): Promise<void> {
try {
const data = job.data as { source?: string };
if (data.source === "parts-catalogs") {
await new Promise((r) => setTimeout(r, 15_000));
if (data.source === "parts-catalogs" && PCAT_PACE_MS > 0) {
await new Promise((r) => setTimeout(r, PCAT_PACE_MS));
}
if (job.name === "backfill-scan") {
@@ -141,8 +168,6 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
const { vehicleId, source } = job.data;
this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`);
this.pcatJobIndex = 0;
await checkCooldown(this.redis, source);
checkTimeWindow(source);
@@ -186,6 +211,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
if (topCategories.length === 0) {
this.logger.log(`[prefetch] No categories for vehicle=${vehicleId}`);
// No categories after a self-seed attempt = no catalog data for this VIN.
// Count it as residue so Phase-1 stops re-picking it every wave.
await this.markNoResult(vehicleId);
await updateProgress(this.redis, vehicleId, {
status: "completed",
total: 0,
@@ -361,6 +389,10 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
const tryPick = async (v: { id: string; source: string | null }): Promise<void> => {
if (picked.length >= BACKFILL_BATCH_SIZE || seen.has(v.id) || !v.source) return;
if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // already in flight
// Skip exhausted residue: vehicles whose prefetch keeps finishing with zero
// parts (no catalog data). They'd otherwise be re-picked every wave forever.
const noResult = await this.redis.get(this.noResultKey(v.id));
if (noResult && Number(noResult) >= NORESULT_MAX_ATTEMPTS) return;
seen.add(v.id);
picked.push({ id: v.id, source: v.source });
};
@@ -515,9 +547,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
jobId: `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`,
};
if (data.source === "parts-catalogs") {
opts.delay = ++this.pcatJobIndex * 20_000;
}
// parts-catalogs pacing is handled per-job in process() (PCAT_PACE_MS) + the
// limiter. The old cumulative `index * 20s` delay was pathological (the Nth
// leaf of a vehicle waited N*20s) and is gone.
await this.queue.add(name, data, opts);
}
@@ -539,6 +571,15 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
if (isFinished) {
this.logger.log(`[prefetch] Completed all jobs for vehicle=${vehicleId}`);
// Genuine residue: the whole chain finished but the vehicle still has no
// parts (all leaves empty / no catalog data). Count it so Phase-1 stops
// re-picking it every wave.
const [hasPart] = await this.db
.select({ id: parts.id })
.from(parts)
.where(eq(parts.vehicleId, vehicleId))
.limit(1);
if (!hasPart) await this.markNoResult(vehicleId);
// Clean up Redis keys — data is in PostgreSQL now
await this.redis.del(`prefetch:scheduled:${vehicleId}`);
await this.redis.del(`prefetch:progress:${vehicleId}`);
@@ -552,4 +593,16 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
errors: (progress.errors || 0) + 1,
});
}
private noResultKey(vehicleId: string): string {
return `prefetch:noresult:${vehicleId}`;
}
/** Record that a backfill attempt finished with the vehicle still at zero parts. */
private async markNoResult(vehicleId: string): Promise<void> {
const key = this.noResultKey(vehicleId);
const n = await this.redis.incr(key);
await this.redis.expire(key, NORESULT_TTL_S);
this.logger.debug(`[prefetch] no-result #${n} for vehicle=${vehicleId}`);
}
}