feat(part-prices): parça kodu bazlı tedarikçi fiyat geçmişi + OEM sayfasında p50/p95/p99 grafiği
- pg: part_price_tracks + part_price_daily (0018) — (kod, kaynak, gün) başına stoktaki tekliflerin p50/p95/p99 + teklif sayısı; source='supplier' şimdilik, perakende ileride aynı tabloya 'retail' olarak girer. Tedarikçi kimliği yok. - API: GET /part-prices/series (ilk istekte takip history'sinden lazy-backfill, sonrası salt-pg + Redis) ve POST /part-prices/current-batch (sayfadaki kodlar için canlı güncel istatistik). P-servisi sözleşmesi: asla throw yok, fail-open. - Worker: part-price-refresh cron'u 19:30 Europe/Istanbul (takip sync'i 19:05'te bitiyor) — izlenen kodlara bugünün satırını upsert eder, sku_map'i artımlı bakar, Redis cache düşürür. SUPPLIER_PRICE_DB_* yoksa sessiz no-op. - Kaynak köprüsü: takip.sku_map (code_norm → product_id; tam sku / ilk-boşluk / ilk-tire sonrası normalize adayları) vmi MySQL'inde kurulu; 6,8M satır. - Web: OEM detayında "Tedarikçi fiyat analizi" kartı (güncel medyan + P95/P99 + teklif sayısı + 30g delta, 30G/90G/Tümü aralıklı step grafik, recharts) ve article/muadil/OE satırlarında fiyat çipi → dialog'da tam geçmiş. - Fix(p): td snapshot'ında gerçek üretici kodu articles.name'de (article_number %96 upstream sayısal ID) — sayfa artık kopyalanabilir gerçek kodu gösteriyor. - compose: SUPPLIER_PRICE_DB_ENABLED/URL api+worker bloklarına eklendi. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
25
apps/api/drizzle/0018_part_price_history.sql
Normal file
25
apps/api/drizzle/0018_part_price_history.sql
Normal file
@@ -0,0 +1,25 @@
|
||||
-- Tedarikçi fiyat geçmişi (parça kodu bazlı, takip MySQL'den beslenir).
|
||||
-- part_price_tracks: hangi kod izleniyor (ilk seriyi API lazy-backfill yazar,
|
||||
-- günlük cron sadece buradaki kodları tazeler). part_price_daily: (kod, kaynak,
|
||||
-- gün) başına stoktaki tekliflerin p50/p95/p99'u + teklif sayısı. source şimdilik
|
||||
-- hep 'supplier'; perakende verisi geldiğinde aynı tabloya 'retail' olarak girer.
|
||||
-- Tedarikçi kimliği bilinçli olarak HİÇBİR kolonda yok.
|
||||
CREATE TABLE "part_price_tracks" (
|
||||
"code_norm" varchar(64) PRIMARY KEY NOT NULL,
|
||||
"first_seen_at" timestamp with time zone DEFAULT now() NOT NULL,
|
||||
"backfilled_at" timestamp with time zone,
|
||||
"last_refreshed_at" timestamp with time zone
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE TABLE "part_price_daily" (
|
||||
"code_norm" varchar(64) NOT NULL,
|
||||
"source" varchar(16) DEFAULT 'supplier' NOT NULL,
|
||||
"date" date NOT NULL,
|
||||
"p50" numeric(15, 4),
|
||||
"p95" numeric(15, 4),
|
||||
"p99" numeric(15, 4),
|
||||
"offer_count" integer DEFAULT 0 NOT NULL,
|
||||
CONSTRAINT "part_price_daily_code_norm_source_date_pk" PRIMARY KEY("code_norm","source","date")
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE INDEX "part_price_daily_code_date_idx" ON "part_price_daily" USING btree ("code_norm","date");
|
||||
@@ -127,6 +127,13 @@
|
||||
"when": 1781395200000,
|
||||
"tag": "0017_oem_expert_rewards",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 18,
|
||||
"version": "7",
|
||||
"when": 1781481600000,
|
||||
"tag": "0018_part_price_history",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -36,6 +36,7 @@ import { MetaCapiModule } from "./meta-capi/meta-capi.module";
|
||||
import { NotificationsModule } from "./notifications/notifications.module";
|
||||
import { OemSuggestionsModule } from "./oem-suggestions/oem-suggestions.module";
|
||||
import { OemVotesModule } from "./oem-votes/oem-votes.module";
|
||||
import { PartPricesModule } from "./part-prices/part-prices.module";
|
||||
import { PartsModule } from "./parts/parts.module";
|
||||
import { PaymentsModule } from "./payments/payments.module";
|
||||
import { PlansModule } from "./plans/plans.module";
|
||||
@@ -90,6 +91,7 @@ import { VehiclesModule } from "./vehicles/vehicles.module";
|
||||
DemoModule,
|
||||
PartsModule,
|
||||
PModule,
|
||||
PartPricesModule,
|
||||
JobsModule,
|
||||
EmexModule,
|
||||
TranslationsModule,
|
||||
|
||||
@@ -96,6 +96,14 @@ export default () => ({
|
||||
enabled: process.env.P_DB_ENABLED === "true",
|
||||
url: process.env.P_DB_URL,
|
||||
},
|
||||
supplierPrices: {
|
||||
// Tedarikçi fiyat geçmişi (takip MySQL, Tailscale). Kapalı/boş URL →
|
||||
// part-prices uçları { matched: false } / boş batch döner; UI fiyat
|
||||
// bölümünü hiç render etmez. Tedarikçi kimliği API cevabına asla çıkmaz —
|
||||
// yalnızca stoktaki tekliflerin p50/p95/p99 dağılımı servis edilir.
|
||||
enabled: process.env.SUPPLIER_PRICE_DB_ENABLED === "true",
|
||||
url: process.env.SUPPLIER_PRICE_DB_URL,
|
||||
},
|
||||
otel: {
|
||||
enabled: process.env.OTEL_ENABLED === "true",
|
||||
endpoint: process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
import {
|
||||
boolean,
|
||||
date,
|
||||
index,
|
||||
integer,
|
||||
jsonb,
|
||||
numeric,
|
||||
pgTable,
|
||||
primaryKey,
|
||||
text,
|
||||
timestamp,
|
||||
uniqueIndex,
|
||||
@@ -770,3 +772,36 @@ export const oemExpertRewards = pgTable(
|
||||
index("oem_expert_rewards_user_id_idx").on(table.userId),
|
||||
],
|
||||
);
|
||||
|
||||
// ─── Part Price History (tedarikçi fiyat verisi) ──────
|
||||
// Parça kodu bazlı piyasa fiyat geçmişi; kaynak takip MySQL'i (Tailscale).
|
||||
// tracks = izlenen kodlar: ilk seriyi API lazy-backfill yazar (kod ilk kez
|
||||
// görüntülendiğinde), günlük cron yalnızca buradaki kodlara bugünün satırını
|
||||
// ekler. daily = (kod, kaynak, gün) başına stoktaki tekliflerin p50/p95/p99
|
||||
// dağılımı + teklif sayısı; sıfır-teklif günleri null fiyat + offerCount 0
|
||||
// ile saklanır ki "stok yok" bilgisi de seriye işlensin. source şimdilik hep
|
||||
// 'supplier'; perakende fiyatı geldiğinde aynı tabloya 'retail' satırları
|
||||
// girer. Tedarikçi kimliği bilinçli olarak hiçbir yerde tutulmaz.
|
||||
export const partPriceTracks = pgTable("part_price_tracks", {
|
||||
codeNorm: varchar("code_norm", { length: 64 }).primaryKey(),
|
||||
firstSeenAt: timestamp("first_seen_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
backfilledAt: timestamp("backfilled_at", { withTimezone: true }),
|
||||
lastRefreshedAt: timestamp("last_refreshed_at", { withTimezone: true }),
|
||||
});
|
||||
|
||||
export const partPriceDaily = pgTable(
|
||||
"part_price_daily",
|
||||
{
|
||||
codeNorm: varchar("code_norm", { length: 64 }).notNull(),
|
||||
source: varchar("source", { length: 16 }).default("supplier").notNull(),
|
||||
date: date("date").notNull(),
|
||||
p50: numeric("p50", { precision: 15, scale: 4 }),
|
||||
p95: numeric("p95", { precision: 15, scale: 4 }),
|
||||
p99: numeric("p99", { precision: 15, scale: 4 }),
|
||||
offerCount: integer("offer_count").default(0).notNull(),
|
||||
},
|
||||
(table) => [
|
||||
primaryKey({ columns: [table.codeNorm, table.source, table.date] }),
|
||||
index("part_price_daily_code_date_idx").on(table.codeNorm, table.date),
|
||||
],
|
||||
);
|
||||
|
||||
@@ -135,8 +135,14 @@ export class PSourceDbService implements OnModuleInit, OnModuleDestroy {
|
||||
SELECT
|
||||
a.id::text AS id,
|
||||
b.name AS brand,
|
||||
a.article_number AS article_number,
|
||||
a.name AS name,
|
||||
-- Snapshot'ta kolonlar şaşırtmacalı: articles.article_number %96
|
||||
-- upstream'in sayısal scrape-ID'si, GERÇEK üretici parça kodu ise
|
||||
-- articles.name kolonunda ("0 092 S40 300", "JTE2311"; 1,1M satırda
|
||||
-- yalnızca 176 istisna). Kullanıcıya kopyalanabilir gerçek kodu servis
|
||||
-- et; name'i ayrıca döndürme (UI'daki ikincil satır zaten kod ile
|
||||
-- aynıysa gizleniyordu).
|
||||
COALESCE(NULLIF(a.name, ''), a.article_number) AS article_number,
|
||||
NULL AS name,
|
||||
a.spare_info AS spare_info,
|
||||
COALESCE((
|
||||
SELECT json_agg(json_build_object('brand', o.brand, 'code', o.code))
|
||||
@@ -167,7 +173,7 @@ export class PSourceDbService implements OnModuleInit, OnModuleDestroy {
|
||||
FROM hit
|
||||
JOIN articles a ON a.id = hit.article_id
|
||||
JOIN article_brands b ON b.id = a.brand_id
|
||||
ORDER BY b.name, a.article_number
|
||||
ORDER BY b.name, COALESCE(NULLIF(a.name, ''), a.article_number)
|
||||
`;
|
||||
|
||||
if (rows.length === 0) return miss;
|
||||
|
||||
@@ -28,4 +28,5 @@ export const QUEUE_NAMES = {
|
||||
TRANSLATION: "translation",
|
||||
LIFECYCLE_EMAIL: "lifecycle-email",
|
||||
EXPERT_REWARDS: "expert-rewards",
|
||||
PART_PRICE_REFRESH: "part-price-refresh",
|
||||
} as const;
|
||||
|
||||
@@ -10,6 +10,10 @@ import {
|
||||
import { EMEX_SCRAPE_QUEUE, EmexScrapeQueueProvider } from "./queues/emex-scrape.queue";
|
||||
import { EXPERT_REWARDS_QUEUE, ExpertRewardsQueueProvider } from "./queues/expert-rewards.queue";
|
||||
import { LIFECYCLE_EMAIL_QUEUE, LifecycleEmailQueueProvider } from "./queues/lifecycle-email.queue";
|
||||
import {
|
||||
PART_PRICE_REFRESH_QUEUE,
|
||||
PartPriceRefreshQueueProvider,
|
||||
} from "./queues/part-price-refresh.queue";
|
||||
import { QUERY_CLEANUP_QUEUE, QueryCleanupQueueProvider } from "./queues/query-cleanup.queue";
|
||||
import {
|
||||
SUBSCRIPTION_EXPIRY_QUEUE,
|
||||
@@ -25,6 +29,7 @@ import {
|
||||
CatalogPrefetchQueueProvider,
|
||||
LifecycleEmailQueueProvider,
|
||||
ExpertRewardsQueueProvider,
|
||||
PartPriceRefreshQueueProvider,
|
||||
PrefetchWorkerService,
|
||||
],
|
||||
exports: [
|
||||
@@ -34,6 +39,7 @@ import {
|
||||
CATALOG_PREFETCH_QUEUE,
|
||||
LIFECYCLE_EMAIL_QUEUE,
|
||||
EXPERT_REWARDS_QUEUE,
|
||||
PART_PRICE_REFRESH_QUEUE,
|
||||
],
|
||||
})
|
||||
export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
@@ -43,6 +49,7 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
@Inject(CATALOG_PREFETCH_QUEUE) private catalogPrefetchQueue: Queue,
|
||||
@Inject(LIFECYCLE_EMAIL_QUEUE) private lifecycleEmailQueue: Queue,
|
||||
@Inject(EXPERT_REWARDS_QUEUE) private expertRewardsQueue: Queue,
|
||||
@Inject(PART_PRICE_REFRESH_QUEUE) private partPriceRefreshQueue: Queue,
|
||||
) {}
|
||||
|
||||
async onModuleInit() {
|
||||
@@ -147,6 +154,25 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
},
|
||||
);
|
||||
console.log("[jobs] Registered expert-rewards cron: 0 0 1 * * (Europe/Istanbul)");
|
||||
|
||||
// Tedarikçi fiyat tazeleme: her gün 19:30 İstanbul — takip MySQL'inin
|
||||
// 18:00 CEST (19:00 İstanbul) sync'i ~5 dk'da bitiyor; sonrasında izlenen
|
||||
// parça kodlarına bugünün p50/p95/p99 satırı yazılır. Dev'de de çalışır:
|
||||
// pg yazımları kendi DB'sine, MySQL tarafı idempotent INSERT IGNORE.
|
||||
// SUPPLIER_PRICE_DB_* env'i yoksa processor sessiz no-op.
|
||||
await this.partPriceRefreshQueue.upsertJobScheduler(
|
||||
"part-price-refresh-daily",
|
||||
{ pattern: "30 19 * * *", tz: "Europe/Istanbul" },
|
||||
{
|
||||
name: "part-price-refresh-run",
|
||||
data: {},
|
||||
opts: {
|
||||
removeOnComplete: { count: 30 },
|
||||
removeOnFail: { count: 100 },
|
||||
},
|
||||
},
|
||||
);
|
||||
console.log("[jobs] Registered part-price-refresh cron: 30 19 * * * (Europe/Istanbul)");
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
@@ -156,6 +182,7 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
this.catalogPrefetchQueue.close(),
|
||||
this.lifecycleEmailQueue.close(),
|
||||
this.expertRewardsQueue.close(),
|
||||
this.partPriceRefreshQueue.close(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
112
apps/api/src/jobs/processors/part-price-refresh.processor.ts
Normal file
112
apps/api/src/jobs/processors/part-price-refresh.processor.ts
Normal file
@@ -0,0 +1,112 @@
|
||||
import { Job } from "bullmq";
|
||||
import { asc, inArray, sql } from "drizzle-orm";
|
||||
import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
|
||||
import Redis from "ioredis";
|
||||
import { partPriceDaily, partPriceTracks } from "../../database/schema/core";
|
||||
import { computeStats, istanbulToday } from "../../part-prices/part-prices.logic";
|
||||
import { createSupplierPriceSource } from "../../part-prices/supplier-price-source";
|
||||
|
||||
type Database = PostgresJsDatabase<Record<string, unknown>>;
|
||||
|
||||
const TRACKED_CAP = 50_000;
|
||||
const BATCH = 300;
|
||||
|
||||
/**
|
||||
* Günlük tedarikçi fiyat tazeleme (19:30 Europe/Istanbul — takip sync'i
|
||||
* ~19:05'te biter):
|
||||
* 1. sku_map artımlı bakım (son 3 günde güncellenen ürünler, INSERT IGNORE).
|
||||
* 2. İzlenen her kod için bugünün p50/p95/p99'unu takip.products'tan hesapla,
|
||||
* part_price_daily'ye upsert et (teklif kalmadıysa null-fiyatlı satır —
|
||||
* "stok yok" da seridir).
|
||||
* 3. Tazelenen kodların Redis cache'ini düşür.
|
||||
* SUPPLIER_PRICE_DB_* yoksa sessiz no-op (dev/prod'da env'le açılır).
|
||||
*/
|
||||
export async function processPartPriceRefresh(
|
||||
job: Job,
|
||||
db: Database,
|
||||
): Promise<{ tracked: number; refreshed: number; skipped: boolean }> {
|
||||
const enabled = process.env.SUPPLIER_PRICE_DB_ENABLED === "true";
|
||||
const url = process.env.SUPPLIER_PRICE_DB_URL;
|
||||
if (!enabled || !url) {
|
||||
console.log(`[part-price-refresh] disabled (enabled=${enabled}, urlSet=${Boolean(url)})`);
|
||||
return { tracked: 0, refreshed: 0, skipped: true };
|
||||
}
|
||||
|
||||
const source = createSupplierPriceSource(url);
|
||||
const redis = new Redis({
|
||||
host: process.env.REDIS_HOST || "localhost",
|
||||
port: Number(process.env.REDIS_PORT) || 6379,
|
||||
password: process.env.REDIS_PASSWORD || undefined,
|
||||
maxRetriesPerRequest: null,
|
||||
});
|
||||
|
||||
try {
|
||||
console.log(`[part-price-refresh] job ${job.id} starting`);
|
||||
await source.refreshSkuMap();
|
||||
|
||||
const tracked = await db
|
||||
.select({ codeNorm: partPriceTracks.codeNorm })
|
||||
.from(partPriceTracks)
|
||||
.where(sql`${partPriceTracks.backfilledAt} IS NOT NULL`)
|
||||
.orderBy(asc(partPriceTracks.codeNorm))
|
||||
.limit(TRACKED_CAP);
|
||||
const today = istanbulToday();
|
||||
let refreshed = 0;
|
||||
|
||||
for (let i = 0; i < tracked.length; i += BATCH) {
|
||||
const chunk = tracked.slice(i, i + BATCH).map((t) => t.codeNorm);
|
||||
const rows = await source.fetchCurrentOfferRows(chunk);
|
||||
const byCode = new Map<string, { price: number; stock: number }[]>();
|
||||
for (const r of rows) {
|
||||
let list = byCode.get(r.codeNorm);
|
||||
if (!list) {
|
||||
list = [];
|
||||
byCode.set(r.codeNorm, list);
|
||||
}
|
||||
list.push({ price: r.price, stock: 1 });
|
||||
}
|
||||
|
||||
const values = chunk.map((codeNorm) => {
|
||||
const stats = computeStats(byCode.get(codeNorm) ?? []);
|
||||
return {
|
||||
codeNorm,
|
||||
source: "supplier" as const,
|
||||
date: today,
|
||||
p50: stats.p50 === null ? null : String(stats.p50),
|
||||
p95: stats.p95 === null ? null : String(stats.p95),
|
||||
p99: stats.p99 === null ? null : String(stats.p99),
|
||||
offerCount: stats.offerCount,
|
||||
};
|
||||
});
|
||||
|
||||
await db
|
||||
.insert(partPriceDaily)
|
||||
.values(values)
|
||||
.onConflictDoUpdate({
|
||||
target: [partPriceDaily.codeNorm, partPriceDaily.source, partPriceDaily.date],
|
||||
set: {
|
||||
p50: sql`excluded.p50`,
|
||||
p95: sql`excluded.p95`,
|
||||
p99: sql`excluded.p99`,
|
||||
offerCount: sql`excluded.offer_count`,
|
||||
},
|
||||
});
|
||||
await db
|
||||
.update(partPriceTracks)
|
||||
.set({ lastRefreshedAt: new Date() })
|
||||
.where(inArray(partPriceTracks.codeNorm, chunk));
|
||||
|
||||
// Cache düşür — bir sonraki sayfa görüntülemesi taze pg satırını okur.
|
||||
const keys = chunk.flatMap((n) => [`partprice:series:v1:${n}`, `partprice:cur:v1:${n}`]);
|
||||
if (keys.length > 0) await redis.del(...keys);
|
||||
|
||||
refreshed += chunk.length;
|
||||
console.log(`[part-price-refresh] ${refreshed}/${tracked.length} refreshed`);
|
||||
}
|
||||
|
||||
return { tracked: tracked.length, refreshed, skipped: false };
|
||||
} finally {
|
||||
await source.end().catch(() => {});
|
||||
redis.disconnect();
|
||||
}
|
||||
}
|
||||
25
apps/api/src/jobs/queues/part-price-refresh.queue.ts
Normal file
25
apps/api/src/jobs/queues/part-price-refresh.queue.ts
Normal file
@@ -0,0 +1,25 @@
|
||||
import { Provider } from "@nestjs/common";
|
||||
import { Queue } from "bullmq";
|
||||
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
|
||||
|
||||
export const PART_PRICE_REFRESH_QUEUE = "PART_PRICE_REFRESH_QUEUE";
|
||||
|
||||
export const PartPriceRefreshQueueProvider: Provider = {
|
||||
provide: PART_PRICE_REFRESH_QUEUE,
|
||||
useFactory: () => {
|
||||
const telemetry = getBullTelemetry();
|
||||
return new Queue(QUEUE_NAMES.PART_PRICE_REFRESH, {
|
||||
connection: getBullConnection(),
|
||||
...(telemetry ? { telemetry } : {}),
|
||||
defaultJobOptions: {
|
||||
attempts: 3,
|
||||
backoff: {
|
||||
type: "exponential",
|
||||
delay: 60000,
|
||||
},
|
||||
removeOnComplete: { count: 30 },
|
||||
removeOnFail: { count: 100 },
|
||||
},
|
||||
});
|
||||
},
|
||||
};
|
||||
28
apps/api/src/part-prices/part-prices.controller.ts
Normal file
28
apps/api/src/part-prices/part-prices.controller.ts
Normal file
@@ -0,0 +1,28 @@
|
||||
import { Body, Controller, Get, Post, Query } from "@nestjs/common";
|
||||
import { PartPricesService } from "./part-prices.service";
|
||||
|
||||
@Controller("part-prices")
|
||||
export class PartPricesController {
|
||||
constructor(private readonly partPrices: PartPricesService) {}
|
||||
|
||||
/**
|
||||
* Bir parça kodunun tedarikçi fiyat serisi (p50/p95/p99, günlük değişim
|
||||
* noktaları). `GET /part-prices/series?code=0986452041` → her durumda 200;
|
||||
* eşleşmeme/kapalı kaynak `matched: false`. İlk istek lazy-backfill yapar.
|
||||
*/
|
||||
@Get("series")
|
||||
async series(@Query("code") code: string) {
|
||||
return this.partPrices.getSeries(code ?? "");
|
||||
}
|
||||
|
||||
/**
|
||||
* Sayfada görünen kodların güncel istatistikleri (tek istek).
|
||||
* `POST /part-prices/current-batch { codes: [...] }` →
|
||||
* `{ prices: { CODENORM: { p50, p95, p99, offerCount } } }` — eşleşmeyen
|
||||
* kodlar haritaya girmez, UI o satıra fiyat çizmez.
|
||||
*/
|
||||
@Post("current-batch")
|
||||
async currentBatch(@Body("codes") codes: string[]) {
|
||||
return this.partPrices.getCurrentBatch(Array.isArray(codes) ? codes : []);
|
||||
}
|
||||
}
|
||||
126
apps/api/src/part-prices/part-prices.logic.spec.ts
Normal file
126
apps/api/src/part-prices/part-prices.logic.spec.ts
Normal file
@@ -0,0 +1,126 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
type HistoryEvent,
|
||||
type SupplierOffer,
|
||||
computeStats,
|
||||
istanbulToday,
|
||||
normPartCode,
|
||||
percentile,
|
||||
reconstructDailySeries,
|
||||
} from "./part-prices.logic";
|
||||
|
||||
describe("normPartCode", () => {
|
||||
it("uppercases and strips non-alphanumerics (web normCode ile aynı kural)", () => {
|
||||
expect(normPartCode("1j0 973-702")).toBe("1J0973702");
|
||||
expect(normPartCode("62 92 9423")).toBe("62929423");
|
||||
expect(normPartCode(" ")).toBe("");
|
||||
});
|
||||
});
|
||||
|
||||
describe("percentile", () => {
|
||||
it("interpolates linearly (PERCENTILE_CONT)", () => {
|
||||
expect(percentile([10], 99)).toBe(10);
|
||||
expect(percentile([10, 20], 50)).toBe(15);
|
||||
expect(percentile([1, 2, 3, 4, 5], 50)).toBe(3);
|
||||
expect(percentile([1, 2, 3, 4, 5], 95)).toBeCloseTo(4.8);
|
||||
});
|
||||
});
|
||||
|
||||
describe("computeStats", () => {
|
||||
it("filters out-of-stock and zero-priced offers", () => {
|
||||
const stats = computeStats([
|
||||
{ price: 100, stock: 2 },
|
||||
{ price: 0, stock: 5 },
|
||||
{ price: 200, stock: 0 },
|
||||
{ price: 300, stock: 1 },
|
||||
]);
|
||||
expect(stats).toEqual({ p50: 200, p95: 290, p99: 298, offerCount: 2 });
|
||||
});
|
||||
|
||||
it("returns null prices when nothing is in stock", () => {
|
||||
expect(computeStats([{ price: 100, stock: 0 }])).toEqual({
|
||||
p50: null,
|
||||
p95: null,
|
||||
p99: null,
|
||||
offerCount: 0,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("reconstructDailySeries", () => {
|
||||
const W = "2026-01-29";
|
||||
const TODAY = "2026-06-12";
|
||||
|
||||
it("hiç event yoksa güncel teklifleri pencere başından beri sabit sayar", () => {
|
||||
const current: SupplierOffer[] = [
|
||||
{ productId: 1, price: 100, stock: 3 },
|
||||
{ productId: 2, price: 140, stock: 1 },
|
||||
];
|
||||
const series = reconstructDailySeries([], current, W, TODAY);
|
||||
expect(series).toHaveLength(1);
|
||||
expect(series[0]).toMatchObject({ date: W, p50: 120, offerCount: 2 });
|
||||
});
|
||||
|
||||
it("ilk event'in prev değerleri baseline olur, takibe yeni girenler olmaz", () => {
|
||||
const events: HistoryEvent[] = [
|
||||
// 1 numara fiyat değiştirdi: 100 → 150
|
||||
{ productId: 1, date: "2026-03-01", price: 150, stock: 2, prevPrice: 100, prevStock: 2 },
|
||||
// 2 numara o gün takibe girdi (prev'ler 0) → baseline'da yok
|
||||
{ productId: 2, date: "2026-03-01", price: 200, stock: 1, prevPrice: 0, prevStock: 0 },
|
||||
];
|
||||
const current: SupplierOffer[] = [
|
||||
{ productId: 1, price: 150, stock: 2 },
|
||||
{ productId: 2, price: 200, stock: 1 },
|
||||
];
|
||||
const series = reconstructDailySeries(events, current, W, TODAY);
|
||||
expect(series[0]).toMatchObject({ date: W, p50: 100, offerCount: 1 });
|
||||
expect(series[1]).toMatchObject({ date: "2026-03-01", p50: 175, offerCount: 2 });
|
||||
// current son durumla aynı → bugün için fazladan nokta yok
|
||||
expect(series).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("stok bitişini null-fiyatlı nokta olarak işler ve aynı istatistiği tekrarlamaz", () => {
|
||||
const events: HistoryEvent[] = [
|
||||
{ productId: 1, date: "2026-02-10", price: 100, stock: 0, prevPrice: 100, prevStock: 5 },
|
||||
// istatistiği değiştirmeyen event (fiyat aynı, hâlâ stok dışı)
|
||||
{ productId: 1, date: "2026-02-20", price: 110, stock: 0, prevPrice: 100, prevStock: 0 },
|
||||
];
|
||||
const current: SupplierOffer[] = [{ productId: 1, price: 110, stock: 0 }];
|
||||
const series = reconstructDailySeries(events, current, W, TODAY);
|
||||
expect(series[0]).toMatchObject({ date: W, p50: 100, offerCount: 1 });
|
||||
expect(series[1]).toMatchObject({ date: "2026-02-10", p50: null, offerCount: 0 });
|
||||
expect(series).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("current ile history çelişirse bugünü current'a göre düzeltir", () => {
|
||||
const events: HistoryEvent[] = [
|
||||
{ productId: 1, date: "2026-04-01", price: 90, stock: 1, prevPrice: 80, prevStock: 1 },
|
||||
];
|
||||
// history'nin bilmediği daha taze gerçek
|
||||
const current: SupplierOffer[] = [{ productId: 1, price: 95, stock: 2 }];
|
||||
const series = reconstructDailySeries(events, current, W, TODAY);
|
||||
const last = series[series.length - 1];
|
||||
expect(last).toMatchObject({ date: TODAY, p50: 95, offerCount: 1 });
|
||||
});
|
||||
|
||||
it("aynı güne birden çok event tek nokta üretir", () => {
|
||||
const events: HistoryEvent[] = [
|
||||
{ productId: 1, date: "2026-05-05", price: 100, stock: 1, prevPrice: 0, prevStock: 0 },
|
||||
{ productId: 2, date: "2026-05-05", price: 300, stock: 1, prevPrice: 0, prevStock: 0 },
|
||||
];
|
||||
const current: SupplierOffer[] = [
|
||||
{ productId: 1, price: 100, stock: 1 },
|
||||
{ productId: 2, price: 300, stock: 1 },
|
||||
];
|
||||
const series = reconstructDailySeries(events, current, W, TODAY);
|
||||
expect(series).toHaveLength(1);
|
||||
expect(series[0]).toMatchObject({ date: "2026-05-05", p50: 200, offerCount: 2 });
|
||||
});
|
||||
});
|
||||
|
||||
describe("istanbulToday", () => {
|
||||
it("UTC gece yarısı civarında İstanbul gününü döner", () => {
|
||||
// 2026-06-11 22:30 UTC = 2026-06-12 01:30 İstanbul
|
||||
expect(istanbulToday(new Date("2026-06-11T22:30:00Z"))).toBe("2026-06-12");
|
||||
});
|
||||
});
|
||||
173
apps/api/src/part-prices/part-prices.logic.ts
Normal file
173
apps/api/src/part-prices/part-prices.logic.ts
Normal file
@@ -0,0 +1,173 @@
|
||||
/**
|
||||
* Tedarikçi fiyat serisi saf mantığı — DB'siz, yan etkisiz.
|
||||
*
|
||||
* Kaynak model (takip MySQL):
|
||||
* - `products` : güncel durum (ürün başına son fiyat + stok; her gün 19:00
|
||||
* İstanbul'da sync biter).
|
||||
* - `product_history`: DEĞİŞİKLİK logu — bir ürünün satırı yalnızca fiyatı
|
||||
* veya stoğu değiştiği gün yazılır (prev_* bir önceki değer;
|
||||
* ürün takibe o gün girdiyse prev'ler 0). Tam-gün snapshot
|
||||
* YOKTUR; günlük durum carry-forward ile kurulur.
|
||||
*
|
||||
* Bir parça kodunun günlük istatistiği: o gün stokta (stock > 0) ve geçerli
|
||||
* fiyatlı (price > 0) tedarikçi tekliflerinin p50/p95/p99'u + teklif sayısı.
|
||||
* Teklif yoksa fiyatlar null, offerCount 0 — "stok yok" da seriye işlenir.
|
||||
*/
|
||||
|
||||
export const PART_CODE_MIN_NORM_LEN = 5;
|
||||
|
||||
/** Web'in normCode'u ve P servisinin norm'uyla birebir aynı kural:
|
||||
* `1J0 973 702` / `1j0-973-702` → `1J0973702`. */
|
||||
export function normPartCode(code: string): string {
|
||||
return (code ?? "").toUpperCase().replace(/[^A-Z0-9]/g, "");
|
||||
}
|
||||
|
||||
export interface SupplierOffer {
|
||||
productId: number;
|
||||
price: number;
|
||||
stock: number;
|
||||
}
|
||||
|
||||
export interface HistoryEvent {
|
||||
productId: number;
|
||||
/** YYYY-MM-DD */
|
||||
date: string;
|
||||
price: number;
|
||||
stock: number;
|
||||
prevPrice: number;
|
||||
prevStock: number;
|
||||
}
|
||||
|
||||
export interface DailyPoint {
|
||||
/** YYYY-MM-DD */
|
||||
date: string;
|
||||
p50: number | null;
|
||||
p95: number | null;
|
||||
p99: number | null;
|
||||
offerCount: number;
|
||||
}
|
||||
|
||||
/** PERCENTILE_CONT (lineer interpolasyon). `sorted` artan sıralı ve boş değil. */
|
||||
export function percentile(sorted: number[], p: number): number {
|
||||
const n = sorted.length;
|
||||
if (n === 1) return sorted[0];
|
||||
const rank = (p / 100) * (n - 1);
|
||||
const lo = Math.floor(rank);
|
||||
const hi = Math.ceil(rank);
|
||||
if (lo === hi) return sorted[lo];
|
||||
return sorted[lo] + (sorted[hi] - sorted[lo]) * (rank - lo);
|
||||
}
|
||||
|
||||
const round2 = (v: number) => Math.round(v * 100) / 100;
|
||||
|
||||
/** Geçerli (price>0, stock>0) tekliflerden günlük istatistik. Teklif yoksa
|
||||
* null-fiyatlı sıfır satırı döner (seriye "stok yok" olarak işlenir). */
|
||||
export function computeStats(offers: Iterable<{ price: number; stock: number }>): {
|
||||
p50: number | null;
|
||||
p95: number | null;
|
||||
p99: number | null;
|
||||
offerCount: number;
|
||||
} {
|
||||
const prices: number[] = [];
|
||||
for (const o of offers) {
|
||||
if (o.stock > 0 && o.price > 0) prices.push(o.price);
|
||||
}
|
||||
if (prices.length === 0) return { p50: null, p95: null, p99: null, offerCount: 0 };
|
||||
prices.sort((a, b) => a - b);
|
||||
return {
|
||||
p50: round2(percentile(prices, 50)),
|
||||
p95: round2(percentile(prices, 95)),
|
||||
p99: round2(percentile(prices, 99)),
|
||||
offerCount: prices.length,
|
||||
};
|
||||
}
|
||||
|
||||
const sameStats = (a: DailyPoint, b: DailyPoint) =>
|
||||
a.p50 === b.p50 && a.p95 === b.p95 && a.p99 === b.p99 && a.offerCount === b.offerCount;
|
||||
|
||||
/**
|
||||
* Change-log'dan sıkıştırılmış günlük seri kurar: yalnızca istatistiğin
|
||||
* değiştiği günler için nokta üretir (grafik step-after çizer, ara günler
|
||||
* carry-forward'dur).
|
||||
*
|
||||
* - Bir ürünün ilk event'inden ÖNCEKİ durumu = (prevPrice, prevStock); ürün
|
||||
* takibe o gün girdiyse (prev'ler 0) öncesinde piyasada yok sayılır.
|
||||
* - Hiç event'i olmayan ürünler tracking başından beri değişmemiştir →
|
||||
* güncel değerleri tüm pencere boyunca sabittir.
|
||||
* - Son nokta her zaman `current` (products tablosu = bugünün gerçeği) ile
|
||||
* tutarlı hale getirilir; fark varsa `todayIso` tarihli nokta eklenir.
|
||||
*
|
||||
* `windowStart` = takip'in veri başlangıcı; baseline (event-öncesi) durumun
|
||||
* tarihi olarak kullanılır. Event'ler tarih artan sırada gelmeli (SQL ORDER BY).
|
||||
*/
|
||||
export function reconstructDailySeries(
|
||||
events: HistoryEvent[],
|
||||
current: SupplierOffer[],
|
||||
windowStart: string,
|
||||
todayIso: string,
|
||||
): DailyPoint[] {
|
||||
// Ürün başına güncel durum — event'i olmayan ürünlerin sabit değeri ve
|
||||
// baseline'da "var mı yok mu" kararı için.
|
||||
const state = new Map<number, { price: number; stock: number }>();
|
||||
const hasEvents = new Set<number>();
|
||||
for (const e of events) hasEvents.add(e.productId);
|
||||
for (const c of current) {
|
||||
if (!hasEvents.has(c.productId)) state.set(c.productId, { price: c.price, stock: c.stock });
|
||||
}
|
||||
|
||||
// Baseline: event'li ürünler ilk event'lerinin prev değerleriyle başlar
|
||||
// (prev'leri 0 olanlar — takibe sonradan girenler — baseline'da yok).
|
||||
const firstSeen = new Set<number>();
|
||||
for (const e of events) {
|
||||
if (firstSeen.has(e.productId)) continue;
|
||||
firstSeen.add(e.productId);
|
||||
if (e.prevPrice > 0 || e.prevStock > 0) {
|
||||
state.set(e.productId, { price: e.prevPrice, stock: e.prevStock });
|
||||
}
|
||||
}
|
||||
|
||||
const series: DailyPoint[] = [];
|
||||
const emit = (date: string) => {
|
||||
const point: DailyPoint = { date, ...computeStats(state.values()) };
|
||||
const last = series[series.length - 1];
|
||||
if (last && sameStats(last, point)) return;
|
||||
if (last && last.date === date) {
|
||||
series[series.length - 1] = point;
|
||||
return;
|
||||
}
|
||||
series.push(point);
|
||||
};
|
||||
|
||||
// Baseline'da en az bir geçerli teklif varsa pencere başına nokta koy.
|
||||
const firstEventDate = events[0]?.date;
|
||||
if (firstEventDate !== windowStart && computeStats(state.values()).offerCount > 0) {
|
||||
emit(windowStart);
|
||||
}
|
||||
|
||||
let cursor: string | null = null;
|
||||
for (const e of events) {
|
||||
if (cursor !== null && e.date !== cursor) emit(cursor);
|
||||
cursor = e.date;
|
||||
state.set(e.productId, { price: e.price, stock: e.stock });
|
||||
}
|
||||
if (cursor !== null) emit(cursor);
|
||||
|
||||
// Bugünün gerçeği: products tablosu. History'den türeyen son durumla
|
||||
// çelişiyorsa (ör. history LIMIT'e takıldı ya da sync arası) bugünü düzelt.
|
||||
state.clear();
|
||||
for (const c of current) state.set(c.productId, { price: c.price, stock: c.stock });
|
||||
const todayPoint: DailyPoint = { date: todayIso, ...computeStats(state.values()) };
|
||||
const last = series[series.length - 1];
|
||||
if (!last || !sameStats(last, todayPoint)) {
|
||||
if (last && last.date === todayIso) series[series.length - 1] = todayPoint;
|
||||
else if (todayPoint.offerCount > 0 || last) series.push(todayPoint);
|
||||
}
|
||||
|
||||
return series;
|
||||
}
|
||||
|
||||
/** Bugünün YYYY-MM-DD'si, İstanbul takvimine göre (takip sync'i 19:00 İstanbul
|
||||
* civarı biter; gece yarısı kaymalarında gün etiketi TR gününe sabitlenir). */
|
||||
export function istanbulToday(now: Date = new Date()): string {
|
||||
return now.toLocaleDateString("en-CA", { timeZone: "Europe/Istanbul" });
|
||||
}
|
||||
16
apps/api/src/part-prices/part-prices.module.ts
Normal file
16
apps/api/src/part-prices/part-prices.module.ts
Normal file
@@ -0,0 +1,16 @@
|
||||
import { Module } from "@nestjs/common";
|
||||
import { PartPricesController } from "./part-prices.controller";
|
||||
import { PartPricesService } from "./part-prices.service";
|
||||
|
||||
/**
|
||||
* Parça kodu bazlı tedarikçi fiyat görünümleri (takip MySQL → pg fiyat
|
||||
* geçmişi). SUPPLIER_PRICE_DB_* env'i yokken kendini devre dışı bırakır;
|
||||
* uçlar boş görünüm döner, OEM detay sayfası fiyat bölümünü çizmez.
|
||||
* (Database + Redis modülleri @Global — import gerekmez.)
|
||||
*/
|
||||
@Module({
|
||||
controllers: [PartPricesController],
|
||||
providers: [PartPricesService],
|
||||
exports: [PartPricesService],
|
||||
})
|
||||
export class PartPricesModule {}
|
||||
285
apps/api/src/part-prices/part-prices.service.ts
Normal file
285
apps/api/src/part-prices/part-prices.service.ts
Normal file
@@ -0,0 +1,285 @@
|
||||
import {
|
||||
Inject,
|
||||
Injectable,
|
||||
Logger,
|
||||
type OnModuleDestroy,
|
||||
type OnModuleInit,
|
||||
} from "@nestjs/common";
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
import { asc, eq } from "drizzle-orm";
|
||||
import { DATABASE, type Database } from "../database/database.provider";
|
||||
import { partPriceDaily, partPriceTracks } from "../database/schema/core";
|
||||
import { RedisService } from "../redis/redis.service";
|
||||
import {
|
||||
type DailyPoint,
|
||||
PART_CODE_MIN_NORM_LEN,
|
||||
computeStats,
|
||||
istanbulToday,
|
||||
normPartCode,
|
||||
reconstructDailySeries,
|
||||
} from "./part-prices.logic";
|
||||
import { type SupplierPriceSource, createSupplierPriceSource } from "./supplier-price-source";
|
||||
|
||||
/** takip tracking'inin veri başlangıcı — baseline noktasının tarihi. */
|
||||
export const SUPPLIER_PRICE_EPOCH = "2026-01-29";
|
||||
|
||||
export interface PartPriceSeriesView {
|
||||
matched: boolean;
|
||||
codeNorm: string;
|
||||
currency: "TRY";
|
||||
source: "supplier";
|
||||
series: DailyPoint[];
|
||||
/** Serinin son noktası (güncel alan bunu gösterir). */
|
||||
latest: DailyPoint | null;
|
||||
truncated: boolean;
|
||||
}
|
||||
|
||||
export interface PartPriceCurrent {
|
||||
p50: number;
|
||||
p95: number;
|
||||
p99: number;
|
||||
offerCount: number;
|
||||
}
|
||||
|
||||
export interface PartPriceBatchView {
|
||||
/** code_norm → güncel istatistik (eşleşmeyen kodlar haritada yer almaz). */
|
||||
prices: Record<string, PartPriceCurrent>;
|
||||
}
|
||||
|
||||
const SERIES_CACHE_TTL = 3600; // 1h — cron 19:30'da tazeler, anahtarları siler
|
||||
const SERIES_MISS_TTL = 1800;
|
||||
const BATCH_CACHE_TTL = 1800;
|
||||
const MAX_BATCH_CODES = 400;
|
||||
|
||||
const seriesKey = (norm: string) => `partprice:series:v1:${norm}`;
|
||||
const currentKey = (norm: string) => `partprice:cur:v1:${norm}`;
|
||||
|
||||
/**
|
||||
* Parça kodu bazlı tedarikçi fiyat görünümleri. P-servisi sözleşmesi: asla
|
||||
* throw etmez — kapalı kaynak, kısa kod, bağlantı hatası ve eşleşmeme hepsi
|
||||
* boş görünüme düşer (UI tek empty-state yolu görür).
|
||||
*
|
||||
* - Seri: kod ilk kez istendiğinde takip history'sinden lazy-backfill edilir
|
||||
* ve pg'ye (part_price_tracks + part_price_daily) kalıcı yazılır; sonraki
|
||||
* istekler salt-pg okur. Günlük cron izlenen kodlara bugünü ekler.
|
||||
* - Batch (sayfadaki kod listesi): canlı MySQL'den tek sorgu + Redis cache;
|
||||
* pg'ye iz BIRAKMAZ (tracking yalnızca seri isteğiyle başlar — sayfa başına
|
||||
* yüzlerce kodu sonsuza dek cron'lamamak için).
|
||||
*/
|
||||
@Injectable()
|
||||
export class PartPricesService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly logger = new Logger(PartPricesService.name);
|
||||
private source: SupplierPriceSource | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly config: ConfigService,
|
||||
private readonly redis: RedisService,
|
||||
@Inject(DATABASE) private readonly db: Database,
|
||||
) {}
|
||||
|
||||
onModuleInit() {
|
||||
const enabled = this.config.get<boolean>("supplierPrices.enabled");
|
||||
const url = this.config.get<string>("supplierPrices.url");
|
||||
if (!enabled || !url) {
|
||||
this.logger.log(`[part-prices] disabled (enabled=${enabled}, urlSet=${Boolean(url)})`);
|
||||
return;
|
||||
}
|
||||
this.source = createSupplierPriceSource(url);
|
||||
this.logger.log("[part-prices] supplier price source connected");
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
if (this.source) {
|
||||
await this.source.end().catch(() => {});
|
||||
this.source = null;
|
||||
}
|
||||
}
|
||||
|
||||
private miss(codeNorm: string): PartPriceSeriesView {
|
||||
return {
|
||||
matched: false,
|
||||
codeNorm,
|
||||
currency: "TRY",
|
||||
source: "supplier",
|
||||
series: [],
|
||||
latest: null,
|
||||
truncated: false,
|
||||
};
|
||||
}
|
||||
|
||||
async getSeries(rawCode: string): Promise<PartPriceSeriesView> {
|
||||
const codeNorm = normPartCode(rawCode);
|
||||
if (codeNorm.length < PART_CODE_MIN_NORM_LEN) return this.miss(codeNorm);
|
||||
|
||||
const cached = await this.redis.getJson<PartPriceSeriesView>(seriesKey(codeNorm));
|
||||
if (cached) return cached;
|
||||
|
||||
try {
|
||||
// İzlenen kod → pg'den oku (hızlı yol; cron güncel tutuyor).
|
||||
const [track] = await this.db
|
||||
.select()
|
||||
.from(partPriceTracks)
|
||||
.where(eq(partPriceTracks.codeNorm, codeNorm))
|
||||
.limit(1);
|
||||
|
||||
let view: PartPriceSeriesView;
|
||||
if (track?.backfilledAt) {
|
||||
const rows = await this.db
|
||||
.select()
|
||||
.from(partPriceDaily)
|
||||
.where(eq(partPriceDaily.codeNorm, codeNorm))
|
||||
.orderBy(asc(partPriceDaily.date));
|
||||
const series: DailyPoint[] = rows
|
||||
.filter((r) => r.source === "supplier")
|
||||
.map((r) => ({
|
||||
date: r.date,
|
||||
p50: r.p50 === null ? null : Number(r.p50),
|
||||
p95: r.p95 === null ? null : Number(r.p95),
|
||||
p99: r.p99 === null ? null : Number(r.p99),
|
||||
offerCount: r.offerCount,
|
||||
}));
|
||||
view = {
|
||||
matched: series.length > 0,
|
||||
codeNorm,
|
||||
currency: "TRY",
|
||||
source: "supplier",
|
||||
series,
|
||||
latest: series[series.length - 1] ?? null,
|
||||
truncated: false,
|
||||
};
|
||||
} else {
|
||||
view = await this.backfill(codeNorm);
|
||||
}
|
||||
|
||||
await this.redis.setJson(
|
||||
seriesKey(codeNorm),
|
||||
view,
|
||||
view.matched ? SERIES_CACHE_TTL : SERIES_MISS_TTL,
|
||||
);
|
||||
return view;
|
||||
} catch (err) {
|
||||
this.logger.warn(`[part-prices] series failed (code=${codeNorm}): ${(err as Error).message}`);
|
||||
return this.miss(codeNorm);
|
||||
}
|
||||
}
|
||||
|
||||
/** İlk görüntülenme: takip history'sinden seriyi kur, pg'ye kalıcı yaz. */
|
||||
private async backfill(codeNorm: string): Promise<PartPriceSeriesView> {
|
||||
if (!this.source) return this.miss(codeNorm);
|
||||
|
||||
const { ids, truncated: idsTruncated } = await this.source.fetchProductIds(codeNorm);
|
||||
if (ids.length === 0) return this.miss(codeNorm);
|
||||
|
||||
const [{ events, truncated: histTruncated }, current] = await Promise.all([
|
||||
this.source.fetchHistory(ids),
|
||||
this.source.fetchCurrentOffers(ids),
|
||||
]);
|
||||
|
||||
const series = reconstructDailySeries(events, current, SUPPLIER_PRICE_EPOCH, istanbulToday());
|
||||
if (series.length === 0) return this.miss(codeNorm);
|
||||
|
||||
// Kalıcılaştır — yarış durumunda (iki istek aynı anda backfill eder)
|
||||
// ON CONFLICT'ler ikinci yazımı sessizce yutar.
|
||||
await this.db
|
||||
.insert(partPriceTracks)
|
||||
.values({ codeNorm, backfilledAt: new Date() })
|
||||
.onConflictDoUpdate({
|
||||
target: partPriceTracks.codeNorm,
|
||||
set: { backfilledAt: new Date() },
|
||||
});
|
||||
const rows = series.map((pt) => ({
|
||||
codeNorm,
|
||||
source: "supplier" as const,
|
||||
date: pt.date,
|
||||
p50: pt.p50 === null ? null : String(pt.p50),
|
||||
p95: pt.p95 === null ? null : String(pt.p95),
|
||||
p99: pt.p99 === null ? null : String(pt.p99),
|
||||
offerCount: pt.offerCount,
|
||||
}));
|
||||
for (let i = 0; i < rows.length; i += 500) {
|
||||
await this.db
|
||||
.insert(partPriceDaily)
|
||||
.values(rows.slice(i, i + 500))
|
||||
.onConflictDoNothing();
|
||||
}
|
||||
|
||||
return {
|
||||
matched: true,
|
||||
codeNorm,
|
||||
currency: "TRY",
|
||||
source: "supplier",
|
||||
series,
|
||||
latest: series[series.length - 1] ?? null,
|
||||
truncated: idsTruncated || histTruncated,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Sayfadaki kodlar için güncel istatistik. Canlı MySQL + Redis; pg'ye
|
||||
* dokunmaz. Kaynak kapalı/ulaşılamaz → boş harita (fail-open).
|
||||
*/
|
||||
async getCurrentBatch(rawCodes: string[]): Promise<PartPriceBatchView> {
|
||||
const norms = [
|
||||
...new Set(
|
||||
(Array.isArray(rawCodes) ? rawCodes : [])
|
||||
.filter((c): c is string => typeof c === "string")
|
||||
.map(normPartCode)
|
||||
.filter((n) => n.length >= PART_CODE_MIN_NORM_LEN && n.length <= 64),
|
||||
),
|
||||
].slice(0, MAX_BATCH_CODES);
|
||||
if (norms.length === 0) return { prices: {} };
|
||||
|
||||
const prices: Record<string, PartPriceCurrent> = {};
|
||||
const pending: string[] = [];
|
||||
|
||||
await Promise.all(
|
||||
norms.map(async (n) => {
|
||||
const hit = await this.redis.getJson<PartPriceCurrent | { miss: true }>(currentKey(n));
|
||||
if (hit === null) {
|
||||
pending.push(n);
|
||||
} else if (!("miss" in hit)) {
|
||||
prices[n] = hit;
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
if (pending.length > 0 && this.source) {
|
||||
try {
|
||||
const rows = await this.source.fetchCurrentOfferRows(pending);
|
||||
const byCode = new Map<string, { price: number; stock: number }[]>();
|
||||
for (const r of rows) {
|
||||
let list = byCode.get(r.codeNorm);
|
||||
if (!list) {
|
||||
list = [];
|
||||
byCode.set(r.codeNorm, list);
|
||||
}
|
||||
list.push({ price: r.price, stock: 1 });
|
||||
}
|
||||
await Promise.all(
|
||||
pending.map(async (n) => {
|
||||
const offers = byCode.get(n);
|
||||
const stats = offers ? computeStats(offers) : null;
|
||||
if (stats && stats.p50 !== null) {
|
||||
const current: PartPriceCurrent = {
|
||||
p50: stats.p50,
|
||||
p95: stats.p95 as number,
|
||||
p99: stats.p99 as number,
|
||||
offerCount: stats.offerCount,
|
||||
};
|
||||
prices[n] = current;
|
||||
await this.redis.setJson(currentKey(n), current, BATCH_CACHE_TTL);
|
||||
} else {
|
||||
await this.redis.setJson(currentKey(n), { miss: true }, BATCH_CACHE_TTL);
|
||||
}
|
||||
}),
|
||||
);
|
||||
} catch (err) {
|
||||
this.logger.warn(
|
||||
`[part-prices] batch failed (${pending.length} codes): ${(err as Error).message}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return { prices };
|
||||
}
|
||||
}
|
||||
125
apps/api/src/part-prices/supplier-price-source.ts
Normal file
125
apps/api/src/part-prices/supplier-price-source.ts
Normal file
@@ -0,0 +1,125 @@
|
||||
import mysql, { type Pool, type RowDataPacket } from "mysql2/promise";
|
||||
import type { HistoryEvent, SupplierOffer } from "./part-prices.logic";
|
||||
|
||||
/**
|
||||
* takip MySQL'ine (Tailscale, read-mostly) framework-bağımsız erişim — Nest
|
||||
* servisi de worker processor'ı da bunu kullanır (Novu istemcisi kalıbı).
|
||||
*
|
||||
* Sorgu yolu: `sku_map(code_norm → product_id)` köprü tablosu (vmi üzerinde
|
||||
* kurulu; tedarikçi SKU'larının "tam / ilk-boşluk-sonrası / ilk-tire-sonrası"
|
||||
* normalize adaylarını indeksler) → `products` (güncel fiyat+stok) ve
|
||||
* `product_history` (değişiklik logu). Tek yazma işi `refreshSkuMap` —
|
||||
* idempotent INSERT IGNORE, son 3 günde güncellenen ürünleri haritaya ekler.
|
||||
*/
|
||||
|
||||
// Bir kod çok genel olduğunda (kampanya/jenerik SKU çakışmaları) sorguyu
|
||||
// sınırlamak için tavanlar — aşılırsa seri "truncated" işaretlenir.
|
||||
const MAX_PRODUCTS_PER_CODE = 2000;
|
||||
const MAX_HISTORY_ROWS = 150_000;
|
||||
const MAX_BATCH_OFFER_ROWS = 60_000;
|
||||
|
||||
export interface CurrentStatsRow {
|
||||
codeNorm: string;
|
||||
price: number;
|
||||
}
|
||||
|
||||
export interface SupplierPriceSource {
|
||||
fetchProductIds(codeNorm: string): Promise<{ ids: number[]; truncated: boolean }>;
|
||||
fetchCurrentOffers(ids: number[]): Promise<SupplierOffer[]>;
|
||||
fetchHistory(ids: number[]): Promise<{ events: HistoryEvent[]; truncated: boolean }>;
|
||||
/** Batch: stoktaki tekliflerin (code_norm, price) satırları — istatistik JS'te. */
|
||||
fetchCurrentOfferRows(codeNorms: string[]): Promise<CurrentStatsRow[]>;
|
||||
/** sku_map artımlı bakım: son 3 günde güncellenen ürünleri haritaya ekler. */
|
||||
refreshSkuMap(): Promise<void>;
|
||||
end(): Promise<void>;
|
||||
}
|
||||
|
||||
export function createSupplierPriceSource(url: string): SupplierPriceSource {
|
||||
const pool: Pool = mysql.createPool({
|
||||
uri: url,
|
||||
connectionLimit: 5,
|
||||
connectTimeout: 10_000,
|
||||
waitForConnections: true,
|
||||
});
|
||||
|
||||
return {
|
||||
async fetchProductIds(codeNorm) {
|
||||
const [rows] = await pool.query<RowDataPacket[]>(
|
||||
"SELECT product_id FROM sku_map WHERE code_norm = ? LIMIT ?",
|
||||
[codeNorm, MAX_PRODUCTS_PER_CODE + 1],
|
||||
);
|
||||
const truncated = rows.length > MAX_PRODUCTS_PER_CODE;
|
||||
return {
|
||||
ids: rows.slice(0, MAX_PRODUCTS_PER_CODE).map((r) => Number(r.product_id)),
|
||||
truncated,
|
||||
};
|
||||
},
|
||||
|
||||
async fetchCurrentOffers(ids) {
|
||||
if (ids.length === 0) return [];
|
||||
const [rows] = await pool.query<RowDataPacket[]>(
|
||||
"SELECT id, price, stock FROM products WHERE id IN (?)",
|
||||
[ids],
|
||||
);
|
||||
return rows.map((r) => ({
|
||||
productId: Number(r.id),
|
||||
price: Number(r.price ?? 0),
|
||||
stock: Number(r.stock ?? 0),
|
||||
}));
|
||||
},
|
||||
|
||||
async fetchHistory(ids) {
|
||||
if (ids.length === 0) return { events: [], truncated: false };
|
||||
const [rows] = await pool.query<RowDataPacket[]>(
|
||||
`SELECT product_id, DATE_FORMAT(date, '%Y-%m-%d') AS d, price, stock, prev_price, prev_stock
|
||||
FROM product_history WHERE product_id IN (?) ORDER BY date ASC LIMIT ?`,
|
||||
[ids, MAX_HISTORY_ROWS + 1],
|
||||
);
|
||||
const truncated = rows.length > MAX_HISTORY_ROWS;
|
||||
const events = rows.slice(0, MAX_HISTORY_ROWS).map((r) => ({
|
||||
productId: Number(r.product_id),
|
||||
date: String(r.d),
|
||||
price: Number(r.price ?? 0),
|
||||
stock: Number(r.stock ?? 0),
|
||||
prevPrice: Number(r.prev_price ?? 0),
|
||||
prevStock: Number(r.prev_stock ?? 0),
|
||||
}));
|
||||
return { events, truncated };
|
||||
},
|
||||
|
||||
async fetchCurrentOfferRows(codeNorms) {
|
||||
if (codeNorms.length === 0) return [];
|
||||
const [rows] = await pool.query<RowDataPacket[]>(
|
||||
`SELECT m.code_norm, p.price
|
||||
FROM sku_map m JOIN products p ON p.id = m.product_id
|
||||
WHERE m.code_norm IN (?) AND p.stock > 0 AND p.price > 0
|
||||
LIMIT ?`,
|
||||
[codeNorms, MAX_BATCH_OFFER_ROWS],
|
||||
);
|
||||
return rows.map((r) => ({ codeNorm: String(r.code_norm), price: Number(r.price) }));
|
||||
},
|
||||
|
||||
async refreshSkuMap() {
|
||||
// sku → code_norm adayları: tam, ilk boşluk sonrası, ilk tire sonrası.
|
||||
// İlk kurulumla (vmi /root/build_sku_map.sql) birebir aynı çıkarım kuralı.
|
||||
await pool.query(
|
||||
`INSERT IGNORE INTO sku_map (code_norm, product_id)
|
||||
SELECT c.code_norm, c.pid FROM (
|
||||
SELECT UPPER(REGEXP_REPLACE(sku, '[^A-Za-z0-9]', '')) AS code_norm, id AS pid
|
||||
FROM products WHERE updated_at >= CURDATE() - INTERVAL 3 DAY
|
||||
UNION ALL
|
||||
SELECT UPPER(REGEXP_REPLACE(SUBSTRING(sku, LOCATE(' ', sku) + 1), '[^A-Za-z0-9]', '')), id
|
||||
FROM products WHERE updated_at >= CURDATE() - INTERVAL 3 DAY AND sku LIKE '% %'
|
||||
UNION ALL
|
||||
SELECT UPPER(REGEXP_REPLACE(SUBSTRING(sku, LOCATE('-', sku) + 1), '[^A-Za-z0-9]', '')), id
|
||||
FROM products WHERE updated_at >= CURDATE() - INTERVAL 3 DAY AND sku LIKE '%-%'
|
||||
) c
|
||||
WHERE CHAR_LENGTH(c.code_norm) BETWEEN 5 AND 64`,
|
||||
);
|
||||
},
|
||||
|
||||
async end() {
|
||||
await pool.end();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -11,6 +11,7 @@ import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "./jobs/bull.co
|
||||
import { processEmexScrape } from "./jobs/processors/emex-scrape.processor";
|
||||
import { processExpertRewards } from "./jobs/processors/expert-rewards.processor";
|
||||
import { processLifecycleEmails } from "./jobs/processors/lifecycle-email.processor";
|
||||
import { processPartPriceRefresh } from "./jobs/processors/part-price-refresh.processor";
|
||||
import { processQueryCleanup } from "./jobs/processors/query-cleanup.processor";
|
||||
import { processSubscriptionExpiry } from "./jobs/processors/subscription-expiry.processor";
|
||||
import { processTranslation } from "./jobs/processors/translation.processor";
|
||||
@@ -186,6 +187,32 @@ expertRewardsWorker.on("failed", (job, err) => {
|
||||
|
||||
workers.push(expertRewardsWorker);
|
||||
|
||||
// Part Price Refresh Worker (daily supplier price stats for tracked part codes)
|
||||
const partPriceRefreshWorker = new Worker(
|
||||
QUEUE_NAMES.PART_PRICE_REFRESH,
|
||||
async (job) => {
|
||||
return processPartPriceRefresh(job, db);
|
||||
},
|
||||
{
|
||||
connection,
|
||||
concurrency: 1,
|
||||
...(telemetry ? { telemetry } : {}),
|
||||
},
|
||||
);
|
||||
|
||||
partPriceRefreshWorker.on("completed", (job) => {
|
||||
console.log(`[worker] part-price-refresh job ${job.id} completed`);
|
||||
});
|
||||
|
||||
partPriceRefreshWorker.on("failed", (job, err) => {
|
||||
console.error(`[worker] part-price-refresh job ${job?.id} failed: ${err.message}`);
|
||||
Sentry.captureException(err, {
|
||||
tags: { queue: QUEUE_NAMES.PART_PRICE_REFRESH, jobId: job?.id },
|
||||
});
|
||||
});
|
||||
|
||||
workers.push(partPriceRefreshWorker);
|
||||
|
||||
// Translation Worker (async LLM translation for new EMEX/PCAT terms)
|
||||
const openrouterApiKey = process.env.OPENROUTER_API_KEY;
|
||||
if (openrouterApiKey) {
|
||||
|
||||
Reference in New Issue
Block a user