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

feat(FN-094): add comment line for deployment verification
Some checks failed
Sync dev → Gitea / Mirror dev to Gitea (push) Has been cancelled
Details

Browse Source 
- Added a comment line to main.ts for deployment verification purposes
This commit is contained in:
Fusion
2026-05-11 02:07:03 +00:00
parent c72f063a25
commit f4fea1e429
274 changed files with 20712 additions and 6305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apps/api/src/jobs/bull.config.ts
View File

@@ -25,4 +25,5 @@ export const QUEUE_NAMES = {
SUBSCRIPTION_EXPIRY: "subscription-expiry",
QUERY_CLEANUP: "query-cleanup",
CATALOG_PREFETCH: "catalog-prefetch",
TRANSLATION: "translation",
} as const;

31
apps/api/src/jobs/jobs.module.ts
View File

@@ -1,17 +1,17 @@
import { Module, OnModuleInit, Inject, OnModuleDestroy } from "@nestjs/common";
import { Inject, Module, type OnModuleDestroy, type OnModuleInit } from "@nestjs/common";
import { Queue } from "bullmq";
import { EmexScrapeQueueProvider, EMEX_SCRAPE_QUEUE } from "./queues/emex-scrape.queue";
import {
SubscriptionExpiryQueueProvider,
SUBSCRIPTION_EXPIRY_QUEUE,
} from "./queues/subscription-expiry.queue";
import { QueryCleanupQueueProvider, QUERY_CLEANUP_QUEUE } from "./queues/query-cleanup.queue";
import {
CatalogPrefetchQueueProvider,
CATALOG_PREFETCH_QUEUE,
} from "./queues/catalog-prefetch.queue";
import { PrefetchWorkerService } from "./prefetch-worker.service";
import { CategoriesModule } from "../categories/categories.module";
import { PrefetchWorkerService } from "./prefetch-worker.service";
import {
CATALOG_PREFETCH_QUEUE,
CatalogPrefetchQueueProvider,
} from "./queues/catalog-prefetch.queue";
import { EMEX_SCRAPE_QUEUE, EmexScrapeQueueProvider } from "./queues/emex-scrape.queue";
import { QUERY_CLEANUP_QUEUE, QueryCleanupQueueProvider } from "./queues/query-cleanup.queue";
import {
SUBSCRIPTION_EXPIRY_QUEUE,
SubscriptionExpiryQueueProvider,
} from "./queues/subscription-expiry.queue";
@Module({
imports: [CategoriesModule],
@@ -22,7 +22,12 @@ import { CategoriesModule } from "../categories/categories.module";
CatalogPrefetchQueueProvider,
PrefetchWorkerService,
],
exports: [EMEX_SCRAPE_QUEUE, SUBSCRIPTION_EXPIRY_QUEUE, QUERY_CLEANUP_QUEUE, CATALOG_PREFETCH_QUEUE],
exports: [
EMEX_SCRAPE_QUEUE,
SUBSCRIPTION_EXPIRY_QUEUE,
QUERY_CLEANUP_QUEUE,
CATALOG_PREFETCH_QUEUE,
],
})
export class JobsModule implements OnModuleInit, OnModuleDestroy {
constructor(

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

@@ -15,10 +15,7 @@ export class RateLimitError extends Error {
* Check if a user is actively using the source.
* Throws RateLimitError (1min retry) if cooldown key exists.
*/
export async function checkCooldown(
redis: RedisService,
source: string,
): Promise<void> {
export async function checkCooldown(redis: RedisService, source: string): Promise<void> {
const key = `prefetch:activity:${source}`;
const exists = await redis.exists(key);
if (exists) {
@@ -39,7 +36,7 @@ export function checkTimeWindow(source: string): void {
hour: "numeric",
hour12: false,
}).format(new Date());
const h = parseInt(hourStr, 10);
const h = Number.parseInt(hourStr, 10);
const endHour = source === "parts-catalogs" ? 19 : 18;
if (h < 9 || h >= endHour) {
@@ -66,7 +63,7 @@ export function msUntilNext9AM(): number {
}).formatToParts(now);
const get = (type: string) =>
parseInt(istParts.find((p) => p.type === type)?.value || "0", 10);
Number.parseInt(istParts.find((p) => p.type === type)?.value || "0", 10);
const hour = get("hour");
const minute = get("minute");
@@ -81,10 +78,7 @@ export function msUntilNext9AM(): number {
hoursToWait = 24 - hour + 9;
}
const ms =
hoursToWait * 3600_000 -
minute * 60_000 -
second * 1000;
const ms = hoursToWait * 3600_000 - minute * 60_000 - second * 1000;
// At least 1 minute, at most 15 hours
return Math.max(60_000, Math.min(ms, 15 * 3600_000));
@@ -104,19 +98,20 @@ export interface PrefetchProgress {
updatedAt: string;
}
export async function initProgress(
redis: RedisService,
vehicleId: string,
): Promise<void> {
export async function initProgress(redis: RedisService, vehicleId: string): Promise<void> {
const now = new Date().toISOString();
await redis.setJson(progressKey(vehicleId), {
status: "running",
total: 0,
completed: 0,
errors: 0,
startedAt: now,
updatedAt: now,
} satisfies PrefetchProgress, 86400);
await redis.setJson(
progressKey(vehicleId),
{
status: "running",
total: 0,
completed: 0,
errors: 0,
startedAt: now,
updatedAt: now,
} satisfies PrefetchProgress,
86400,
);
}
export async function updateProgress(

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

@@ -2,17 +2,16 @@ import {
Inject,
Injectable,
Logger,
OnModuleDestroy,
OnModuleInit,
type OnModuleDestroy,
type OnModuleInit,
} from "@nestjs/common";
import { Job, Queue, Worker } from "bullmq";
import { eq, and, isNull } from "drizzle-orm";
import { CATALOG_PREFETCH_QUEUE } from "./queues/catalog-prefetch.queue";
import { getBullConnection, QUEUE_NAMES } from "./bull.config";
import {
PrefetchInitJobData,
PrefetchCategoryJobData,
} from "./prefetch.types";
import { type Job, type Queue, Worker } from "bullmq";
import { and, eq, isNull } from "drizzle-orm";
import { CategoriesService } from "../categories/categories.service";
import { DATABASE, type Database } from "../database/database.provider";
import { categories, parts, vehicles } from "../database/schema/core";
import { RedisService } from "../redis/redis.service";
import { QUEUE_NAMES, getBullConnection } from "./bull.config";
import {
RateLimitError,
checkCooldown,
@@ -20,10 +19,8 @@ import {
initProgress,
updateProgress,
} from "./prefetch-utils";
import { CategoriesService } from "../categories/categories.service";
import { RedisService } from "../redis/redis.service";
import { DATABASE, Database } from "../database/database.provider";
import { categories, parts, vehicles } from "../database/schema/core";
import { PrefetchCategoryJobData, PrefetchInitJobData } from "./prefetch.types";
import { CATALOG_PREFETCH_QUEUE } from "./queues/catalog-prefetch.queue";
const MAX_DEPTH = 5;
@@ -41,15 +38,11 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
) {}
onModuleInit() {
this.worker = new Worker(
QUEUE_NAMES.CATALOG_PREFETCH,
(job) => this.process(job),
{
connection: getBullConnection(),
concurrency: 1,
limiter: { max: 5, duration: 60_000 },
},
);
this.worker = new Worker(QUEUE_NAMES.CATALOG_PREFETCH, (job) => this.process(job), {
connection: getBullConnection(),
concurrency: 1,
limiter: { max: 5, duration: 60_000 },
});
this.worker.on("failed", (job, err) => {
if (err instanceof RateLimitError) {
@@ -57,9 +50,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
`[prefetch] Job ${job?.name} rate-limited, will retry in ${err.retryAfterMs}ms`,
);
} else {
this.logger.warn(
`[prefetch] Job ${job?.name} failed: ${err.message}`,
);
this.logger.warn(`[prefetch] Job ${job?.name} failed: ${err.message}`);
}
});
@@ -125,12 +116,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
const topCategories = await this.db
.select()
.from(categories)
.where(
and(
eq(categories.vehicleId, vehicleId),
isNull(categories.parentId),
),
);
.where(and(eq(categories.vehicleId, vehicleId), isNull(categories.parentId)));
if (topCategories.length === 0) {
this.logger.log(`[prefetch] No categories for vehicle=${vehicleId}`);
@@ -203,21 +189,15 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
/**
* Fetch children (sub-categories) for a category.
*/
private async processChildren(
job: Job<PrefetchCategoryJobData>,
): Promise<void> {
private async processChildren(job: Job<PrefetchCategoryJobData>): Promise<void> {
const { vehicleId, categoryId, source, depth } = job.data;
this.logger.log(
`[prefetch] Children for category=${categoryId}, depth=${depth}`,
);
this.logger.log(`[prefetch] Children for category=${categoryId}, depth=${depth}`);
await checkCooldown(this.redis, source);
checkTimeWindow(source);
if (depth >= MAX_DEPTH) {
this.logger.warn(
`[prefetch] Max depth reached for category=${categoryId}`,
);
this.logger.warn(`[prefetch] Max depth reached for category=${categoryId}`);
return;
}
@@ -234,9 +214,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
if (queued > 0) {
await updateProgress(this.redis, vehicleId, {
total:
((await this.redis.getJson<{ total: number }>(
`prefetch:progress:${vehicleId}`,
))?.total || 0) + queued,
((await this.redis.getJson<{ total: number }>(`prefetch:progress:${vehicleId}`))
?.total || 0) + queued,
});
}
@@ -253,9 +232,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
/**
* Fetch parts + schema for a leaf category.
*/
private async processParts(
job: Job<PrefetchCategoryJobData>,
): Promise<void> {
private async processParts(job: Job<PrefetchCategoryJobData>): Promise<void> {
const { vehicleId, categoryId, source } = job.data;
this.logger.log(`[prefetch] Parts for category=${categoryId}`);
@@ -345,10 +322,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
);
}
private async addJob(
name: string,
data: PrefetchCategoryJobData,
): Promise<void> {
private async addJob(name: string, data: PrefetchCategoryJobData): Promise<void> {
const opts: Record<string, unknown> = {
jobId: `prefetch:${data.vehicleId}:${data.categoryId}:${data.action}`,
};
@@ -384,9 +358,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
}
private async incrementErrors(vehicleId: string): Promise<void> {
const progress = await this.redis.getJson<{ errors: number }>(
`prefetch:progress:${vehicleId}`,
);
const progress = await this.redis.getJson<{ errors: number }>(`prefetch:progress:${vehicleId}`);
if (!progress) return;
await updateProgress(this.redis, vehicleId, {
errors: (progress.errors || 0) + 1,

11
apps/api/src/jobs/processors/emex-scrape.processor.ts
View File

@@ -2,13 +2,15 @@ import { Job } from "bullmq";
import { eq } from "drizzle-orm";
import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
import {
emexVehicles,
emexVehicleVins,
emexCatalogs,
emexPartGroups,
emexPartNumbers,
emexParts,
emexScrapeSessions,
emexVehicleGroupLinks,
emexVehiclePartLinks,
emexVehicleVins,
emexVehicles,
} from "../../database/schema/emex";
// Legacy type — kept inline since emex.types.ts was rewritten for emexdwc.ae integration
interface EmexScrapeJobData {
@@ -30,10 +32,11 @@ export async function processEmexScrape(
console.log(`[emex-scrape] Processing job ${job.id} for VIN: ${vin}, user: ${userId}`);
// Update scrape session to active
if (!job.id) throw new Error("BullMQ job missing id");
const [session] = await db
.select()
.from(emexScrapeSessions)
.where(eq(emexScrapeSessions.jobId, job.id!))
.where(eq(emexScrapeSessions.jobId, job.id))
.limit(1);
if (session) {
@@ -47,7 +50,7 @@ export async function processEmexScrape(
try {
// ── Step 1: Resolve vehicle from VIN ──────────────────
let emexVehicleRecord = await db
const emexVehicleRecord = await db
.select({ id: emexVehicles.id, vehicleId: emexVehicles.vehicleId })
.from(emexVehicles)
.innerJoin(emexVehicleVins, eq(emexVehicleVins.emexVehicleId, emexVehicles.id))

9
apps/api/src/jobs/processors/subscription-expiry.processor.ts
View File

@@ -1,7 +1,7 @@
import { Job } from "bullmq";
import { and, eq, lt } from "drizzle-orm";
import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
import { userSubscriptions, userBrands } from "../../database/schema/core";
import { userBrands, userSubscriptions } from "../../database/schema/core";
type Database = PostgresJsDatabase<Record<string, unknown>>;
@@ -17,12 +17,7 @@ export async function processSubscriptionExpiry(
const expiredSubs = await db
.select({ id: userSubscriptions.id, userId: userSubscriptions.userId })
.from(userSubscriptions)
.where(
and(
eq(userSubscriptions.status, "active"),
lt(userSubscriptions.endDate, now),
),
);
.where(and(eq(userSubscriptions.status, "active"), lt(userSubscriptions.endDate, now)));
if (expiredSubs.length === 0) {
console.log("[subscription-expiry] No expired subscriptions found");

142
apps/api/src/jobs/processors/translation.processor.ts Normal file
View File

@@ -0,0 +1,142 @@
import { Job } from "bullmq";
import { sql as drizzleSql, inArray } from "drizzle-orm";
import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
import Redis from "ioredis";
import OpenAI from "openai";
import { emexCategoryTranslations } from "../../database/schema/core";
// Schema-loose local alias (matches the shape used by worker.ts which builds
// drizzle without schema generics). Other processors do the same.
type Database = PostgresJsDatabase<Record<string, unknown>>;
export interface TranslationJobData {
/** Raw original names (from EMEX/PCAT scraper output) to translate. */
terms: string[];
}
const MODEL = "deepseek/deepseek-chat";
const SYSTEM_PROMPT = `Sen bir Türk otomotiv çevirmenisin. Görevin: araç yedek parça kataloğundan gelen kategori ve parça isimlerini İngilizce'den (zaman zaman Rusça'dan) Türkçe'ye çevirmek.
Kurallar:
1. Türkiye yedek parça sektöründe kullanılan terminolojiyi kullan ("Brake Pad" → "Fren Balatası", "Spark Plug" → "Buji").
2. Türkçe karakterleri (ş, ı, ğ, ü, ö, ç) doğru kullan.
3. Marka isimleri (BMW, VW, Toyota), model kodları, OEM parça kodları ve teknik kısaltmalar (ABS, ESP, OBD, ECU, R.H., L.H.) çevirmeden olduğu gibi kalır.
4. Belirsiz terim için en yakın TR karşılığını yaz; çok belirsizse orijinali koru.
5. Kısa, UI'da gösterilebilir (1-5 kelime ideal).
6. "Boot" otomotiv bağlamında "Bagaj" demektir.
7. Çıktı: girdi listesinin AYNI sırasında, eşit uzunlukta JSON dizisi.
Yanıt KESİN olarak şu JSON formatında olmalı:
{"translations": ["çeviri1", "çeviri2", ...]}`;
async function callLLM(ai: OpenAI, terms: string[]): Promise<string[]> {
let lastErr: unknown = null;
for (let attempt = 0; attempt < 3; attempt++) {
try {
const response = await ai.chat.completions.create({
model: MODEL,
max_tokens: 4096,
response_format: { type: "json_object" },
messages: [
{ role: "system", content: SYSTEM_PROMPT },
{ role: "user", content: JSON.stringify({ terms }) },
],
});
const content = response.choices[0]?.message?.content;
if (!content) throw new Error("Empty response");
const match = content.match(/\{[\s\S]*?"translations"[\s\S]*?\}/);
if (!match) throw new Error("No JSON in response");
const parsed = JSON.parse(match[0]) as { translations: string[] };
if (!Array.isArray(parsed.translations) || parsed.translations.length !== terms.length) {
throw new Error(`Length mismatch: ${parsed.translations?.length} vs ${terms.length}`);
}
return parsed.translations;
} catch (err) {
lastErr = err;
const status =
(err as { status?: number })?.status ??
(err as { response?: { status?: number } })?.response?.status;
if (status === 429 || status === 529 || (status && status >= 500)) {
await new Promise((r) => setTimeout(r, 2 ** attempt * 1000));
continue;
}
throw err;
}
}
throw lastErr;
}
/**
* Process a translation job: translate `terms` via LLM, persist to
* emex_category_translations, then UPDATE categories/parts rows where
* `name = name_original` (still raw English) so already-fetched data
* shows up Turkish on the next refresh. Cache invalidation flushes
* Redis tree caches.
*/
export async function processTranslation(
job: Job<TranslationJobData>,
db: Database,
ai: OpenAI,
redis: Redis,
): Promise<{ translated: number; skipped: number }> {
const { terms } = job.data;
if (!terms?.length) return { translated: 0, skipped: 0 };
// Skip terms already in DB (another worker may have raced ahead)
const existing = await db
.select({ originalName: emexCategoryTranslations.originalName })
.from(emexCategoryTranslations)
.where(inArray(emexCategoryTranslations.originalName, terms));
const existingSet = new Set(existing.map((r) => r.originalName));
const todo = terms.filter((t) => !existingSet.has(t));
if (!todo.length) return { translated: 0, skipped: terms.length };
// LLM call (single batch)
const translations = await callLLM(ai, todo);
// Persist translations
await db
.insert(emexCategoryTranslations)
.values(
todo.map((orig, i) => ({
originalName: orig,
translatedName: translations[i] || orig,
isManual: false,
})),
)
.onConflictDoNothing();
// Update existing categories/parts rows where name was left as English.
// Run as a single statement per term — small N (≤50), and we keep it
// scoped to the two scraper sources.
for (let i = 0; i < todo.length; i++) {
const orig = todo[i];
const tr = translations[i];
if (!tr || tr === orig) continue;
await db.execute(drizzleSql`
UPDATE categories
SET name = ${tr}
WHERE source IN ('emex', 'parts-catalogs')
AND name_original = ${orig}
AND name = name_original
`);
await db.execute(drizzleSql`
UPDATE parts
SET name = ${tr}
WHERE source IN ('emex', 'parts-catalogs')
AND name_original = ${orig}
AND name = name_original
`);
}
// Invalidate translation lookup cache so next read sees DB value.
// Tree cache (cat:tree:*) is intentionally NOT flushed wholesale here
// because that would punish unrelated vehicles; a stale tree just
// expires within 1h and the underlying name is already updated in DB.
const trKeys = todo.map((t) => `tr:${t}`);
if (trKeys.length) {
await redis.del(...trKeys).catch(() => undefined);
}
return { translated: todo.length, skipped: terms.length - todo.length };
}

2
apps/api/src/jobs/queues/catalog-prefetch.queue.ts
View File

@@ -1,6 +1,6 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
export const CATALOG_PREFETCH_QUEUE = "CATALOG_PREFETCH_QUEUE";

2
apps/api/src/jobs/queues/emex-scrape.queue.ts
View File

@@ -1,6 +1,6 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
export const EMEX_SCRAPE_QUEUE = "EMEX_SCRAPE_QUEUE";

2
apps/api/src/jobs/queues/query-cleanup.queue.ts
View File

@@ -1,6 +1,6 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
export const QUERY_CLEANUP_QUEUE = "QUERY_CLEANUP_QUEUE";

2
apps/api/src/jobs/queues/subscription-expiry.queue.ts
View File

@@ -1,6 +1,6 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
export const SUBSCRIPTION_EXPIRY_QUEUE = "SUBSCRIPTION_EXPIRY_QUEUE";

22
apps/api/src/jobs/queues/translation.queue.ts Normal file
View File

@@ -0,0 +1,22 @@
import { Provider } from "@nestjs/common";
import { Queue } from "bullmq";
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
export const TRANSLATION_QUEUE = "TRANSLATION_QUEUE";
export const TranslationQueueProvider: Provider = {
provide: TRANSLATION_QUEUE,
useFactory: () => {
const telemetry = getBullTelemetry();
return new Queue(QUEUE_NAMES.TRANSLATION, {
connection: getBullConnection(),
...(telemetry ? { telemetry } : {}),
defaultJobOptions: {
attempts: 3,
backoff: { type: "exponential", delay: 5000 },
removeOnComplete: { count: 500 },
removeOnFail: { count: 1000 },
},
});
},
};