feat: المجموعة B شريحة 2 — نقاط توقف + جدولة + مطابقة الوجهة

B4 نقاط التوقف: stops jsonb؛ المسار والتسعير يمرّان بالنقاط بالترتيب (أصل → توقف₁ → … → وجهة) بجمع أرجل المسار. بلا توقفات = رِجل واحدة كالسابق. B5 الحجز المسبق: scheduled_at + حالة scheduled جديدة (لا تدخل Redis ولا تُوزَّع الآن). ScheduledTripsSweeper يعمل داخل الـAPI لا الـworker — لأن الإطلاق يبثّ عروضاً عبر RealtimeGateway الموجود في الـAPI فقط. آمن مع عدة نسخ: releaseDueScheduled يستعمل UPDATE شرطياً (scheduled→searching) فلا تُطلَق رحلة مرتين. فهرس جزئي على scheduled المستحقة. B8 مطابقة الوجهة: is_destination_match + destination_match_discount + destination_match_discount_pct في التعرفة، يُطبَّق في settleFare. الخصم على الراكب والسائق يقبض المخفَّض كاملاً (يربح رحلة في طريقه أصلاً). مؤجَّل: محرّك المطابقة الذي يضبط العلم (وضع وجهة السائق). هجرة: TripStopsSchedulingMatch. حالة trip جديدة: scheduled (تنتقل إلى searching أو cancelled). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-17 19:49:10 +03:00
parent c2a4b9d1a6
commit a58f611063
8 changed files with 205 additions and 12 deletions
@@ -70,6 +70,11 @@ export default () => ({
    },
  },

