Merge pull request 'dev' (#137) from dev into main
This commit was merged in pull request #137.
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");
|
||||
32
apps/api/drizzle/0019_part_price_brand.sql
Normal file
32
apps/api/drizzle/0019_part_price_brand.sql
Normal file
@@ -0,0 +1,32 @@
|
||||
-- Parça fiyat anahtarına marka eklendi: kısa sayısal kodlar (FEBI 27155 vs
|
||||
-- GROS 27155 vs İBRAŞ 27155) markalar arası çakışıyor ve FARKLI fiziksel
|
||||
-- parçaların fiyatları tek havuzda karışıyordu. Kimlik artık (kod, marka);
|
||||
-- markasız izleme yalnızca uzun/benzersiz kodlar için ('' brand_norm).
|
||||
-- Tablolar 0018'den bu yana yalnızca dev'de üç test kodu içeriyordu — veri
|
||||
-- taşımak yerine drop+create (lazy-backfill ilk görüntülemede yeniden doldurur).
|
||||
DROP TABLE IF EXISTS "part_price_daily";
|
||||
--> statement-breakpoint
|
||||
DROP TABLE IF EXISTS "part_price_tracks";
|
||||
--> statement-breakpoint
|
||||
CREATE TABLE "part_price_tracks" (
|
||||
"code_norm" varchar(64) NOT NULL,
|
||||
"brand_norm" varchar(64) DEFAULT '' 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,
|
||||
CONSTRAINT "part_price_tracks_code_norm_brand_norm_pk" PRIMARY KEY("code_norm","brand_norm")
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE TABLE "part_price_daily" (
|
||||
"code_norm" varchar(64) NOT NULL,
|
||||
"brand_norm" varchar(64) DEFAULT '' 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_brand_norm_source_date_pk" PRIMARY KEY("code_norm","brand_norm","source","date")
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE INDEX "part_price_daily_code_date_idx" ON "part_price_daily" USING btree ("code_norm","brand_norm","date");
|
||||
17
apps/api/drizzle/0020_stripe_recurring.sql
Normal file
17
apps/api/drizzle/0020_stripe_recurring.sql
Normal file
@@ -0,0 +1,17 @@
|
||||
-- Stripe recurring billing (Checkout mode:"subscription") bağlantı kolonları.
|
||||
-- users.stripe_customer_id → sonraki checkout'lar aynı Stripe Customer'da
|
||||
-- toplanır (kayıtlı kart + fatura geçmişi tek kayıtta).
|
||||
-- user_subscriptions.stripe_subscription_id → yenileme faturaları
|
||||
-- (invoice.paid/payment_failed webhook'ları) bizim abonelik satırına bu
|
||||
-- kolondan çözülür; trial ve eski tek-seferlik satırlarda NULL kalır.
|
||||
-- payments.stripe_invoice_id → invoice.paid webhook retry'larında mükerrer
|
||||
-- gelir kaydını önleyen dedupe anahtarı.
|
||||
ALTER TABLE "users" ADD COLUMN "stripe_customer_id" text;
|
||||
--> statement-breakpoint
|
||||
ALTER TABLE "user_subscriptions" ADD COLUMN "stripe_subscription_id" text;
|
||||
--> statement-breakpoint
|
||||
ALTER TABLE "payments" ADD COLUMN "stripe_invoice_id" text;
|
||||
--> statement-breakpoint
|
||||
CREATE INDEX "user_subscriptions_stripe_sub_idx" ON "user_subscriptions" USING btree ("stripe_subscription_id");
|
||||
--> statement-breakpoint
|
||||
CREATE INDEX "payments_stripe_invoice_id_idx" ON "payments" USING btree ("stripe_invoice_id");
|
||||
@@ -127,6 +127,27 @@
|
||||
"when": 1781395200000,
|
||||
"tag": "0017_oem_expert_rewards",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 18,
|
||||
"version": "7",
|
||||
"when": 1781481600000,
|
||||
"tag": "0018_part_price_history",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 19,
|
||||
"version": "7",
|
||||
"when": 1781485200000,
|
||||
"tag": "0019_part_price_brand",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 20,
|
||||
"version": "7",
|
||||
"when": 1781268490495,
|
||||
"tag": "0020_stripe_recurring",
|
||||
"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,
|
||||
@@ -37,6 +39,10 @@ export const users = pgTable(
|
||||
// subscription yet (referrer had no active/trial sub at grant time).
|
||||
// Consumed when the user next starts a trial or activates a subscription.
|
||||
referralCreditDays: integer("referral_credit_days").default(0).notNull(),
|
||||
// Stripe Customer backing this user's recurring billing. Set on the first
|
||||
// completed checkout and reused on later checkouts so saved cards and
|
||||
// invoices stay on one customer record.
|
||||
stripeCustomerId: text("stripe_customer_id"),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
@@ -153,12 +159,17 @@ export const userSubscriptions = pgTable(
|
||||
startDate: timestamp("start_date", { withTimezone: true }),
|
||||
endDate: timestamp("end_date", { withTimezone: true }),
|
||||
cancelledAt: timestamp("cancelled_at", { withTimezone: true }),
|
||||
// Stripe Subscription id for recurring (mode:"subscription") billing.
|
||||
// NULL for trials and legacy one-time purchases. Renewal invoices resolve
|
||||
// our row through this.
|
||||
stripeSubscriptionId: text("stripe_subscription_id"),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
index("user_subscriptions_user_id_idx").on(table.userId),
|
||||
index("user_subscriptions_status_idx").on(table.status),
|
||||
index("user_subscriptions_stripe_sub_idx").on(table.stripeSubscriptionId),
|
||||
],
|
||||
);
|
||||
|
||||
@@ -227,6 +238,9 @@ export const payments = pgTable(
|
||||
iyzicoPaymentId: text("iyzico_payment_id"),
|
||||
stripeSessionId: text("stripe_session_id"),
|
||||
stripePaymentIntentId: text("stripe_payment_intent_id"),
|
||||
// Stripe Invoice behind a recurring charge (renewals). Dedupe key against
|
||||
// invoice.paid webhook retries.
|
||||
stripeInvoiceId: text("stripe_invoice_id"),
|
||||
/** @deprecated EFT/Havale retired (Stripe-only). Kept for historical data. */
|
||||
bankAccountId: uuid("bank_account_id").references(() => bankAccounts.id),
|
||||
/** @deprecated EFT/Havale retired (Stripe-only). Kept for historical data. */
|
||||
@@ -239,6 +253,7 @@ export const payments = pgTable(
|
||||
index("payments_user_id_idx").on(table.userId),
|
||||
index("payments_status_idx").on(table.status),
|
||||
index("payments_stripe_session_id_idx").on(table.stripeSessionId),
|
||||
index("payments_stripe_invoice_id_idx").on(table.stripeInvoiceId),
|
||||
],
|
||||
);
|
||||
|
||||
@@ -770,3 +785,44 @@ export const oemExpertRewards = pgTable(
|
||||
index("oem_expert_rewards_user_id_idx").on(table.userId),
|
||||
],
|
||||
);
|
||||
|
||||
// ─── Part Price History (tedarikçi fiyat verisi) ──────
|
||||
// Parça bazlı piyasa fiyat geçmişi; kaynak takip MySQL'i (Tailscale). Kimlik
|
||||
// (kod, marka): kısa sayısal kodlar markalar arası çakışır (FEBI 27155 ≠
|
||||
// GROS 27155) — marka olmadan percentile anlamsızlaşır; markasız ('' brand)
|
||||
// izleme yalnızca uzun/benzersiz kodlar içindir (OEM detayın ana kodu).
|
||||
// tracks = izlenen parçalar: ilk seriyi API lazy-backfill yazar, günlük cron
|
||||
// yalnızca buradakilere bugünün satırını ekler. daily = (kod, marka, kaynak,
|
||||
// gün) başına stoktaki tekliflerin p50/p95/p99 + teklif sayısı; sıfır-teklif
|
||||
// günleri null fiyat + offerCount 0 ile saklanır. source şimdilik hep
|
||||
// 'supplier'; perakende geldiğinde aynı tabloya 'retail' 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 }).notNull(),
|
||||
brandNorm: varchar("brand_norm", { length: 64 }).default("").notNull(),
|
||||
firstSeenAt: timestamp("first_seen_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
backfilledAt: timestamp("backfilled_at", { withTimezone: true }),
|
||||
lastRefreshedAt: timestamp("last_refreshed_at", { withTimezone: true }),
|
||||
},
|
||||
(table) => [primaryKey({ columns: [table.codeNorm, table.brandNorm] })],
|
||||
);
|
||||
|
||||
export const partPriceDaily = pgTable(
|
||||
"part_price_daily",
|
||||
{
|
||||
codeNorm: varchar("code_norm", { length: 64 }).notNull(),
|
||||
brandNorm: varchar("brand_norm", { length: 64 }).default("").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.brandNorm, table.source, table.date] }),
|
||||
index("part_price_daily_code_date_idx").on(table.codeNorm, table.brandNorm, 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() {
|
||||
@@ -133,6 +140,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() {
|
||||
@@ -142,6 +168,7 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
this.catalogPrefetchQueue.close(),
|
||||
this.lifecycleEmailQueue.close(),
|
||||
this.expertRewardsQueue.close(),
|
||||
this.partPriceRefreshQueue.close(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
131
apps/api/src/jobs/processors/part-price-refresh.processor.ts
Normal file
131
apps/api/src/jobs/processors/part-price-refresh.processor.ts
Normal file
@@ -0,0 +1,131 @@
|
||||
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,
|
||||
filterOffersForBrand,
|
||||
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, marka) için bugünün p50/p95/p99'unu takip.products'tan
|
||||
* hesapla (teklifler markaya süzülür — kısa kodlar markalar arası çakışır),
|
||||
* part_price_daily'ye upsert et (teklif kalmadıysa null-fiyatlı satır —
|
||||
* "stok yok" da seridir).
|
||||
* 3. Tazelenen parçaları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, brandNorm: partPriceTracks.brandNorm })
|
||||
.from(partPriceTracks)
|
||||
.where(sql`${partPriceTracks.backfilledAt} IS NOT NULL`)
|
||||
.orderBy(asc(partPriceTracks.codeNorm), asc(partPriceTracks.brandNorm))
|
||||
.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);
|
||||
const codes = [...new Set(chunk.map((t) => t.codeNorm))];
|
||||
const rows = await source.fetchCurrentOfferRows(codes);
|
||||
const byCode = new Map<string, { brandNorm: 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({ brandNorm: r.brandNorm, price: r.price, stock: 1 });
|
||||
}
|
||||
|
||||
const values = chunk.map((t) => {
|
||||
const offers = filterOffersForBrand(byCode.get(t.codeNorm) ?? [], t.brandNorm, t.codeNorm);
|
||||
const stats = computeStats(offers);
|
||||
return {
|
||||
codeNorm: t.codeNorm,
|
||||
brandNorm: t.brandNorm,
|
||||
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.brandNorm,
|
||||
partPriceDaily.source,
|
||||
partPriceDaily.date,
|
||||
],
|
||||
set: {
|
||||
p50: sql`excluded.p50`,
|
||||
p95: sql`excluded.p95`,
|
||||
p99: sql`excluded.p99`,
|
||||
offerCount: sql`excluded.offer_count`,
|
||||
},
|
||||
});
|
||||
// Bileşik anahtarla tek tek değil çift-kolon inArray ile güncellenemez;
|
||||
// chunk küçük (≤300) — kod listesi yeterli (marka varyantları birlikte
|
||||
// tazelenir, fazladan timestamp güncellemesi zararsız).
|
||||
await db
|
||||
.update(partPriceTracks)
|
||||
.set({ lastRefreshedAt: new Date() })
|
||||
.where(inArray(partPriceTracks.codeNorm, codes));
|
||||
|
||||
// Cache düşür — bir sonraki sayfa görüntülemesi taze pg satırını okur.
|
||||
const keys = chunk.flatMap((t) => [
|
||||
`partprice:series:v2:${t.codeNorm}::${t.brandNorm}`,
|
||||
`partprice:cur:v2:${t.codeNorm}::${t.brandNorm}`,
|
||||
]);
|
||||
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 },
|
||||
},
|
||||
});
|
||||
},
|
||||
};
|
||||
30
apps/api/src/part-prices/part-prices.controller.ts
Normal file
30
apps/api/src/part-prices/part-prices.controller.ts
Normal file
@@ -0,0 +1,30 @@
|
||||
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çanın (kod + opsiyonel marka) tedarikçi fiyat serisi (p50/p95/p99,
|
||||
* günlük değişim noktaları). `GET /part-prices/series?code=27155&brand=FEBI
|
||||
* BILSTEIN` → her durumda 200; eşleşmeme/kapalı kaynak `matched: false`.
|
||||
* Marka verilmezse yalnızca uzun/benzersiz kodlar eşleşir (kısa sayısal
|
||||
* kodlar markalar arası çakışır). İlk istek lazy-backfill yapar.
|
||||
*/
|
||||
@Get("series")
|
||||
async series(@Query("code") code: string, @Query("brand") brand?: string) {
|
||||
return this.partPrices.getSeries(code ?? "", brand);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sayfada görünen (kod, marka) çiftlerinin güncel istatistikleri.
|
||||
* `POST /part-prices/current-batch { parts: [{ code, brand? }] }` →
|
||||
* `{ prices: { "CODENORM::BRANDNORM": { p50, p95, p99, offerCount } } }` —
|
||||
* eşleşmeyenler haritaya girmez, UI o satıra fiyat çizmez.
|
||||
*/
|
||||
@Post("current-batch")
|
||||
async currentBatch(@Body("parts") parts: Array<{ code?: unknown; brand?: unknown }>) {
|
||||
return this.partPrices.getCurrentBatch(Array.isArray(parts) ? parts : []);
|
||||
}
|
||||
}
|
||||
196
apps/api/src/part-prices/part-prices.logic.spec.ts
Normal file
196
apps/api/src/part-prices/part-prices.logic.spec.ts
Normal file
@@ -0,0 +1,196 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
type HistoryEvent,
|
||||
type SupplierOffer,
|
||||
allowBrandless,
|
||||
brandCompatible,
|
||||
computeStats,
|
||||
filterOffersForBrand,
|
||||
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("brandCompatible", () => {
|
||||
it("önek ve kısaltmaları tanır", () => {
|
||||
expect(brandCompatible("BOSCH", "BOSCH")).toBe(true);
|
||||
expect(brandCompatible("BCH", "BOSCH")).toBe(true); // sıralı altdizi
|
||||
expect(brandCompatible("B", "BOSCH")).toBe(true); // tek harf önek
|
||||
expect(brandCompatible("BLP", "BLUEPRINT")).toBe(true);
|
||||
expect(brandCompatible("BLUEPRNT", "BLUEPRINT")).toBe(true); // Türkçe ı düşmüş
|
||||
expect(brandCompatible("FEBI", "FEBIBILSTEIN")).toBe(true);
|
||||
expect(brandCompatible("MANN", "MANNFILTER")).toBe(true);
|
||||
expect(brandCompatible("BRA", "IBRAS")).toBe(true); // İBRAŞ → BRA
|
||||
});
|
||||
|
||||
it("alakasız markaları reddeder", () => {
|
||||
expect(brandCompatible("GROS", "FEBIBILSTEIN")).toBe(false);
|
||||
expect(brandCompatible("NIFEA", "FEBIBILSTEIN")).toBe(false);
|
||||
expect(brandCompatible("MAIS", "RENAULT")).toBe(false); // dağıtıcı — uzun-kod fallback'i halleder
|
||||
expect(brandCompatible("", "BOSCH")).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("allowBrandless", () => {
|
||||
it("uzun kodlar ve harf+rakam karışımı serbest, kısa salt-sayısal yasak", () => {
|
||||
expect(allowBrandless("8200768913")).toBe(true); // 10 hane OE
|
||||
expect(allowBrandless("46805832")).toBe(true); // 8 hane OE
|
||||
expect(allowBrandless("W7008")).toBe(true); // harf+rakam
|
||||
expect(allowBrandless("0249C6")).toBe(true); // PSA kısa OE, harf içerir
|
||||
expect(allowBrandless("27155")).toBe(false); // kısa salt-sayısal — çakışma sınıfı
|
||||
expect(allowBrandless("615014")).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("filterOffersForBrand", () => {
|
||||
// 2026-06-12 vakası: 27155 kodu FEBI/GROS/İBRAŞ/NIFEA'da FARKLI parçalar.
|
||||
const offers = [
|
||||
{ brandNorm: "FEBI", price: 160 },
|
||||
{ brandNorm: "GROS", price: 224 },
|
||||
{ brandNorm: "IBR", price: 909 },
|
||||
{ brandNorm: "NIFEA", price: 1229 },
|
||||
];
|
||||
|
||||
it("kısa kodda istenen markanın tekliflerine süzer", () => {
|
||||
const out = filterOffersForBrand(offers, "FEBIBILSTEIN", "27155");
|
||||
expect(out).toEqual([{ brandNorm: "FEBI", price: 160 }]);
|
||||
});
|
||||
|
||||
it("kısa kod + uyumsuz marka → boş (yanlış veri göstermez)", () => {
|
||||
expect(filterOffersForBrand(offers, "TRW", "27155")).toEqual([]);
|
||||
});
|
||||
|
||||
it("kısa kod + markasız istek → boş", () => {
|
||||
expect(filterOffersForBrand(offers, "", "27155")).toEqual([]);
|
||||
});
|
||||
|
||||
it("uzun kodda markasız istek tüm teklifleri kullanır (OE vakası)", () => {
|
||||
const oe = [
|
||||
{ brandNorm: "MAIS", price: 195 },
|
||||
{ brandNorm: "RENAULT", price: 247 },
|
||||
];
|
||||
expect(filterOffersForBrand(oe, "", "8200768913")).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("uzun kod + uyumsuz marka etiketi → dağıtıcı fallback'i (tümü)", () => {
|
||||
const oe = [{ brandNorm: "MAIS", price: 195 }];
|
||||
expect(filterOffersForBrand(oe, "RENAULT", "8200768913")).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
230
apps/api/src/part-prices/part-prices.logic.ts
Normal file
230
apps/api/src/part-prices/part-prices.logic.ts
Normal file
@@ -0,0 +1,230 @@
|
||||
/**
|
||||
* 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`. Markalar için de kullanılır
|
||||
* ("FEBI BILSTEIN" → FEBIBILSTEIN; Türkçe harfler düşer: "İBRAŞ" → BRA). */
|
||||
export function normPartCode(code: string): string {
|
||||
return (code ?? "").toUpperCase().replace(/[^A-Z0-9]/g, "");
|
||||
}
|
||||
|
||||
// ─── Marka eşleme ──────────────────────────────────────────────────────────
|
||||
// Kısa sayısal kodlar (FEBI 27155 / GROS 27155 / İBRAŞ 27155) markalar arası
|
||||
// çakışır — bunlar FARKLI fiziksel parçalardır; tek havuzda percentile almak
|
||||
// fiyatı anlamsızlaştırır (2026-06-12 vakası). Bu yüzden teklifler taşıdıkları
|
||||
// marka etiketiyle (takip p10 markası ya da sku öneki) istenen markaya
|
||||
// süzülür. Tedarikçiler markayı kısaltarak yazar (BCH/B→Bosch, BLP→Blue
|
||||
// Print, IBR→İbraş) — eşleşme önek VEYA sıralı-altdizi ile yapılır.
|
||||
|
||||
/** a'nın tüm karakterleri b içinde aynı sırayla geçiyor mu (BCH ⊂ BOSCH). */
|
||||
export function inOrderSubsequence(a: string, b: string): boolean {
|
||||
let i = 0;
|
||||
for (const ch of b) {
|
||||
if (ch === a[i]) i++;
|
||||
if (i === a.length) return true;
|
||||
}
|
||||
return i === a.length;
|
||||
}
|
||||
|
||||
/** Teklifin marka etiketi istenen markayla uyumlu mu (ikisi de normalize). */
|
||||
export function brandCompatible(offerBrand: string, requestedBrand: string): boolean {
|
||||
if (!offerBrand || !requestedBrand) return false;
|
||||
if (offerBrand.startsWith(requestedBrand) || requestedBrand.startsWith(offerBrand)) return true;
|
||||
// Kısaltma: en az 2 karakter, istenen markanın sıralı altdizisi (BLP→BLUEPRINT).
|
||||
return offerBrand.length >= 2 && inOrderSubsequence(offerBrand, requestedBrand);
|
||||
}
|
||||
|
||||
/** Marka süzgeci olmadan tüm teklifleri kullanmak güvenli mi? Uzun kodlar
|
||||
* (OE numaraları) pratikte benzersizdir; kısa SALT-SAYISAL kodlar markalar
|
||||
* arası çakışmanın ta kendisidir → markasız gösterilmez. */
|
||||
export function allowBrandless(codeNorm: string): boolean {
|
||||
if (codeNorm.length >= 8) return true;
|
||||
return (
|
||||
codeNorm.length >= PART_CODE_MIN_NORM_LEN && /[A-Z]/.test(codeNorm) && /[0-9]/.test(codeNorm)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Teklifleri istenen markaya süz:
|
||||
* - marka istendi → uyumlular; hiçbiri uymuyorsa ve kod markasız-güvenliyse
|
||||
* hepsi (OE/dağıtıcı etiketi vakası: RENAULT istenir, teklifler MAIS taşır);
|
||||
* kısa kodda boş (yanlış veri göstermekten iyidir),
|
||||
* - marka istenmedi → kod markasız-güvenliyse hepsi, değilse boş.
|
||||
*/
|
||||
export function filterOffersForBrand<T extends { brandNorm: string }>(
|
||||
offers: T[],
|
||||
requestedBrandNorm: string,
|
||||
codeNorm: string,
|
||||
): T[] {
|
||||
if (requestedBrandNorm) {
|
||||
const compat = offers.filter((o) => brandCompatible(o.brandNorm, requestedBrandNorm));
|
||||
if (compat.length > 0) return compat;
|
||||
return allowBrandless(codeNorm) ? offers : [];
|
||||
}
|
||||
return allowBrandless(codeNorm) ? offers : [];
|
||||
}
|
||||
|
||||
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 {}
|
||||
325
apps/api/src/part-prices/part-prices.service.ts
Normal file
325
apps/api/src/part-prices/part-prices.service.ts
Normal file
@@ -0,0 +1,325 @@
|
||||
import {
|
||||
Inject,
|
||||
Injectable,
|
||||
Logger,
|
||||
type OnModuleDestroy,
|
||||
type OnModuleInit,
|
||||
} from "@nestjs/common";
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
import { and, 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,
|
||||
filterOffersForBrand,
|
||||
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;
|
||||
brandNorm: 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 {
|
||||
/** `${codeNorm}::${brandNorm}` → güncel istatistik (eşleşmeyenler yok). */
|
||||
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_PARTS = 400;
|
||||
|
||||
export const partPriceKey = (codeNorm: string, brandNorm: string) => `${codeNorm}::${brandNorm}`;
|
||||
const seriesKey = (codeNorm: string, brandNorm: string) =>
|
||||
`partprice:series:v2:${codeNorm}::${brandNorm}`;
|
||||
const currentKey = (codeNorm: string, brandNorm: string) =>
|
||||
`partprice:cur:v2:${codeNorm}::${brandNorm}`;
|
||||
|
||||
/**
|
||||
* Parça (kod + marka) 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).
|
||||
*
|
||||
* Kimlik (kod, marka): kısa sayısal kodlar markalar arası çakışır (FEBI 27155
|
||||
* ≠ GROS 27155 ≠ İBRAŞ 27155 — farklı fiziksel parçalar). Teklifler sku_map'in
|
||||
* marka etiketiyle istenen markaya süzülür (filterOffersForBrand); markasız
|
||||
* sorgu yalnızca uzun/benzersiz kodlarda tüm teklifleri kullanır.
|
||||
*
|
||||
* - Seri: parça 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 parçalara bugünü ekler.
|
||||
* - Batch (sayfadaki satırlar): canlı MySQL'den tek sorgu + Redis cache;
|
||||
* pg'ye iz BIRAKMAZ (tracking yalnızca seri isteğiyle başlar).
|
||||
*/
|
||||
@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, brandNorm: string): PartPriceSeriesView {
|
||||
return {
|
||||
matched: false,
|
||||
codeNorm,
|
||||
brandNorm,
|
||||
currency: "TRY",
|
||||
source: "supplier",
|
||||
series: [],
|
||||
latest: null,
|
||||
truncated: false,
|
||||
};
|
||||
}
|
||||
|
||||
async getSeries(rawCode: string, rawBrand?: string): Promise<PartPriceSeriesView> {
|
||||
const codeNorm = normPartCode(rawCode);
|
||||
const brandNorm = normPartCode(rawBrand ?? "");
|
||||
if (codeNorm.length < PART_CODE_MIN_NORM_LEN) return this.miss(codeNorm, brandNorm);
|
||||
|
||||
const cached = await this.redis.getJson<PartPriceSeriesView>(seriesKey(codeNorm, brandNorm));
|
||||
if (cached) return cached;
|
||||
|
||||
try {
|
||||
// İzlenen parça → pg'den oku (hızlı yol; cron güncel tutuyor).
|
||||
const [track] = await this.db
|
||||
.select()
|
||||
.from(partPriceTracks)
|
||||
.where(
|
||||
and(eq(partPriceTracks.codeNorm, codeNorm), eq(partPriceTracks.brandNorm, brandNorm)),
|
||||
)
|
||||
.limit(1);
|
||||
|
||||
let view: PartPriceSeriesView;
|
||||
if (track?.backfilledAt) {
|
||||
const rows = await this.db
|
||||
.select()
|
||||
.from(partPriceDaily)
|
||||
.where(
|
||||
and(eq(partPriceDaily.codeNorm, codeNorm), eq(partPriceDaily.brandNorm, brandNorm)),
|
||||
)
|
||||
.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,
|
||||
brandNorm,
|
||||
currency: "TRY",
|
||||
source: "supplier",
|
||||
series,
|
||||
latest: series[series.length - 1] ?? null,
|
||||
truncated: false,
|
||||
};
|
||||
} else {
|
||||
view = await this.backfill(codeNorm, brandNorm);
|
||||
}
|
||||
|
||||
await this.redis.setJson(
|
||||
seriesKey(codeNorm, brandNorm),
|
||||
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, brandNorm);
|
||||
}
|
||||
}
|
||||
|
||||
/** İlk görüntülenme: takip history'sinden seriyi kur, pg'ye kalıcı yaz. */
|
||||
private async backfill(codeNorm: string, brandNorm: string): Promise<PartPriceSeriesView> {
|
||||
if (!this.source) return this.miss(codeNorm, brandNorm);
|
||||
|
||||
const { rows: mapped, truncated: idsTruncated } =
|
||||
await this.source.fetchMappedProducts(codeNorm);
|
||||
const selected = filterOffersForBrand(mapped, brandNorm, codeNorm);
|
||||
if (selected.length === 0) return this.miss(codeNorm, brandNorm);
|
||||
const ids = selected.map((m) => m.productId);
|
||||
|
||||
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, brandNorm);
|
||||
|
||||
// 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, brandNorm, backfilledAt: new Date() })
|
||||
.onConflictDoUpdate({
|
||||
target: [partPriceTracks.codeNorm, partPriceTracks.brandNorm],
|
||||
set: { backfilledAt: new Date() },
|
||||
});
|
||||
const rows = series.map((pt) => ({
|
||||
codeNorm,
|
||||
brandNorm,
|
||||
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,
|
||||
brandNorm,
|
||||
currency: "TRY",
|
||||
source: "supplier",
|
||||
series,
|
||||
latest: series[series.length - 1] ?? null,
|
||||
truncated: idsTruncated || histTruncated,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Sayfadaki (kod, marka) çiftleri için güncel istatistik. Canlı MySQL +
|
||||
* Redis; pg'ye dokunmaz. Kaynak kapalı/ulaşılamaz → boş harita (fail-open).
|
||||
*/
|
||||
async getCurrentBatch(
|
||||
rawParts: Array<{ code?: unknown; brand?: unknown }>,
|
||||
): Promise<PartPriceBatchView> {
|
||||
const parts = new Map<string, { codeNorm: string; brandNorm: string }>();
|
||||
for (const p of Array.isArray(rawParts) ? rawParts : []) {
|
||||
if (typeof p?.code !== "string") continue;
|
||||
const codeNorm = normPartCode(p.code);
|
||||
const brandNorm = typeof p.brand === "string" ? normPartCode(p.brand) : "";
|
||||
if (codeNorm.length < PART_CODE_MIN_NORM_LEN || codeNorm.length > 64) continue;
|
||||
const key = partPriceKey(codeNorm, brandNorm);
|
||||
if (!parts.has(key)) parts.set(key, { codeNorm, brandNorm });
|
||||
if (parts.size >= MAX_BATCH_PARTS) break;
|
||||
}
|
||||
if (parts.size === 0) return { prices: {} };
|
||||
|
||||
const prices: Record<string, PartPriceCurrent> = {};
|
||||
const pending: Array<{ key: string; codeNorm: string; brandNorm: string }> = [];
|
||||
|
||||
await Promise.all(
|
||||
[...parts.entries()].map(async ([key, p]) => {
|
||||
const hit = await this.redis.getJson<PartPriceCurrent | { miss: true }>(
|
||||
currentKey(p.codeNorm, p.brandNorm),
|
||||
);
|
||||
if (hit === null) {
|
||||
pending.push({ key, ...p });
|
||||
} else if (!("miss" in hit)) {
|
||||
prices[key] = hit;
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
if (pending.length > 0 && this.source) {
|
||||
try {
|
||||
const codes = [...new Set(pending.map((p) => p.codeNorm))];
|
||||
const rows = await this.source.fetchCurrentOfferRows(codes);
|
||||
const byCode = new Map<string, { brandNorm: 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({ brandNorm: r.brandNorm, price: r.price, stock: 1 });
|
||||
}
|
||||
await Promise.all(
|
||||
pending.map(async (p) => {
|
||||
const offers = filterOffersForBrand(
|
||||
byCode.get(p.codeNorm) ?? [],
|
||||
p.brandNorm,
|
||||
p.codeNorm,
|
||||
);
|
||||
const stats = computeStats(offers);
|
||||
if (stats.p50 !== null) {
|
||||
const current: PartPriceCurrent = {
|
||||
p50: stats.p50,
|
||||
p95: stats.p95 as number,
|
||||
p99: stats.p99 as number,
|
||||
offerCount: stats.offerCount,
|
||||
};
|
||||
prices[p.key] = current;
|
||||
await this.redis.setJson(
|
||||
currentKey(p.codeNorm, p.brandNorm),
|
||||
current,
|
||||
BATCH_CACHE_TTL,
|
||||
);
|
||||
} else {
|
||||
await this.redis.setJson(
|
||||
currentKey(p.codeNorm, p.brandNorm),
|
||||
{ miss: true },
|
||||
BATCH_CACHE_TTL,
|
||||
);
|
||||
}
|
||||
}),
|
||||
);
|
||||
} catch (err) {
|
||||
this.logger.warn(
|
||||
`[part-prices] batch failed (${pending.length} parts): ${(err as Error).message}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return { prices };
|
||||
}
|
||||
}
|
||||
153
apps/api/src/part-prices/supplier-price-source.ts
Normal file
153
apps/api/src/part-prices/supplier-price-source.ts
Normal file
@@ -0,0 +1,153 @@
|
||||
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, brand_norm)` köprü tablosu (vmi
|
||||
* üzerinde kurulu; tedarikçi SKU'larının "tam / ilk-boşluk-sonrası /
|
||||
* ilk-tire-sonrası" normalize adaylarını, markayı takip p10 kolonundan —
|
||||
* yoksa sku önekinden — alarak 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;
|
||||
|
||||
/** sku_map satırı: ürünün marka etiketiyle birlikte. */
|
||||
export interface MappedProduct {
|
||||
productId: number;
|
||||
brandNorm: string;
|
||||
}
|
||||
|
||||
export interface OfferRow {
|
||||
codeNorm: string;
|
||||
brandNorm: string;
|
||||
price: number;
|
||||
}
|
||||
|
||||
export interface SupplierPriceSource {
|
||||
fetchMappedProducts(codeNorm: string): Promise<{ rows: MappedProduct[]; truncated: boolean }>;
|
||||
fetchCurrentOffers(ids: number[]): Promise<SupplierOffer[]>;
|
||||
fetchHistory(ids: number[]): Promise<{ events: HistoryEvent[]; truncated: boolean }>;
|
||||
/** Batch: stoktaki tekliflerin (code_norm, brand_norm, price) satırları —
|
||||
* marka süzgeci ve istatistik JS'te (part-prices.logic). */
|
||||
fetchCurrentOfferRows(codeNorms: string[]): Promise<OfferRow[]>;
|
||||
/** 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 fetchMappedProducts(codeNorm) {
|
||||
const [rows] = await pool.query<RowDataPacket[]>(
|
||||
"SELECT product_id, brand_norm FROM sku_map WHERE code_norm = ? LIMIT ?",
|
||||
[codeNorm, MAX_PRODUCTS_PER_CODE + 1],
|
||||
);
|
||||
const truncated = rows.length > MAX_PRODUCTS_PER_CODE;
|
||||
return {
|
||||
rows: rows.slice(0, MAX_PRODUCTS_PER_CODE).map((r) => ({
|
||||
productId: Number(r.product_id),
|
||||
brandNorm: String(r.brand_norm ?? ""),
|
||||
})),
|
||||
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, m.brand_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),
|
||||
brandNorm: String(r.brand_norm ?? ""),
|
||||
price: Number(r.price),
|
||||
}));
|
||||
},
|
||||
|
||||
async refreshSkuMap() {
|
||||
// sku → code_norm adayları: tam, ilk boşluk sonrası, ilk tire sonrası;
|
||||
// marka = p10'un son-token-öncesi kısmı, yoksa sku öneki. İ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, brand_norm)
|
||||
SELECT c.code_norm, c.pid, c.brand_norm FROM (
|
||||
SELECT UPPER(REGEXP_REPLACE(sku, '[^A-Za-z0-9]', '')) AS code_norm, id AS pid,
|
||||
UPPER(REGEXP_REPLACE(CASE
|
||||
WHEN p10 IS NOT NULL AND p10 LIKE '% %' THEN LEFT(p10, CHAR_LENGTH(p10) - CHAR_LENGTH(SUBSTRING_INDEX(p10, ' ', -1)) - 1)
|
||||
WHEN sku LIKE '% %' THEN SUBSTRING_INDEX(sku, ' ', 1)
|
||||
WHEN sku LIKE '%-%' THEN SUBSTRING_INDEX(sku, '-', 1)
|
||||
ELSE '' END, '[^A-Za-z0-9]', '')) AS brand_norm
|
||||
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,
|
||||
UPPER(REGEXP_REPLACE(CASE
|
||||
WHEN p10 IS NOT NULL AND p10 LIKE '% %' THEN LEFT(p10, CHAR_LENGTH(p10) - CHAR_LENGTH(SUBSTRING_INDEX(p10, ' ', -1)) - 1)
|
||||
ELSE SUBSTRING_INDEX(sku, ' ', 1) END, '[^A-Za-z0-9]', ''))
|
||||
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,
|
||||
UPPER(REGEXP_REPLACE(CASE
|
||||
WHEN p10 IS NOT NULL AND p10 LIKE '% %' THEN LEFT(p10, CHAR_LENGTH(p10) - CHAR_LENGTH(SUBSTRING_INDEX(p10, ' ', -1)) - 1)
|
||||
ELSE SUBSTRING_INDEX(sku, '-', 1) END, '[^A-Za-z0-9]', ''))
|
||||
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();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,10 +1,12 @@
|
||||
import { Module } from "@nestjs/common";
|
||||
import { Module, forwardRef } from "@nestjs/common";
|
||||
import { SubscriptionsModule } from "../../subscriptions/subscriptions.module";
|
||||
import { StripeController } from "./stripe.controller";
|
||||
import { StripeService } from "./stripe.service";
|
||||
|
||||
@Module({
|
||||
imports: [SubscriptionsModule],
|
||||
// forwardRef: SubscriptionsModule imports us back (cancel/resume sync
|
||||
// auto-renewal to Stripe; the webhook here activates subscriptions).
|
||||
imports: [forwardRef(() => SubscriptionsModule)],
|
||||
controllers: [StripeController],
|
||||
providers: [StripeService],
|
||||
exports: [StripeService],
|
||||
|
||||
192
apps/api/src/payments/stripe/stripe.service.spec.ts
Normal file
192
apps/api/src/payments/stripe/stripe.service.spec.ts
Normal file
@@ -0,0 +1,192 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { StripeService } from "./stripe.service";
|
||||
|
||||
/**
|
||||
* Unit tests for the recurring-billing webhook handlers (invoice.paid /
|
||||
* invoice.payment_failed). The service is constructed WITHOUT a Stripe key
|
||||
* (client = null), which the handlers tolerate: the only Stripe API call on
|
||||
* these paths (payment-intent backfill) is fail-open.
|
||||
*/
|
||||
|
||||
/** Sequenced select mock: call N resolves results[N] (last repeats). */
|
||||
function sequencedSelect(results: unknown[][]) {
|
||||
let i = 0;
|
||||
return vi.fn().mockImplementation(() => {
|
||||
const rows = results[Math.min(i, results.length - 1)];
|
||||
i++;
|
||||
return {
|
||||
from: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnThis(),
|
||||
innerJoin: vi.fn().mockReturnThis(),
|
||||
orderBy: vi.fn().mockReturnThis(),
|
||||
limit: vi.fn().mockReturnValue(rows),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
function createMocks(selectResults: unknown[][]) {
|
||||
const updateChains: Array<{ set: ReturnType<typeof vi.fn> }> = [];
|
||||
const db = {
|
||||
select: sequencedSelect(selectResults),
|
||||
update: vi.fn().mockImplementation(() => {
|
||||
const chain = {
|
||||
set: vi.fn(),
|
||||
where: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
chain.set.mockReturnValue(chain);
|
||||
updateChains.push(chain);
|
||||
return chain;
|
||||
}),
|
||||
insert: vi.fn().mockReturnValue({ values: vi.fn().mockResolvedValue(undefined) }),
|
||||
delete: vi.fn().mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) }),
|
||||
};
|
||||
const posthog = {
|
||||
captureForUser: vi.fn(),
|
||||
capture: vi.fn(),
|
||||
flush: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
const novu = {
|
||||
paymentSuccess: vi.fn().mockResolvedValue(undefined),
|
||||
paymentFailed: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
const config = { get: vi.fn().mockReturnValue(undefined) }; // no Stripe key → client null
|
||||
const service = new StripeService(
|
||||
db as any,
|
||||
config as any,
|
||||
{} as any, // SubscriptionsService — unused on invoice paths
|
||||
posthog as any,
|
||||
novu as any,
|
||||
);
|
||||
return { service: service as any, db, updateChains, posthog, novu };
|
||||
}
|
||||
|
||||
const PERIOD_END_SEC = 1784278800; // 2026-07-17T01:00:00Z
|
||||
const RETRY_AT_SEC = 1781700000;
|
||||
|
||||
function invoiceFixture(over: Record<string, unknown> = {}) {
|
||||
return {
|
||||
id: "in_1",
|
||||
billing_reason: "subscription_cycle",
|
||||
amount_paid: 5000,
|
||||
amount_due: 5000,
|
||||
next_payment_attempt: RETRY_AT_SEC,
|
||||
parent: {
|
||||
type: "subscription_details",
|
||||
subscription_details: {
|
||||
subscription: "sub_stripe1",
|
||||
metadata: { subscription_id: "our-sub-1", user_id: "user-1" },
|
||||
},
|
||||
},
|
||||
lines: { data: [{ period: { end: PERIOD_END_SEC } }] },
|
||||
...over,
|
||||
};
|
||||
}
|
||||
|
||||
const activeSub = {
|
||||
id: "our-sub-1",
|
||||
userId: "user-1",
|
||||
planId: "plan-1",
|
||||
status: "active",
|
||||
billingPeriod: "monthly",
|
||||
stripeSubscriptionId: "sub_stripe1",
|
||||
cancelledAt: null,
|
||||
};
|
||||
const fullPlan = { id: "plan-1", name: "Full Paket", brandCount: 0, priceMonthly: 5000 };
|
||||
const user = { id: "user-1", email: "u@example.com", name: "U" };
|
||||
|
||||
describe("StripeService recurring webhooks", () => {
|
||||
describe("handleInvoicePaid (renewal)", () => {
|
||||
it("extends the paid-through date, records the payment once, counts revenue, mails the receipt", async () => {
|
||||
// selects: resolve sub → dedupe (none) → plan → user (for the mail)
|
||||
const { service, db, updateChains, posthog, novu } = createMocks([
|
||||
[activeSub],
|
||||
[],
|
||||
[fullPlan],
|
||||
[user],
|
||||
]);
|
||||
|
||||
await service.handleInvoicePaid(invoiceFixture());
|
||||
|
||||
// end_date extended to Stripe's billing-line period end
|
||||
expect(db.update).toHaveBeenCalledTimes(1);
|
||||
const setArg = updateChains[0].set.mock.calls[0][0] as { endDate?: Date };
|
||||
expect(setArg.endDate?.getTime()).toBe(PERIOD_END_SEC * 1000);
|
||||
|
||||
// exactly one completed payment row, keyed to the invoice
|
||||
expect(db.insert).toHaveBeenCalledTimes(1);
|
||||
const valuesArg = (db.insert.mock.results[0].value as { values: ReturnType<typeof vi.fn> })
|
||||
.values.mock.calls[0][0];
|
||||
expect(valuesArg).toEqual(
|
||||
expect.objectContaining({
|
||||
status: "completed",
|
||||
amount: 5000,
|
||||
stripeInvoiceId: "in_1",
|
||||
userId: "user-1",
|
||||
}),
|
||||
);
|
||||
|
||||
// renewal revenue + receipt mail (success mail YES, reminder mails never)
|
||||
expect(posthog.captureForUser).toHaveBeenCalledWith(
|
||||
"user-1",
|
||||
"subscription_renewed",
|
||||
expect.objectContaining({ $revenue: 50, amount_kurus: 5000 }),
|
||||
);
|
||||
expect(novu.paymentSuccess).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("ignores a webhook retry for an already-recorded invoice", async () => {
|
||||
// selects: resolve sub → dedupe finds the prior payment row
|
||||
const { service, db, posthog, novu } = createMocks([[activeSub], [{ id: "pay-1" }]]);
|
||||
|
||||
await service.handleInvoicePaid(invoiceFixture());
|
||||
|
||||
expect(db.insert).not.toHaveBeenCalled();
|
||||
expect(db.update).not.toHaveBeenCalled();
|
||||
expect(posthog.captureForUser).not.toHaveBeenCalled();
|
||||
expect(novu.paymentSuccess).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("subscription_create invoices only enrich the checkout's payment row (no double revenue)", async () => {
|
||||
const { service, db, updateChains, posthog, novu } = createMocks([[activeSub]]);
|
||||
|
||||
await service.handleInvoicePaid(invoiceFixture({ billing_reason: "subscription_create" }));
|
||||
|
||||
// single update: invoice id (+ intent when available) onto the payment row
|
||||
expect(db.update).toHaveBeenCalledTimes(1);
|
||||
expect(updateChains[0].set).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ stripeInvoiceId: "in_1" }),
|
||||
);
|
||||
expect(db.insert).not.toHaveBeenCalled();
|
||||
expect(posthog.captureForUser).not.toHaveBeenCalled();
|
||||
expect(novu.paymentSuccess).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("handleInvoiceFailed", () => {
|
||||
it("mails dunning with Stripe's retry date and leaves the paid-through date alone", async () => {
|
||||
// selects: resolve sub → user
|
||||
const { service, db, posthog, novu } = createMocks([[activeSub], [user]]);
|
||||
|
||||
await service.handleInvoiceFailed(invoiceFixture());
|
||||
|
||||
expect(db.update).not.toHaveBeenCalled(); // access keeps running to end_date
|
||||
expect(novu.paymentFailed).toHaveBeenCalledTimes(1);
|
||||
const opts = novu.paymentFailed.mock.calls[0][1] as { retryDate?: Date };
|
||||
expect(opts.retryDate?.getTime()).toBe(RETRY_AT_SEC * 1000);
|
||||
expect(posthog.captureForUser).toHaveBeenCalledWith(
|
||||
"user-1",
|
||||
"payment_failed",
|
||||
expect.objectContaining({ reason: "renewal_charge_failed" }),
|
||||
);
|
||||
});
|
||||
|
||||
it("ignores first-charge failures (checkout flow owns those)", async () => {
|
||||
const { service, posthog, novu } = createMocks([[activeSub]]);
|
||||
|
||||
await service.handleInvoiceFailed(invoiceFixture({ billing_reason: "subscription_create" }));
|
||||
|
||||
expect(novu.paymentFailed).not.toHaveBeenCalled();
|
||||
expect(posthog.captureForUser).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -5,9 +5,10 @@ import {
|
||||
Logger,
|
||||
NotFoundException,
|
||||
ServiceUnavailableException,
|
||||
forwardRef,
|
||||
} from "@nestjs/common";
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { and, eq, isNull } from "drizzle-orm";
|
||||
import Stripe from "stripe";
|
||||
import { DATABASE, type Database } from "../../database/database.provider";
|
||||
|
||||
@@ -17,7 +18,17 @@ import { DATABASE, type Database } from "../../database/database.provider";
|
||||
type StripeNs = import("stripe/cjs/stripe.core").Stripe;
|
||||
type StripeEvent = import("stripe/cjs/stripe.core").Stripe.Event;
|
||||
type CheckoutSession = import("stripe/cjs/stripe.core").Stripe.Checkout.Session;
|
||||
import { payments, plans, userSubscriptions, users } from "../../database/schema/core";
|
||||
type SessionCreateParams = import("stripe/cjs/stripe.core").Stripe.Checkout.SessionCreateParams;
|
||||
type StripeInvoice = import("stripe/cjs/stripe.core").Stripe.Invoice;
|
||||
type StripeSubscriptionObj = import("stripe/cjs/stripe.core").Stripe.Subscription;
|
||||
import {
|
||||
brands,
|
||||
payments,
|
||||
plans,
|
||||
userBrands,
|
||||
userSubscriptions,
|
||||
users,
|
||||
} from "../../database/schema/core";
|
||||
import { NovuService } from "../../notifications/novu.service";
|
||||
import { PostHogService } from "../../posthog/posthog.service";
|
||||
import { SubscriptionsService } from "../../subscriptions/subscriptions.service";
|
||||
@@ -40,6 +51,9 @@ export class StripeService {
|
||||
constructor(
|
||||
@Inject(DATABASE) private db: Database,
|
||||
private configService: ConfigService,
|
||||
// forwardRef: SubscriptionsService also injects StripeService (cancel/resume
|
||||
// must sync auto-renewal to Stripe), so the two providers are circular.
|
||||
@Inject(forwardRef(() => SubscriptionsService))
|
||||
private subscriptionsService: SubscriptionsService,
|
||||
private posthog: PostHogService,
|
||||
private novu: NovuService,
|
||||
@@ -141,24 +155,48 @@ export class StripeService {
|
||||
})
|
||||
.returning();
|
||||
|
||||
const session = await this.stripe.checkout.sessions.create({
|
||||
mode: "payment",
|
||||
// Reuse the Stripe customer from earlier purchases so the saved card and
|
||||
// invoice history stay on one record.
|
||||
const [buyer] = await this.db
|
||||
.select({ stripeCustomerId: users.stripeCustomerId })
|
||||
.from(users)
|
||||
.where(eq(users.id, userId))
|
||||
.limit(1);
|
||||
|
||||
const buildParams = (customer: string | null): SessionCreateParams => ({
|
||||
// Recurring billing: Stripe owns the renewal schedule and auto-charges the
|
||||
// saved card each period. No pre-charge reminder mails (product decision) —
|
||||
// only a receipt on success and dunning on failure, both webhook-driven.
|
||||
// Our row's end_date is extended on every paid invoice (handleInvoicePaid);
|
||||
// access keeps gating on end_date, so a failed renewal naturally lapses.
|
||||
mode: "subscription",
|
||||
// Render Stripe's hosted page in Turkish. The audience is Turkish B2B; a
|
||||
// foreign-language checkout is a known abandonment driver (~60% of sessions
|
||||
// reached the page but never started a payment intent).
|
||||
locale: "tr",
|
||||
payment_method_types: ["card"],
|
||||
customer_email: userEmail,
|
||||
...(customer ? { customer } : { customer_email: userEmail }),
|
||||
line_items: [
|
||||
{
|
||||
price_data: {
|
||||
currency: "try",
|
||||
product_data: { name: productName },
|
||||
unit_amount: amount, // already in kuruş (smallest unit)
|
||||
recurring: { interval: billingPeriod === "yearly" ? "year" : "month" },
|
||||
},
|
||||
quantity: 1,
|
||||
},
|
||||
],
|
||||
subscription_data: {
|
||||
// Mirrored onto every invoice (parent.subscription_details.metadata), so
|
||||
// renewal webhooks can resolve our rows even before/without the
|
||||
// stripe_subscription_id column link.
|
||||
metadata: {
|
||||
subscription_id: subscription.id,
|
||||
user_id: userId,
|
||||
plan_key: planKey,
|
||||
},
|
||||
},
|
||||
success_url: `${this.successUrl}&session_id={CHECKOUT_SESSION_ID}`,
|
||||
cancel_url: this.cancelUrl,
|
||||
client_reference_id: payment.id,
|
||||
@@ -171,6 +209,28 @@ export class StripeService {
|
||||
},
|
||||
});
|
||||
|
||||
let session: CheckoutSession;
|
||||
try {
|
||||
session = await this.stripe.checkout.sessions.create(
|
||||
buildParams(buyer?.stripeCustomerId ?? null),
|
||||
);
|
||||
} catch (err) {
|
||||
// A stored customer deleted on Stripe's side must not brick the user's
|
||||
// checkout forever: clear the stale id and retry with a fresh customer.
|
||||
if (buyer?.stripeCustomerId && String(err).includes("No such customer")) {
|
||||
this.logger.warn(
|
||||
`Stored Stripe customer ${buyer.stripeCustomerId} is gone — clearing and retrying (user=${userId})`,
|
||||
);
|
||||
await this.db
|
||||
.update(users)
|
||||
.set({ stripeCustomerId: null, updatedAt: new Date() })
|
||||
.where(eq(users.id, userId));
|
||||
session = await this.stripe.checkout.sessions.create(buildParams(null));
|
||||
} else {
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
await this.db
|
||||
.update(payments)
|
||||
.set({ stripeSessionId: session.id, updatedAt: new Date() })
|
||||
@@ -222,6 +282,18 @@ export class StripeService {
|
||||
await this.handleCheckoutFailed(session, event.type);
|
||||
break;
|
||||
}
|
||||
case "invoice.paid": {
|
||||
await this.handleInvoicePaid(event.data.object as StripeInvoice);
|
||||
break;
|
||||
}
|
||||
case "invoice.payment_failed": {
|
||||
await this.handleInvoiceFailed(event.data.object as StripeInvoice);
|
||||
break;
|
||||
}
|
||||
case "customer.subscription.deleted": {
|
||||
await this.handleSubscriptionDeleted(event.data.object as StripeSubscriptionObj);
|
||||
break;
|
||||
}
|
||||
default:
|
||||
this.logger.debug(`Unhandled Stripe event type: ${event.type}`);
|
||||
}
|
||||
@@ -258,6 +330,31 @@ export class StripeService {
|
||||
return;
|
||||
}
|
||||
|
||||
// Persist the Stripe linkage this checkout created BEFORE activating, so a
|
||||
// failed activation (webhook retry) never loses it: the recurring
|
||||
// subscription id (renewal invoices resolve through it) and the customer
|
||||
// id (the next checkout reuses the same Stripe customer).
|
||||
const stripeSubId =
|
||||
typeof session.subscription === "string"
|
||||
? session.subscription
|
||||
: (session.subscription?.id ?? null);
|
||||
if (stripeSubId) {
|
||||
await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ stripeSubscriptionId: stripeSubId, updatedAt: new Date() })
|
||||
.where(eq(userSubscriptions.id, payment.subscriptionId));
|
||||
}
|
||||
const stripeCustomerId =
|
||||
typeof session.customer === "string" ? session.customer : (session.customer?.id ?? null);
|
||||
if (stripeCustomerId) {
|
||||
await this.db
|
||||
.update(users)
|
||||
.set({ stripeCustomerId, updatedAt: new Date() })
|
||||
.where(eq(users.id, payment.userId));
|
||||
}
|
||||
|
||||
// In subscription mode the charge lives on the first invoice, so the
|
||||
// session itself carries no payment_intent — handleInvoicePaid backfills it.
|
||||
const paymentIntentId =
|
||||
typeof session.payment_intent === "string" ? session.payment_intent : null;
|
||||
|
||||
@@ -382,6 +479,308 @@ export class StripeService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve our user_subscriptions row for a subscription invoice: primarily
|
||||
* via the stripe_subscription_id column, falling back to the subscription
|
||||
* metadata we stamp at checkout (covers invoice events that arrive before
|
||||
* checkout.session.completed has linked the column).
|
||||
*/
|
||||
private async resolveOurSubscription(stripeSubId: string | null, ourSubId: string | null) {
|
||||
if (stripeSubId) {
|
||||
const [byColumn] = await this.db
|
||||
.select()
|
||||
.from(userSubscriptions)
|
||||
.where(eq(userSubscriptions.stripeSubscriptionId, stripeSubId))
|
||||
.limit(1);
|
||||
if (byColumn) return byColumn;
|
||||
}
|
||||
if (ourSubId) {
|
||||
const [byMetadata] = await this.db
|
||||
.select()
|
||||
.from(userSubscriptions)
|
||||
.where(eq(userSubscriptions.id, ourSubId))
|
||||
.limit(1);
|
||||
if (byMetadata) return byMetadata;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch the PaymentIntent id behind an invoice's charge (webhook payloads
|
||||
* don't embed it). Receipts and panel refunds key on the intent. Best-effort.
|
||||
*/
|
||||
private async fetchInvoicePaymentIntentId(invoiceId: string): Promise<string | null> {
|
||||
if (!this.stripe) return null;
|
||||
try {
|
||||
const inv = await this.stripe.invoices.retrieve(invoiceId, {
|
||||
expand: ["payments.data.payment.payment_intent"],
|
||||
});
|
||||
const paid = inv.payments?.data?.find((p) => p.status === "paid") ?? inv.payments?.data?.[0];
|
||||
const pi = paid?.payment?.payment_intent;
|
||||
return typeof pi === "string" ? pi : (pi?.id ?? null);
|
||||
} catch (err) {
|
||||
this.logger.warn(`invoice PI fetch failed (${invoiceId}): ${String(err)}`);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* invoice.paid — the heartbeat of recurring billing.
|
||||
*
|
||||
* billing_reason=subscription_create (first charge): activation, revenue and
|
||||
* the receipt mail are owned by checkout.session.completed; here we only
|
||||
* enrich the original payment row (intent + invoice id) and backfill the
|
||||
* subscription link in case events arrived out of order.
|
||||
*
|
||||
* Any other billing_reason (subscription_cycle renewals, updates): extend the
|
||||
* paid-through date, record a completed payment (deduped on invoice id),
|
||||
* count renewal revenue, and mail the receipt.
|
||||
*/
|
||||
private async handleInvoicePaid(invoice: StripeInvoice) {
|
||||
const details = invoice.parent?.subscription_details ?? null;
|
||||
if (!details) return; // not a subscription invoice
|
||||
|
||||
const subRef = details.subscription;
|
||||
const stripeSubId = typeof subRef === "string" ? subRef : (subRef?.id ?? null);
|
||||
const metaSubId = details.metadata?.subscription_id ?? null;
|
||||
|
||||
const sub = await this.resolveOurSubscription(stripeSubId, metaSubId);
|
||||
if (!sub) {
|
||||
this.logger.warn(`invoice.paid ${invoice.id}: no matching subscription (${stripeSubId})`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Backfill the column link (events can beat checkout.session.completed).
|
||||
if (!sub.stripeSubscriptionId && stripeSubId) {
|
||||
await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ stripeSubscriptionId: stripeSubId, updatedAt: new Date() })
|
||||
.where(eq(userSubscriptions.id, sub.id));
|
||||
}
|
||||
|
||||
if (invoice.billing_reason === "subscription_create") {
|
||||
const paymentIntentId = await this.fetchInvoicePaymentIntentId(invoice.id);
|
||||
await this.db
|
||||
.update(payments)
|
||||
.set({
|
||||
...(paymentIntentId ? { stripePaymentIntentId: paymentIntentId } : {}),
|
||||
stripeInvoiceId: invoice.id,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(and(eq(payments.subscriptionId, sub.id), isNull(payments.stripeInvoiceId)));
|
||||
return;
|
||||
}
|
||||
|
||||
// ── Renewal ──
|
||||
const [already] = await this.db
|
||||
.select({ id: payments.id })
|
||||
.from(payments)
|
||||
.where(eq(payments.stripeInvoiceId, invoice.id))
|
||||
.limit(1);
|
||||
if (already) {
|
||||
this.logger.debug(`invoice.paid ${invoice.id} already recorded — skipping`);
|
||||
return;
|
||||
}
|
||||
|
||||
const now = new Date();
|
||||
// Paid-through date comes from the invoice line period (Stripe's truth for
|
||||
// the billing window); fall back to now+period if the payload lacks lines.
|
||||
let newEndDate: Date;
|
||||
const periodEndSec = invoice.lines?.data?.[0]?.period?.end;
|
||||
if (periodEndSec) {
|
||||
newEndDate = new Date(periodEndSec * 1000);
|
||||
} else {
|
||||
newEndDate = new Date(now);
|
||||
if (sub.billingPeriod === "yearly") newEndDate.setFullYear(newEndDate.getFullYear() + 1);
|
||||
else newEndDate.setMonth(newEndDate.getMonth() + 1);
|
||||
}
|
||||
|
||||
const [plan] = await this.db.select().from(plans).where(eq(plans.id, sub.planId)).limit(1);
|
||||
const recovered = sub.status !== "active" && sub.status !== "cancelled";
|
||||
|
||||
if (!recovered) {
|
||||
// Normal renewal — extend. A 'cancelled' row that still got billed keeps
|
||||
// its status; access is governed by end_date either way.
|
||||
await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ endDate: newEndDate, updatedAt: now })
|
||||
.where(eq(userSubscriptions.id, sub.id));
|
||||
} else {
|
||||
// Late dunning recovery: the nightly cron already expired the row and
|
||||
// purged its brand grants. Restore access for the freshly paid period.
|
||||
await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ status: "active", endDate: newEndDate, updatedAt: now })
|
||||
.where(eq(userSubscriptions.id, sub.id));
|
||||
if (plan?.brandCount === 0) {
|
||||
await this.db.delete(userBrands).where(eq(userBrands.subscriptionId, sub.id));
|
||||
const allBrands = await this.db.select().from(brands).where(eq(brands.isActive, true));
|
||||
if (allBrands.length > 0) {
|
||||
await this.db
|
||||
.insert(userBrands)
|
||||
.values(
|
||||
allBrands.map((b) => ({ userId: sub.userId, subscriptionId: sub.id, brandId: b.id })),
|
||||
);
|
||||
}
|
||||
} else {
|
||||
this.logger.error(
|
||||
`invoice.paid ${invoice.id}: sub ${sub.id} recovered from '${sub.status}' but its ` +
|
||||
"brand-plan grants were purged at expiry — re-grant brands manually",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const paymentIntentId = await this.fetchInvoicePaymentIntentId(invoice.id);
|
||||
const amountKurus = invoice.amount_paid ?? 0;
|
||||
await this.db.insert(payments).values({
|
||||
userId: sub.userId,
|
||||
subscriptionId: sub.id,
|
||||
amount: amountKurus,
|
||||
currency: "TRY",
|
||||
method: "stripe",
|
||||
status: "completed",
|
||||
stripeInvoiceId: invoice.id,
|
||||
...(paymentIntentId ? { stripePaymentIntentId: paymentIntentId } : {}),
|
||||
});
|
||||
|
||||
// Renewal revenue is realized revenue: counted here as subscription_renewed,
|
||||
// while first-charge revenue stays on subscription_activated — together they
|
||||
// are the only $revenue sources (funnel steps intentionally carry none).
|
||||
this.posthog.captureForUser(sub.userId, "subscription_renewed", {
|
||||
$revenue: amountKurus / 100,
|
||||
currency: "TRY",
|
||||
amount_kurus: amountKurus,
|
||||
mrr: sub.billingPeriod === "yearly" ? amountKurus / 12 / 100 : amountKurus / 100,
|
||||
plan: plan?.name ?? null,
|
||||
plan_id: sub.planId,
|
||||
billing_period: sub.billingPeriod,
|
||||
stripe_invoice_id: invoice.id,
|
||||
recovered,
|
||||
});
|
||||
|
||||
// Receipt mail. Product rule: mail on success, mail on failure, never a
|
||||
// pre-charge reminder.
|
||||
try {
|
||||
const [user] = await this.db
|
||||
.select({ id: users.id, email: users.email, name: users.name })
|
||||
.from(users)
|
||||
.where(eq(users.id, sub.userId))
|
||||
.limit(1);
|
||||
if (user) {
|
||||
await this.novu.paymentSuccess(user, {
|
||||
amountKurus,
|
||||
plan: plan?.name ?? null,
|
||||
nextBillingDate: newEndDate,
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.error(`renewal receipt mail failed (user=${sub.userId}): ${String(err)}`);
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`Renewal recorded: sub ${sub.id} paid ${amountKurus} kuruş through ` +
|
||||
`${newEndDate.toISOString()} (invoice ${invoice.id})`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* invoice.payment_failed — a renewal charge bounced. Mail the dunning notice
|
||||
* with Stripe's next retry date. Access is NOT cut here: end_date still
|
||||
* governs, Stripe keeps smart-retrying, and the nightly expiry cron closes
|
||||
* access only when the paid-through date lapses.
|
||||
*/
|
||||
private async handleInvoiceFailed(invoice: StripeInvoice) {
|
||||
const details = invoice.parent?.subscription_details ?? null;
|
||||
if (!details) return;
|
||||
// First-charge failures surface inline in the checkout flow; the
|
||||
// session-expiry handler owns dunning for abandons.
|
||||
if (invoice.billing_reason === "subscription_create") return;
|
||||
|
||||
const subRef = details.subscription;
|
||||
const stripeSubId = typeof subRef === "string" ? subRef : (subRef?.id ?? null);
|
||||
const metaSubId = details.metadata?.subscription_id ?? null;
|
||||
|
||||
const sub = await this.resolveOurSubscription(stripeSubId, metaSubId);
|
||||
if (!sub) {
|
||||
this.logger.warn(
|
||||
`invoice.payment_failed ${invoice.id}: no matching subscription (${stripeSubId})`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const retryDate = invoice.next_payment_attempt
|
||||
? new Date(invoice.next_payment_attempt * 1000)
|
||||
: null;
|
||||
|
||||
this.posthog.captureForUser(sub.userId, "payment_failed", {
|
||||
method: "stripe",
|
||||
subscription_id: sub.id,
|
||||
reason: "renewal_charge_failed",
|
||||
amount: invoice.amount_due ?? null,
|
||||
stripe_invoice_id: invoice.id,
|
||||
next_retry_at: retryDate ? retryDate.toISOString() : null,
|
||||
});
|
||||
|
||||
try {
|
||||
const [user] = await this.db
|
||||
.select({ id: users.id, email: users.email, name: users.name })
|
||||
.from(users)
|
||||
.where(eq(users.id, sub.userId))
|
||||
.limit(1);
|
||||
if (user) {
|
||||
await this.novu.paymentFailed(user, {
|
||||
amountKurus: invoice.amount_due ?? undefined,
|
||||
retryDate,
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.error(`renewal dunning mail failed (user=${sub.userId}): ${String(err)}`);
|
||||
}
|
||||
|
||||
this.logger.warn(
|
||||
`Renewal charge failed: sub ${sub.id} (invoice ${invoice.id}), ` +
|
||||
`next retry ${retryDate ? retryDate.toISOString() : "none (final)"}`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* customer.subscription.deleted — Stripe will send no further invoices
|
||||
* (cancel-at-period-end executed, or dunning gave up). Access keeps running
|
||||
* to end_date; the nightly cron flips the row to expired after that.
|
||||
*/
|
||||
private async handleSubscriptionDeleted(subscription: StripeSubscriptionObj) {
|
||||
const [sub] = await this.db
|
||||
.select()
|
||||
.from(userSubscriptions)
|
||||
.where(eq(userSubscriptions.stripeSubscriptionId, subscription.id))
|
||||
.limit(1);
|
||||
if (!sub) return;
|
||||
|
||||
await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ cancelledAt: sub.cancelledAt ?? new Date(), updatedAt: new Date() })
|
||||
.where(eq(userSubscriptions.id, sub.id));
|
||||
|
||||
this.logger.log(
|
||||
`Stripe subscription ${subscription.id} deleted — our sub ${sub.id} ` +
|
||||
`(status=${sub.status}) will not renew; access runs out at its end_date`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Toggle auto-renewal on the Stripe subscription backing a recurring plan.
|
||||
* cancel() sets it (no further charges; access runs to end_date), resume()
|
||||
* clears it. Callers skip legacy one-time subs (no Stripe subscription id).
|
||||
*/
|
||||
async setCancelAtPeriodEnd(stripeSubscriptionId: string, cancel: boolean): Promise<void> {
|
||||
if (!this.stripe) {
|
||||
throw new ServiceUnavailableException("Stripe ödeme şu an kullanılamıyor");
|
||||
}
|
||||
await this.stripe.subscriptions.update(stripeSubscriptionId, {
|
||||
cancel_at_period_end: cancel,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Refund a completed Stripe payment. Called from the InternalAdmin module
|
||||
* via Süper Panel. `amount` is in the smallest currency unit (kuruş for TRY)
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
import { Module } from "@nestjs/common";
|
||||
import { Module, forwardRef } from "@nestjs/common";
|
||||
import { BrandsModule } from "../brands/brands.module";
|
||||
import { StripeModule } from "../payments/stripe/stripe.module";
|
||||
import { PlansModule } from "../plans/plans.module";
|
||||
import { SubscriptionsController } from "./subscriptions.controller";
|
||||
import { SubscriptionsService } from "./subscriptions.service";
|
||||
|
||||
@Module({
|
||||
imports: [BrandsModule, PlansModule],
|
||||
// forwardRef: StripeModule imports us back (webhook activation needs
|
||||
// SubscriptionsService; cancel/resume here need StripeService).
|
||||
imports: [BrandsModule, PlansModule, forwardRef(() => StripeModule)],
|
||||
controllers: [SubscriptionsController],
|
||||
providers: [SubscriptionsService],
|
||||
exports: [SubscriptionsService],
|
||||
|
||||
@@ -55,14 +55,23 @@ function createMockDb(overrides: Record<string, unknown> = {}) {
|
||||
/**
|
||||
* Creates the service with a given mock db injected via reflection.
|
||||
*/
|
||||
function createService(db: unknown): SubscriptionsService {
|
||||
function createService(
|
||||
db: unknown,
|
||||
stripe?: { setCancelAtPeriodEnd: ReturnType<typeof vi.fn> },
|
||||
): SubscriptionsService {
|
||||
const posthog = {
|
||||
captureForUser: vi.fn(),
|
||||
capture: vi.fn(),
|
||||
flush: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
const metaCapi = { sendPurchase: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = new SubscriptionsService(db as any, posthog as any, metaCapi as any);
|
||||
const stripeService = stripe ?? { setCancelAtPeriodEnd: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = new SubscriptionsService(
|
||||
db as any,
|
||||
posthog as any,
|
||||
metaCapi as any,
|
||||
stripeService as any,
|
||||
);
|
||||
return service;
|
||||
}
|
||||
|
||||
@@ -187,6 +196,55 @@ describe("SubscriptionsService", () => {
|
||||
status: "pending",
|
||||
});
|
||||
});
|
||||
|
||||
it("does NOT expire a live trial at checkout start (only stale pending)", async () => {
|
||||
// Regression: create() runs on the "Öde" click, before any payment.
|
||||
// It used to expire the live trial right there, so abandoning the
|
||||
// Stripe page locked the user out without them ever paying.
|
||||
let callCount = 0;
|
||||
const updateChains: Array<{ set: ReturnType<typeof vi.fn> }> = [];
|
||||
const db = {
|
||||
select: vi.fn().mockImplementation(() => {
|
||||
callCount++;
|
||||
return {
|
||||
from: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnThis(),
|
||||
limit: vi.fn().mockImplementation(() => {
|
||||
if (callCount === 1) return []; // no existing ACTIVE subscription
|
||||
if (callCount === 2) return [{ id: "plan-full", brandCount: 0 }];
|
||||
return [];
|
||||
}),
|
||||
};
|
||||
}),
|
||||
update: vi.fn().mockImplementation(() => {
|
||||
const chain = {
|
||||
set: vi.fn(),
|
||||
where: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
chain.set.mockReturnValue(chain);
|
||||
updateChains.push(chain);
|
||||
return chain;
|
||||
}),
|
||||
insert: vi.fn().mockReturnValue({
|
||||
values: vi.fn().mockReturnThis(),
|
||||
returning: vi.fn().mockReturnValue([{ id: "sub-1", status: "pending" }]),
|
||||
}),
|
||||
};
|
||||
const service = createService(db);
|
||||
|
||||
await service.create("user-1", {
|
||||
planId: "plan-full",
|
||||
brandIds: [],
|
||||
billingPeriod: "monthly",
|
||||
});
|
||||
|
||||
// Exactly ONE expiry update may run at checkout start: the stale-pending
|
||||
// cleanup. A second one (the old trial kill) is the lockout bug.
|
||||
expect(db.update).toHaveBeenCalledTimes(1);
|
||||
expect(updateChains[0].set).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ status: "expired" }),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("cancel", () => {
|
||||
@@ -289,6 +347,67 @@ describe("SubscriptionsService", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("Stripe auto-renewal sync", () => {
|
||||
function dbWithSub(sub: Record<string, unknown>, updated: Record<string, unknown>) {
|
||||
return {
|
||||
select: vi.fn().mockReturnValue({
|
||||
from: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnThis(),
|
||||
limit: vi.fn().mockReturnValue([sub]),
|
||||
}),
|
||||
update: vi.fn().mockReturnValue({
|
||||
set: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnThis(),
|
||||
returning: vi.fn().mockReturnValue([updated]),
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
it("cancel() stops Stripe auto-renewal for a recurring subscription", async () => {
|
||||
const sub = { id: "sub-1", stripeSubscriptionId: "sub_stripe1", status: "active" };
|
||||
const db = dbWithSub(sub, { ...sub, status: "cancelled" });
|
||||
const stripe = { setCancelAtPeriodEnd: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = createService(db, stripe);
|
||||
|
||||
await service.cancel("user-1");
|
||||
|
||||
expect(stripe.setCancelAtPeriodEnd).toHaveBeenCalledWith("sub_stripe1", true);
|
||||
});
|
||||
|
||||
it("cancel() skips Stripe for legacy one-time subscriptions", async () => {
|
||||
const sub = { id: "sub-1", stripeSubscriptionId: null, status: "active" };
|
||||
const db = dbWithSub(sub, { ...sub, status: "cancelled" });
|
||||
const stripe = { setCancelAtPeriodEnd: vi.fn() };
|
||||
const service = createService(db, stripe);
|
||||
|
||||
await service.cancel("user-1");
|
||||
|
||||
expect(stripe.setCancelAtPeriodEnd).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("resume() re-enables Stripe auto-renewal for a recurring subscription", async () => {
|
||||
const sub = { id: "sub-1", stripeSubscriptionId: "sub_stripe1", status: "cancelled" };
|
||||
const db = dbWithSub(sub, { ...sub, status: "active" });
|
||||
const stripe = { setCancelAtPeriodEnd: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = createService(db, stripe);
|
||||
|
||||
await service.resume("user-1");
|
||||
|
||||
expect(stripe.setCancelAtPeriodEnd).toHaveBeenCalledWith("sub_stripe1", false);
|
||||
});
|
||||
|
||||
it("resume() maps a closed Stripe subscription to ConflictException (fresh checkout needed)", async () => {
|
||||
const sub = { id: "sub-1", stripeSubscriptionId: "sub_gone", status: "cancelled" };
|
||||
const db = dbWithSub(sub, { ...sub, status: "active" });
|
||||
const stripe = {
|
||||
setCancelAtPeriodEnd: vi.fn().mockRejectedValue(new Error("No such subscription")),
|
||||
};
|
||||
const service = createService(db, stripe);
|
||||
|
||||
await expect(service.resume("user-1")).rejects.toThrow(ConflictException);
|
||||
});
|
||||
});
|
||||
|
||||
describe("activateSubscription", () => {
|
||||
it("is idempotent: skips re-activation when already active (no revenue double-count)", async () => {
|
||||
// A Stripe webhook retry must not re-fire revenue events, re-consume
|
||||
@@ -316,7 +435,13 @@ describe("SubscriptionsService", () => {
|
||||
flush: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
const metaCapi = { sendPurchase: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = new SubscriptionsService(db as any, posthog as any, metaCapi as any);
|
||||
const stripe = { setCancelAtPeriodEnd: vi.fn() };
|
||||
const service = new SubscriptionsService(
|
||||
db as any,
|
||||
posthog as any,
|
||||
metaCapi as any,
|
||||
stripe as any,
|
||||
);
|
||||
|
||||
const result = await service.activateSubscription("sub-1");
|
||||
|
||||
@@ -326,5 +451,92 @@ describe("SubscriptionsService", () => {
|
||||
expect(posthog.captureForUser).not.toHaveBeenCalled();
|
||||
expect(metaCapi.sendPurchase).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("supersedes the live trial only when the paid subscription actually activates", async () => {
|
||||
const pendingSub = {
|
||||
id: "sub-2",
|
||||
userId: "user-1",
|
||||
planId: "plan-1",
|
||||
status: "pending",
|
||||
billingPeriod: "monthly",
|
||||
};
|
||||
const updateChains: Array<{ set: ReturnType<typeof vi.fn> }> = [];
|
||||
const db = {
|
||||
select: vi.fn().mockImplementation(() => ({
|
||||
from: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnThis(),
|
||||
innerJoin: vi.fn().mockReturnThis(),
|
||||
orderBy: vi.fn().mockReturnThis(),
|
||||
limit: vi.fn().mockReturnValue([pendingSub]),
|
||||
})),
|
||||
update: vi.fn().mockImplementation(() => {
|
||||
const chain = {
|
||||
set: vi.fn(),
|
||||
where: vi.fn(),
|
||||
returning: vi.fn().mockReturnValue([{ ...pendingSub, status: "active" }]),
|
||||
};
|
||||
chain.set.mockReturnValue(chain);
|
||||
chain.where.mockReturnValue(chain);
|
||||
updateChains.push(chain);
|
||||
return chain;
|
||||
}),
|
||||
insert: vi.fn().mockReturnValue({
|
||||
values: vi.fn().mockResolvedValue(undefined),
|
||||
}),
|
||||
};
|
||||
const service = createService(db);
|
||||
|
||||
await service.activateSubscription("sub-2");
|
||||
|
||||
const setPayloads = updateChains.flatMap((c) =>
|
||||
c.set.mock.calls.map((call) => call[0] as { status?: string }),
|
||||
);
|
||||
// One update activates the paid sub, exactly one expires the trial.
|
||||
expect(setPayloads.filter((p) => p?.status === "active")).toHaveLength(1);
|
||||
expect(setPayloads.filter((p) => p?.status === "expired")).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getMySubscription", () => {
|
||||
it("returns the live trial instead of the newer expired remains of an abandoned checkout", async () => {
|
||||
// After an abandoned checkout the newest row is the dead pending
|
||||
// (pending→expired). The user's real standing is their still-live trial.
|
||||
const expiredPending = { id: "sub-dead", status: "expired", planId: "plan-1" };
|
||||
const liveTrial = { id: "sub-trial", status: "trial", planId: "plan-trial" };
|
||||
let callCount = 0;
|
||||
const db = {
|
||||
select: vi.fn().mockImplementation(() => {
|
||||
callCount++;
|
||||
if (callCount === 1) {
|
||||
// all subs, newest first
|
||||
return {
|
||||
from: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnThis(),
|
||||
orderBy: vi.fn().mockReturnValue([expiredPending, liveTrial]),
|
||||
};
|
||||
}
|
||||
if (callCount === 2) {
|
||||
// brands of the chosen sub (innerJoin chain resolves at .where)
|
||||
return {
|
||||
from: vi.fn().mockReturnThis(),
|
||||
innerJoin: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnValue([]),
|
||||
};
|
||||
}
|
||||
// plan lookup
|
||||
return {
|
||||
from: vi.fn().mockReturnThis(),
|
||||
where: vi.fn().mockReturnThis(),
|
||||
limit: vi.fn().mockReturnValue([]),
|
||||
};
|
||||
}),
|
||||
};
|
||||
const service = createService(db);
|
||||
|
||||
const result = await service.getMySubscription("user-1");
|
||||
|
||||
expect(result?.id).toBe("sub-trial");
|
||||
expect(result?.status).toBe("trial");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
Injectable,
|
||||
Logger,
|
||||
NotFoundException,
|
||||
forwardRef,
|
||||
} from "@nestjs/common";
|
||||
import { and, desc, eq, inArray, or } from "drizzle-orm";
|
||||
import { DATABASE, type Database } from "../database/database.provider";
|
||||
@@ -17,6 +18,7 @@ import {
|
||||
users,
|
||||
} from "../database/schema/core";
|
||||
import { MetaCapiService } from "../meta-capi/meta-capi.service";
|
||||
import { StripeService } from "../payments/stripe/stripe.service";
|
||||
import { PostHogService } from "../posthog/posthog.service";
|
||||
|
||||
@Injectable()
|
||||
@@ -27,6 +29,11 @@ export class SubscriptionsService {
|
||||
@Inject(DATABASE) private db: Database,
|
||||
private posthog: PostHogService,
|
||||
private metaCapi: MetaCapiService,
|
||||
// forwardRef: StripeService also injects SubscriptionsService (webhook
|
||||
// activation), so the two providers are circular. cancel()/resume() must
|
||||
// sync auto-renewal to Stripe for recurring subscriptions.
|
||||
@Inject(forwardRef(() => StripeService))
|
||||
private stripeService: StripeService,
|
||||
) {}
|
||||
|
||||
/**
|
||||
@@ -64,11 +71,11 @@ export class SubscriptionsService {
|
||||
throw new ConflictException("Zaten aktif bir aboneliğiniz var");
|
||||
}
|
||||
|
||||
// Expire any existing trial subscription
|
||||
await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ status: "expired", updatedAt: new Date() })
|
||||
.where(and(eq(userSubscriptions.userId, userId), eq(userSubscriptions.status, "trial")));
|
||||
// Deliberately do NOT touch an existing trial here. create() runs the
|
||||
// moment the user clicks "Öde" — before any money moves — and expiring
|
||||
// the trial at that point locked out everyone who then abandoned the
|
||||
// Stripe page: they lost their remaining trial days without ever paying.
|
||||
// The trial is superseded in activateSubscription(), once payment lands.
|
||||
|
||||
// Expire any existing pending subscription so an abandoned checkout
|
||||
// doesn't block the user from starting a new one with different choices.
|
||||
@@ -161,6 +168,15 @@ export class SubscriptionsService {
|
||||
.where(eq(userSubscriptions.id, subscriptionId))
|
||||
.returning();
|
||||
|
||||
// Supersede any live trial now that a PAID subscription has taken over.
|
||||
// Checkout start intentionally leaves the trial alone (see create()), so
|
||||
// an abandoned checkout keeps trial access intact; this is the single
|
||||
// point where a trial legitimately ends early.
|
||||
await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ status: "expired", updatedAt: now })
|
||||
.where(and(eq(userSubscriptions.userId, sub.userId), eq(userSubscriptions.status, "trial")));
|
||||
|
||||
// Get the plan to determine brands
|
||||
const plan = await this.db.select().from(plans).where(eq(plans.id, sub.planId)).limit(1);
|
||||
|
||||
@@ -184,7 +200,9 @@ export class SubscriptionsService {
|
||||
// activation, so total paid revenue becomes measurable in PostHog regardless
|
||||
// of payment method (a Stripe DWH connector alone would miss EFT/havale).
|
||||
// $revenue is in major TRY (PostHog revenue convention); plan prices are kuruş.
|
||||
// This is the single source of $revenue — funnel steps (payment_initiated etc.)
|
||||
// First-charge revenue lives here; renewal revenue is captured as
|
||||
// subscription_renewed (stripe.service handleInvoicePaid). Together they are
|
||||
// the only $revenue sources — funnel steps (payment_initiated etc.)
|
||||
// intentionally do NOT carry $revenue so revenue isn't double-counted.
|
||||
const priceKurus =
|
||||
sub.billingPeriod === "yearly" ? (plan[0]?.priceYearly ?? 0) : (plan[0]?.priceMonthly ?? 0);
|
||||
@@ -245,16 +263,27 @@ export class SubscriptionsService {
|
||||
}
|
||||
|
||||
async getMySubscription(userId: string) {
|
||||
const result = await this.db
|
||||
const subs = await this.db
|
||||
.select()
|
||||
.from(userSubscriptions)
|
||||
.where(and(eq(userSubscriptions.userId, userId)))
|
||||
.orderBy(desc(userSubscriptions.createdAt))
|
||||
.limit(1);
|
||||
.where(eq(userSubscriptions.userId, userId))
|
||||
.orderBy(desc(userSubscriptions.createdAt));
|
||||
|
||||
if (result.length === 0) return null;
|
||||
if (subs.length === 0) return null;
|
||||
|
||||
const sub = result[0];
|
||||
// Pick the row that best represents the user's current standing rather
|
||||
// than blindly the newest one: a live trial must not be eclipsed by the
|
||||
// expired remains of an abandoned checkout (pending→expired), which is
|
||||
// always the newer row. A live pending still outranks the trial so the
|
||||
// complete/cancel-payment UI stays reachable. Newest wins within a tier.
|
||||
const statusPriority: Record<string, number> = {
|
||||
active: 0,
|
||||
pending: 1,
|
||||
trial: 2,
|
||||
cancelled: 3,
|
||||
};
|
||||
const rank = (s: { status: string }) => statusPriority[s.status] ?? 4;
|
||||
const sub = subs.reduce((best, cur) => (rank(cur) < rank(best) ? cur : best));
|
||||
const subBrands = await this.db
|
||||
.select({ brandId: userBrands.brandId, brandName: brands.name })
|
||||
.from(userBrands)
|
||||
@@ -284,6 +313,13 @@ export class SubscriptionsService {
|
||||
|
||||
if (!sub) throw new NotFoundException("Aktif abonelik bulunamadı");
|
||||
|
||||
// Recurring (Stripe-billed) sub: stop auto-renewal at Stripe FIRST. If that
|
||||
// call fails we keep our row active — a DB row that says "cancelled" while
|
||||
// the card keeps being charged is the one unacceptable state.
|
||||
if (sub.stripeSubscriptionId) {
|
||||
await this.stripeService.setCancelAtPeriodEnd(sub.stripeSubscriptionId, true);
|
||||
}
|
||||
|
||||
const [updated] = await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ status: "cancelled", cancelledAt: new Date(), updatedAt: new Date() })
|
||||
@@ -302,6 +338,19 @@ export class SubscriptionsService {
|
||||
|
||||
if (!sub) throw new NotFoundException("Devam ettirilecek iptal edilmiş abonelik bulunamadı");
|
||||
|
||||
// Mirror of cancel(): re-enable auto-renewal at Stripe first. If the Stripe
|
||||
// subscription is already fully closed (period ended), resuming is no
|
||||
// longer possible — the user needs a fresh checkout.
|
||||
if (sub.stripeSubscriptionId) {
|
||||
try {
|
||||
await this.stripeService.setCancelAtPeriodEnd(sub.stripeSubscriptionId, false);
|
||||
} catch {
|
||||
throw new ConflictException(
|
||||
"Aboneliğin yenilemesi tamamen kapanmış — devam ettirmek için yeni bir ödeme başlatın",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const [updated] = await this.db
|
||||
.update(userSubscriptions)
|
||||
.set({ status: "active", cancelledAt: null, updatedAt: new Date() })
|
||||
|
||||
@@ -631,6 +631,45 @@ describe("VehiclesService", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("extractYearFromPcatCar (filter-axis shadowing)", () => {
|
||||
// Real prod payload for Opel ASTRA-J W0VPD5EC1MG063839: the faceting axis
|
||||
// `year`="All" precedes the concrete `Year`="2021" param. The old find()
|
||||
// grabbed "All" → NaN → null, leaving the vehicle year blank.
|
||||
const astraJParams = [
|
||||
{ key: "year", name: "Year", value: "All" },
|
||||
{ key: "sales_region", name: "Region", value: "Another region" },
|
||||
{ key: "Model", name: "Model", value: "D69 (4 Door Saloon) (Enjoy / Exclusiv)" },
|
||||
{ key: "Engine", name: "Engine", value: "A14NET (LUJ)" },
|
||||
{ key: "Year", name: "Year", value: "2021" },
|
||||
{ key: "production_date", name: "Production date", value: "2020/10/05" },
|
||||
];
|
||||
|
||||
it("picks the concrete Year param, not the year=All filter axis", () => {
|
||||
const { service } = createService();
|
||||
const year = (
|
||||
service as unknown as {
|
||||
extractYearFromPcatCar: (c: { parameters: unknown[] }) => number | null;
|
||||
}
|
||||
).extractYearFromPcatCar({ parameters: astraJParams });
|
||||
expect(year).toBe(2021);
|
||||
});
|
||||
|
||||
it("falls back to production_date when no numeric Year param exists", () => {
|
||||
const { service } = createService();
|
||||
const year = (
|
||||
service as unknown as {
|
||||
extractYearFromPcatCar: (c: { parameters: unknown[] }) => number | null;
|
||||
}
|
||||
).extractYearFromPcatCar({
|
||||
parameters: [
|
||||
{ key: "year", name: "Year", value: "All" },
|
||||
{ key: "production_date", name: "Production date", value: "2020/10/05" },
|
||||
],
|
||||
});
|
||||
expect(year).toBe(2020);
|
||||
});
|
||||
});
|
||||
|
||||
// ─── Q2: PL24 circuit breaker must only count transient transport faults ───
|
||||
describe("PL24 circuit breaker fault classification (Q2)", () => {
|
||||
it("does NOT trip on a definitive upstream negative (non-transient)", async () => {
|
||||
|
||||
@@ -1186,12 +1186,27 @@ export class VehiclesService {
|
||||
|
||||
private extractYearFromPcatCar(car: PcatCar): number | null {
|
||||
if (!car.parameters) return null;
|
||||
const yearParam = car.parameters.find(
|
||||
(p) => p.key.toLowerCase().includes("year") || p.key.toLowerCase().includes("model_year"),
|
||||
);
|
||||
if (yearParam?.value) {
|
||||
const num = Number.parseInt(yearParam.value, 10);
|
||||
if (num > 1900 && num < 2100) return num;
|
||||
// pcat returns a faceting AXIS param `year` (often value "All") BEFORE the
|
||||
// concrete per-car `Year` param, so a naive find(includes("year")) grabs the
|
||||
// axis → parseInt("All") = NaN → null. Scan every year-ish param and take the
|
||||
// first that parses to a plausible year; only then fall back to a
|
||||
// production_date like "2020/10/05".
|
||||
const plausible = (raw: string | undefined): number | null => {
|
||||
const num = Number.parseInt(raw ?? "", 10);
|
||||
return num > 1900 && num < 2100 ? num : null;
|
||||
};
|
||||
for (const p of car.parameters) {
|
||||
const key = p.key.toLowerCase();
|
||||
if (key.includes("year") || key.includes("model_year")) {
|
||||
const year = plausible(p.value);
|
||||
if (year) return year;
|
||||
}
|
||||
}
|
||||
for (const p of car.parameters) {
|
||||
if (p.key.toLowerCase().includes("production_date")) {
|
||||
const year = plausible(p.value);
|
||||
if (year) return year;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -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