+  // الرحلات: كل كم يفحص كنس الرحلات المجدولة (docs/17 — B5).
+  trips: {
+    scheduleSweepMs: parseInt(process.env.TRIP_SCHEDULE_SWEEP_MS ?? '30000', 10),
+  },
+
  nabeh: {
    baseUrl: process.env.NABEH_BASE_URL ?? 'https://nabeh.intaleqapp.com',
    email: process.env.NABEH_EMAIL ?? '',
@@ -0,0 +1,34 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+/**
+ * شريحة B الثانية (docs/17 — B4/B5/B8):
+ * - نقاط التوقف (`stops` jsonb).
+ * - الحجز المسبق (`scheduled_at` + حالة `scheduled`).
+ * - مطابقة الوجهة (`is_destination_match` + `destination_match_discount`).
+ */
+export class TripStopsSchedulingMatch1721920000000 implements MigrationInterface {
+  public async up(q: QueryRunner): Promise<void> {
+    await q.query(`ALTER TABLE tripz_trips ADD COLUMN IF NOT EXISTS stops jsonb NOT NULL DEFAULT '[]'`);
+    await q.query(`ALTER TABLE tripz_trips ADD COLUMN IF NOT EXISTS scheduled_at timestamptz`);
+    await q.query(
+      `ALTER TABLE tripz_trips ADD COLUMN IF NOT EXISTS is_destination_match boolean NOT NULL DEFAULT false`,
+    );
+    await q.query(
+      `ALTER TABLE tripz_trips ADD COLUMN IF NOT EXISTS destination_match_discount numeric(12,3)`,
+    );
+    // الكنس يبحث عن scheduled المستحقة زمنياً — فهرس جزئي عليها فقط.
+    await q.query(`
+      CREATE INDEX IF NOT EXISTS "IDX_tripz_trips_scheduled_due"
+      ON tripz_trips (scheduled_at)
+      WHERE status = 'scheduled'
+    `);
+  }
+
+  public async down(q: QueryRunner): Promise<void> {
+    await q.query(`DROP INDEX IF EXISTS "IDX_tripz_trips_scheduled_due"`);
+    await q.query(`ALTER TABLE tripz_trips DROP COLUMN IF EXISTS destination_match_discount`);
+    await q.query(`ALTER TABLE tripz_trips DROP COLUMN IF EXISTS is_destination_match`);
+    await q.query(`ALTER TABLE tripz_trips DROP COLUMN IF EXISTS scheduled_at`);
+    await q.query(`ALTER TABLE tripz_trips DROP COLUMN IF EXISTS stops`);
+  }
+}
@@ -69,4 +69,7 @@ export interface TariffDefinition {
   * **لا تُقتطع من أجرة السائق** — تُخصم من رصيده التشغيلي المدفوع سلفاً.
   */
  commission?: { percent?: number; flat?: number; min?: number };
+
+  /** نسبة خصم مطابقة الوجهة (docs/17 — B8). غيابها = لا خصم. */
+  destination_match_discount_pct?: number;
 }
@@ -8,6 +8,7 @@ import {
 } from 'typeorm';

 export type TripStatus =
+  | 'scheduled' // محجوزة لوقت لاحق — لا تُوزَّع حتى يحين موعدها (docs/17 — B5)
  | 'searching'
  | 'assigned'
  | 'driver_arriving'
@@ -19,6 +20,13 @@ export type TripStatus =
  | 'no_drivers'
  | 'expired';

+/** نقطة توقف ضمن الرحلة (docs/17 — B4). */
+export interface TripStop {
+  lat: number;
+  lng: number;
+  label?: string;
+}
+
 /** الرحلة. الجدول: tripz_trips. آلة الحالة مصدرها الباك إند (docs/02). */
@Entity('trips')
@Index(['tenant_id', 'status'])
@@ -115,6 +123,26 @@ export class Trip {
  @Column({ default: 'cash' })
  payment_method: string;

+  // ---- نقاط التوقف (docs/17 — B4) ----
+  // نقاط وسيطة بين الأصل والوجهة؛ المسار والتسعير يمرّان بها بالترتيب.
+  @Column({ type: 'jsonb', default: [] })
+  stops: TripStop[];
+
+  // ---- الحجز المسبق (docs/17 — B5) ----
+  // موعد الرحلة المجدولة؛ يبقى null للرحلات الفورية. الرحلة المجدولة تنتظر
+  // بحالة `scheduled` حتى يطلقها الـworker قبيل موعدها.
+  @Column({ type: 'timestamptz', nullable: true })
+  scheduled_at: Date | null;
+
+  // ---- مطابقة الوجهة (docs/17 — B8) ----
+  // صحيحة حين تتوافق وجهة الرحلة مع اتجاه السائق (وضع «السائق عائد») —
+  // يُطبَّق خصم مطابقة الوجهة على الراكب.
+  @Column({ default: false })
+  is_destination_match: boolean;
+
+  @Column({ type: 'numeric', precision: 12, scale: 3, nullable: true })
+  destination_match_discount: number | null;
+
  @Column({ type: 'varchar', nullable: true })
  cancelled_by: string | null; // rider | driver

@@ -0,0 +1,46 @@
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { TripsService } from './trips.service';
+
+/**
+ * يطلق الرحلات المجدولة التي حان موعدها (docs/17 — B5).
+ *
+ * يعمل **داخل عملية الـAPI** لا الـworker: الإطلاق يبثّ عروضاً عبر
+ * RealtimeGateway (WebSocket) الموجود هنا فقط. آمن مع عدة نسخ API لأن
+ * `releaseDueScheduled` يستعمل UPDATE شرطياً (scheduled→searching) فلا
+ * تُطلَق رحلة مرتين.
+ */
+@Injectable()
+export class ScheduledTripsSweeper implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger('ScheduledTrips');
+  private timer?: NodeJS.Timeout;
+  private running = false;
+
+  constructor(
+    private readonly trips: TripsService,
+    private readonly config: ConfigService,
+  ) {}
+
+  onModuleInit(): void {
+    const intervalMs = this.config.get<number>('trips.scheduleSweepMs') ?? 30_000;
+    this.timer = setInterval(() => void this.tick(), intervalMs);
+    this.logger.log(`scheduled-trips sweeper up (every ${intervalMs}ms)`);
+  }
+
+  onModuleDestroy(): void {
+    if (this.timer) clearInterval(this.timer);
+  }
+
+  private async tick(): Promise<void> {
+    if (this.running) return; // لا تتراكب دورة بطيئة مع التالية
+    this.running = true;
+    try {
+      const n = await this.trips.releaseDueScheduled();
+      if (n > 0) this.logger.log(`released ${n} scheduled trip(s)`);
+    } catch (e: any) {
+      this.logger.error(`sweep failed: ${e?.message}`);
+    } finally {
+      this.running = false;
+    }
+  }
+}
@@ -3,6 +3,7 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 import { Trip } from './entities/trip.entity';
 import { TripEvent } from './entities/trip-event.entity';
 import { TripsService } from './trips.service';
+import { ScheduledTripsSweeper } from './scheduled-trips.sweeper';
 import { TripStateService } from './trip-state.service';
 import { TripsController } from './trips.controller';
 import { MapsModule } from '../maps/maps.module';
@@ -31,7 +32,7 @@ import { CreditModule } from '../credit/credit.module';
    CreditModule,
  ],
  controllers: [TripsController],
-  providers: [TripsService, TripStateService],
+  providers: [TripsService, TripStateService, ScheduledTripsSweeper],
  exports: [TripsService],
 })
 export class TripsModule {}
@@ -6,7 +6,7 @@ import {
  NotFoundException,
 } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
-import { Repository } from 'typeorm';
+import { LessThanOrEqual, Repository } from 'typeorm';
 import { Trip, TripStatus } from './entities/trip.entity';
 import { TripEvent } from './entities/trip-event.entity';
 import { MapsService } from '../maps/maps.service';
@@ -33,6 +33,8 @@ const CANCEL_FEE_BY_STAGE: Partial<Record<TripStatus, number>> = {

 /** الانتقالات المسموحة في آلة حالة الرحلة (docs/02). */
 const TRANSITIONS: Record<TripStatus, TripStatus[]> = {
+  // المجدولة تُطلَق إلى searching عند موعدها، أو تُلغى قبله (docs/17 — B5).
+  scheduled: ['searching', 'cancelled', 'expired'],
  searching: ['assigned', 'cancelled', 'no_drivers', 'expired'],
  assigned: ['driver_arriving', 'cancelled'],
  driver_arriving: ['driver_arrived', 'cancelled'],
@@ -55,6 +57,8 @@ export interface RequestTripDto {
  city?: string;
  is_round_trip?: boolean;
  rider_id?: string; // يُستخدم من dispatch لإنشاء رحلة نيابةً عن راكب
+  stops?: Array<{ lat: number; lng: number; label?: string }>; // docs/17 — B4
+  scheduled_at?: string; // ISO — حجز مسبق (docs/17 — B5)
 }

@Injectable()
@@ -104,13 +108,29 @@ export class TripsService {
    const serviceClass = dto.service_class ?? 'economy';
    const city = dto.city ?? 'default';
    const isRound = !!dto.is_round_trip;
-    const oneWay = await this.maps.route(dto.origin, dto.destination);
+
+    // المسار عبر نقاط التوقف بالترتيب (docs/17 — B4): أصل → توقف₁ → … → وجهة.
+    // بلا توقفات = رِجل واحدة كالسابق.
+    const stops = Array.isArray(dto.stops) ? dto.stops : [];
+    const legPoints = [dto.origin, ...stops, dto.destination];
+    let legDistance = 0;
+    let legDuration = 0;
+    for (let i = 0; i < legPoints.length - 1; i++) {
+      const leg = await this.maps.route(legPoints[i], legPoints[i + 1]);
+      legDistance += leg.distanceKm;
+      legDuration += leg.durationMin;
+    }
    // ذهاب وعودة: يُضاعف المسافة والزمن للتسعير
    const route = {
-      distanceKm: isRound ? Number((oneWay.distanceKm * 2).toFixed(3)) : oneWay.distanceKm,
-      durationMin: isRound ? Number((oneWay.durationMin * 2).toFixed(1)) : oneWay.durationMin,
+      distanceKm: Number((isRound ? legDistance * 2 : legDistance).toFixed(3)),
+      durationMin: Number((isRound ? legDuration * 2 : legDuration).toFixed(1)),
    };

+    // الحجز المسبق (docs/17 — B5): موعد مستقبلي → الرحلة تنتظر بحالة scheduled
+    // ولا تُوزَّع الآن؛ الـworker يطلقها قبيل موعدها.
+    const scheduledAt = dto.scheduled_at ? new Date(dto.scheduled_at) : null;
+    const isScheduled = scheduledAt != null && scheduledAt.getTime() > Date.now() + 60_000;
+
    // تسعير (اختياري — لو ما في تعرفة مفعّلة نكمل بلا سعر مقفول)
    let quotedFare: number | null = null;
    let currency: string | null = null;
@@ -149,7 +169,9 @@ export class TripsService {
      origin_lng: dto.origin.lng,
      dest_lat: dto.destination.lat,
      dest_lng: dto.destination.lng,
-      status: 'searching',
+      stops,
+      scheduled_at: scheduledAt,
+      status: isScheduled ? 'scheduled' : 'searching',
      distance_km: route.distanceKm,
      duration_min: route.durationMin,
      tariff_id: tariffId,
@@ -164,6 +186,13 @@ export class TripsService {
      payment_method: (dto as any).payment_method ?? 'cash',
    });
    trip = await this.trips.save(trip);
+
+    // المجدولة لا تدخل Redis ولا تُوزَّع الآن — تنتظر الـworker (docs/17 — B5).
+    if (isScheduled) {
+      await this.recordEvent(trip, null, 'scheduled', 'rider');
+      return { trip, offeredDrivers: 0, scheduled: true };
+    }
+
    await this.state.save(trip);
    await this.recordEvent(trip, null, 'searching', 'rider');

@@ -174,6 +203,42 @@ export class TripsService {
    return { trip, offeredDrivers: offered };
  }

+  /**
+   * إطلاق الرحلات المجدولة التي حان موعدها (docs/17 — B5). يناديها الـworker
+   * دورياً. كل رحلة تنتقل `scheduled → searching` وتُوزَّع كرحلة فورية.
+   * يرجع عدد ما أُطلق.
+   */
+  async releaseDueScheduled(tenantId?: string, leadMinutes = 10): Promise<number> {
+    const dueBefore = new Date(Date.now() + leadMinutes * 60_000);
+    const where: any = { status: 'scheduled' };
+    if (tenantId) where.tenant_id = tenantId;
+    const due = await this.trips.find({
+      where: { ...where, scheduled_at: LessThanOrEqual(dueBefore) },
+      take: 100,
+    });
+
+    let released = 0;
+    for (const trip of due) {
+      // UPDATE شرطي: نسختا worker متزامنتان لا تطلقان نفس الرحلة مرتين.
+      const res = await this.trips.update(
+        { id: trip.id, status: 'scheduled' },
+        { status: 'searching' },
+      );
+      if (!res.affected) continue;
+
+      trip.status = 'searching';
+      await this.state.save(trip);
+      await this.recordEvent(trip, 'scheduled', 'searching', 'rider');
+      await this.offerToNearbyDrivers(trip.tenant_id, trip, {
+        lat: trip.origin_lat,
+        lng: trip.origin_lng,
+      });
+      this.gateway.dispatch(trip.tenant_id, 'trip:new', { tripId: trip.id });
+      released++;
+    }
+    return released;
+  }
+
  /** قبول سائق للرحلة (أول قبول يفوز — docs/17 A4). */
  async accept(tenantId: string, tripId: string, driverUserId: string) {
    const driver = await this.drivers.findByUser(tenantId, driverUserId);
@@ -554,18 +619,29 @@ export class TripsService {
    // الانتظار ومشوار الوصول حُسبا وخُزّنا سابقاً — نقرأهما من الصف.
    const row = await this.trips.findOne({
      where: { tenant_id: tenantId, id: tripId },
-      select: { waiting_charge: true, pickup_charge: true },
+      select: { waiting_charge: true, pickup_charge: true, is_destination_match: true },
    });
    const base = snapshot.quoted_fare ?? 0;
    const waiting = Number(row?.waiting_charge ?? 0);
    // مشوار الوصول لا يدخل أجرة الرحلة — هو تعويض إلغاء فقط (docs/18، تصحيح B3).
-    const total = Number((base + waiting).toFixed(3));
+    const gross = base + waiting;

    const def = await this.tariffDef(tenantId, snapshot);
+
+    // خصم مطابقة الوجهة (docs/17 — B8): الرحلة توافق اتجاه السائق فيُخفَّض
+    // على الراكب. الخصم من أجرة الراكب لا من دخل السائق — السائق يقبض الأجرة
+    // المخفَّضة كاملة (يربح رحلةً في طريقه أصلاً).
+    let discount = 0;
+    if (row?.is_destination_match && def?.destination_match_discount_pct) {
+      discount = Number((gross * (def.destination_match_discount_pct / 100)).toFixed(3));
+    }
+    const total = Number((gross - discount).toFixed(3));
+
    const commission = def ? TariffEngine.commission(def, total) : { amount: 0, rate: 0 };

    return {
      final_fare: total,
+      destination_match_discount: discount || null,
      // السائق يقبض ما يدفعه الراكب **كاملاً** (docs/18). العمولة تُخصم من
      // رصيده التشغيلي لا من هذا المبلغ.
      price_for_passenger: total,
@@ -32,9 +32,9 @@
 | B3 | **احتساب مشوار الوصول للراكب** | يُقاس من موقع السائق **الحيّ لحظة القبول** حتى نقطة الالتقاط. | 🟡 **القياس ✅ والقاعدة ❌** — راجع التصحيح أدناه |
 | B6 | **فصل السعر** | `price_for_passenger` · `price_for_driver` · `commission_amount` · `commission_rate`. العمولة تُعرَّف في التعرفة: `commission: { percent, flat, min }`، ولا تتجاوز الأجرة أبداً. | ✅ + **تسوية المحفظة صُحّحت**: كانت تُضيف للسائق كامل أجرة الراكب — أي أن العمولة كانت تضيع |
 | B7 | **طوابع زمنية دقيقة** | `driver_going_at` (`DriverIsGoingToPassenger`) · `arrived_at` · `started_at` (`rideTimeStart`) · `completed_at` (`rideTimeFinish`). | ✅ + كشف «الإنهاء السريع» صار يقيس من **بدء** الرحلة لا إسنادها |
-| B4 | **نقاط توقف (stops)** | نقطتا توقف ضمن الرحلة (كما في سيرو/شير). | ⏳ الشريحة التالية |
-| B5 | **حجز مسبق/جدولة** | جداول الرحلة (`date`/`time`/`endtime` في سيرو). | ⏳ الشريحة التالية |
-| B8 | **مطابقة الوجهة** | `is_destination_match` + خصم الراكب. | ⏳ الشريحة التالية |
+| B4 | **نقاط توقف (stops)** | ✅ `stops` jsonb؛ المسار والتسعير يمرّان بالنقاط بالترتيب (أصل → توقف₁ → … → وجهة) بجمع أرجل المسار. |
+| B5 | **حجز مسبق/جدولة** | ✅ `scheduled_at` + حالة `scheduled` (لا تدخل Redis ولا تُوزَّع). `ScheduledTripsSweeper` **داخل الـAPI** (يملك الـgateway) يطلقها قبيل موعدها بـUPDATE شرطي (آمن عبر النسخ). |
+| B8 | **مطابقة الوجهة** | 🟡 الحقول والخصم منفَّذان: `is_destination_match` + `destination_match_discount` + `destination_match_discount_pct` في التعرفة، يُطبَّق في التسوية (الخصم على الراكب، السائق يقبض المخفَّض كاملاً). **مؤجَّل**: محرّك المطابقة الذي **يضبط** العلم (وضع «وجهة السائق») — يحتاج تخزين وجهة السائق + مطابقة اتجاهية في محرّك التوزيع. |
 | B9 | **تعديل التعرفة من لوحة الأدمن** | جداول تعرفة قابلة للتحرير (موجودة كـ jsonb — نحتاج واجهة/نقاط CRUD). | ⏳ |
 | B10 | **كتالوج أنواع الرحلات العالمي** | `ride_types` فكرته سليمة (الأدمن/المستأجر يضيف أنواعه). المطلوب: كتالوج جاهز بما هو شائع عالمياً كخيارات جاهزة للاختيار (عندنا 6 فقط الآن). | ⏳ |
 | B11 | **تعرفة بالوزن** | بُعد تسعير إضافي بالوزن (للشحن/التوصيل) بجانب المسافة والزمن. | ⏳ |
@@ -287,7 +287,7 @@
 2. ~~**I1** (سباق المحفظة)~~ ✅ **منفَّذة ومُثبَتة بتزامن حقيقي**.
 3. ~~**G** (Redis خط أول: لغة/توكنات/tenant/تعرفة/تقييم/كاش الخرائط)~~ ✅ **منفَّذة**.
 4. ~~**H** (المواقع: إيقاف الكتابة لكل نبضة + batching + tracks)~~ ✅ **منفَّذة** (عدا H6/H7 — تحليلات مؤجَّلة).
-5. **B** — 🔵 الشريحة الأولى ✅ (الانتظار B2 · مشوار الوصول B3 · فصل السعر B6 · الطوابع B7). الباقي: B4 stops · B5 جدولة · B8 مطابقة الوجهة · B9 · B10 · B11.
+5. **B** — ✅ الشريحتان الأولى والثانية (B2·B3·B6·B7 · B4 stops · B5 جدولة · B8 مطابقة الوجهة). الباقي تحسينات: B9 (واجهة تعرفة) · B10 (كتالوج أنواع عالمي) · B11 (تعرفة بالوزن).
 6. **I** الباقي (المدفوعات: جداول لكل طريقة + OTP/بصمة/HMAC للسحب).
 7. **C** (بيانات المركبة والسائق + ai_data).
 8. **D** (تطبيع الهاتف + بصمة الجهاز + HMAC).