feat: المجموعة A (زمن حقيقي + FCM + Redis) + إصلاح سباق المحفظة
المجموعة A (docs/17): - A1: FCM على كل انتقال حالة (priority high) + حذف التوكنات الميتة - A2: common/i18n (ar/en) + عمود users.language — الإشعارات بلغة المستخدم - A3: TripStateService — حالة الرحلة الجارية في Redis hash (TTL 6س)؛ الانتقال صار UPDATE شرطي + قراءة واحدة بدل ~5 استعلامات - A4: قبول ذرّي — CAS بـLua في Redis + UPDATE ... WHERE status='searching' كحَكَم نهائي؛ أول سائق يفوز والباقي يُرفضون بلا لمس القاعدة - A5: مجموعة العروض في Redis + بث trip:offer_taken و FCM لبقية السائقين - A6: GET /trips/available — السائق يسحب الطلبات القريبة - A7: FCM data-only بحمولة كاملة للـoverlay I1 — إصلاح سباق المحفظة (ثغرة مالية): - credit/debit كانا read-modify-write على balance بلا قفل → خصمان متزامنان يكتبان فوق بعضهما. صارا UPDATE ذرّي واحد بشرط balance >= :amount، والقيد+الرصيد في معاملة واحدة - wallet_txns.balance_after للتدقيق + CHECK (balance >= 0) كشبكة أمان - إنشاء المحفظة عبر ON CONFLICT DO NOTHING (سباق ثانٍ كان كامناً) الاختبارات تعمل على السيرفر (docs/15): - npm test صار جزءاً من مرحلة builder — فشل اختبار = فشل بناء = لا نشر - pg-mem + ioredis-mock: بلا شبكة وبلا قاعدة حقيقية - scripts/wallet-race-test.mjs للتزامن الحقيقي على السيرفر Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Hamza-Ayed
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
b7e91e5de6
commit
9d6b752ea8
@@ -4,5 +4,6 @@ npm-debug.log
|
||||
.env
|
||||
.git
|
||||
.gitignore
|
||||
test
|
||||
**/*.spec.ts
|
||||
# ملاحظة: ملفات *.spec.ts تدخل السياق عمداً — الاختبارات تعمل داخل مرحلة
|
||||
# builder على السيرفر (راجع docs/15). tsconfig.build.json يستثنيها من dist،
|
||||
# فلا تصل صورة التشغيل.
|
||||
|
||||
@@ -8,6 +8,10 @@ COPY package*.json ./
|
||||
RUN npm install
|
||||
COPY . .
|
||||
RUN npm run build
|
||||
# بوابة الجودة: الاختبارات تعمل هنا — أي على السيرفر وقت البناء، لا على الماك
|
||||
# (قاعدة docs/15). تعتمد pg-mem و ioredis-mock فقط: بلا شبكة وبلا قاعدة حقيقية.
|
||||
# فشل أي اختبار = فشل البناء = لا نشر لكود مكسور.
|
||||
RUN npm test
|
||||
|
||||
# ---- runtime ----
|
||||
FROM node:22-alpine AS runtime
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
/** @type {import('jest').Config} */
|
||||
module.exports = {
|
||||
preset: 'ts-jest',
|
||||
testEnvironment: 'node',
|
||||
rootDir: 'src',
|
||||
testRegex: '.*\\.spec\\.ts$',
|
||||
};
|
||||
@@ -51,7 +51,9 @@
|
||||
"@types/jest": "^29.5.12",
|
||||
"@types/node": "^22.0.0",
|
||||
"@types/passport-jwt": "^4.0.1",
|
||||
"ioredis-mock": "^8.13.1",
|
||||
"jest": "^29.7.0",
|
||||
"pg-mem": "^3.0.14",
|
||||
"ts-jest": "^29.2.0",
|
||||
"ts-loader": "^9.5.1",
|
||||
"ts-node": "^10.9.2",
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
// اختبار تسابق المحفظة (docs/17 — I1): يثبت أن المال لا يُفقد ولا يُخلق تحت التزامن.
|
||||
// يستخدم fetch المدمج (Node 18+) — بلا تبعيات.
|
||||
//
|
||||
// لماذا سكربت منفصل: pg-mem في اختبارات jest أحادي الخيط فيثبت صحة الـSQL فقط،
|
||||
// أما التزامن الحقيقي على نفس الصف فيحتاج Postgres حقيقياً.
|
||||
//
|
||||
// التشغيل (حاوية node مؤقتة على شبكة tripz):
|
||||
// docker run --rm --network tripz-net -e BASE=http://tripz-api:4010/api \
|
||||
// -v /home/tripz-llc/backend/scripts:/s node:22-alpine node /s/wallet-race-test.mjs 100 5
|
||||
// الوسائط: <عدد محاولات الخصم المتزامنة> <مبلغ كل خصم>
|
||||
|
||||
const BASE = process.env.BASE || 'http://localhost:4010/api';
|
||||
const TENANT = process.env.TENANT || 'siro';
|
||||
const ATTEMPTS = parseInt(process.argv[2] || '100', 10);
|
||||
const AMOUNT = parseFloat(process.argv[3] || '5');
|
||||
const CODE = '1234';
|
||||
|
||||
const H = (t) => ({ 'Content-Type': 'application/json', ...(t ? { Authorization: `Bearer ${t}` } : {}) });
|
||||
|
||||
async function api(method, path, token, body) {
|
||||
const res = await fetch(`${BASE}${path}`, {
|
||||
method,
|
||||
headers: { ...H(token), 'x-tenant-id': TENANT },
|
||||
body: body ? JSON.stringify(body) : undefined,
|
||||
});
|
||||
const text = await res.text().catch(() => '');
|
||||
let json = {};
|
||||
try { json = text ? JSON.parse(text) : {}; } catch { /* نص غير JSON */ }
|
||||
return { ok: res.ok, status: res.status, body: json };
|
||||
}
|
||||
|
||||
async function token(phone) {
|
||||
const r = await api('POST', '/auth/verify-otp', null, { phone, code: CODE });
|
||||
if (!r.ok) throw new Error(`auth failed: ${r.status} ${JSON.stringify(r.body)}`);
|
||||
return r.body.access_token;
|
||||
}
|
||||
|
||||
async function main() {
|
||||
// مبلغ يكفي نصف المحاولات فقط — النصف الآخر *يجب* أن يُرفض.
|
||||
const expectedSuccesses = Math.floor(ATTEMPTS / 2);
|
||||
const funded = expectedSuccesses * AMOUNT;
|
||||
|
||||
const phone = `0796${String(Date.now()).slice(-6)}`;
|
||||
const t = await token(phone);
|
||||
|
||||
await api('POST', '/wallet/topup', t, { amount: funded });
|
||||
const before = await api('GET', '/wallet', t);
|
||||
const startBalance = Number(before.body.balance);
|
||||
console.log(`المحفظة: ${phone} | الرصيد الابتدائي = ${startBalance}`);
|
||||
if (startBalance !== funded) {
|
||||
console.log(`⚠️ الشحن لم يطابق المتوقع (${funded}) — أوقفت الاختبار.`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
// كل الطلبات تنطلق دفعة واحدة على نفس الصف.
|
||||
console.log(`إطلاق ${ATTEMPTS} خصم متزامن × ${AMOUNT} (يكفي ${expectedSuccesses} فقط)…`);
|
||||
const t0 = Date.now();
|
||||
const results = await Promise.all(
|
||||
Array.from({ length: ATTEMPTS }, () =>
|
||||
api('POST', '/payouts/request', t, { amount: AMOUNT, channel: 'cliq' }).catch((e) => ({
|
||||
ok: false,
|
||||
status: 0,
|
||||
body: { message: String(e) },
|
||||
})),
|
||||
),
|
||||
);
|
||||
const elapsed = Date.now() - t0;
|
||||
|
||||
const ok = results.filter((r) => r.ok).length;
|
||||
const rejected = results.filter((r) => !r.ok).length;
|
||||
|
||||
const after = await api('GET', '/wallet', t);
|
||||
const endBalance = Number(after.body.balance);
|
||||
const spent = startBalance - endBalance;
|
||||
|
||||
console.log('');
|
||||
console.log(`نجح : ${ok} (المتوقع ${expectedSuccesses})`);
|
||||
console.log(`رُفض : ${rejected}`);
|
||||
console.log(`الرصيد : ${startBalance} → ${endBalance} (خُصم ${spent})`);
|
||||
console.log(`الزمن : ${elapsed}ms`);
|
||||
console.log('');
|
||||
|
||||
// المحكّات الثلاثة
|
||||
const checks = [
|
||||
['الرصيد لم يصبح سالباً', endBalance >= 0],
|
||||
['عدد النجاحات = ما يسمح به الرصيد', ok === expectedSuccesses],
|
||||
['المخصوم = النجاحات × المبلغ (لا مال ضائع/مخلوق)', Math.abs(spent - ok * AMOUNT) < 1e-6],
|
||||
];
|
||||
let failed = 0;
|
||||
for (const [name, pass] of checks) {
|
||||
console.log(`${pass ? '✅' : '❌'} ${name}`);
|
||||
if (!pass) failed++;
|
||||
}
|
||||
|
||||
console.log('');
|
||||
console.log(failed === 0 ? '✅ لا سباق: المحفظة ذرّية.' : `❌ ${failed} محكّ فشل — يوجد سباق.`);
|
||||
process.exit(failed === 0 ? 0 : 1);
|
||||
}
|
||||
|
||||
main().catch((e) => {
|
||||
console.error('خطأ:', e.message);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -5,6 +5,7 @@ import { ThrottlerModule } from '@nestjs/throttler';
|
||||
import configuration from './config/configuration';
|
||||
import { TenantMiddleware } from './common/tenant/tenant.middleware';
|
||||
import { RedisModule } from './common/redis/redis.module';
|
||||
import { I18nModule } from './common/i18n/i18n.module';
|
||||
import { SeedModule } from './common/seed/seed.module';
|
||||
import { HealthModule } from './modules/health/health.module';
|
||||
import { TenantsModule } from './modules/tenants/tenants.module';
|
||||
@@ -54,6 +55,7 @@ import { GeminiModule } from './integrations/gemini/gemini.module';
|
||||
ThrottlerModule.forRoot([{ ttl: 60000, limit: 120 }]),
|
||||
|
||||
RedisModule, // عالمي — عميل Redis للمطابقة و OTP
|
||||
I18nModule, // عالمي — ترجمة نصوص الإشعارات
|
||||
NabehModule, // عالمي — إرسال OTP واتساب
|
||||
NotificationsModule, // عالمي — FCM
|
||||
StorageModule, // عالمي — تخزين ملفات الوثائق
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
import { Global, Module } from '@nestjs/common';
|
||||
import { I18nService } from './i18n.service';
|
||||
|
||||
@Global()
|
||||
@Module({
|
||||
providers: [I18nService],
|
||||
exports: [I18nService],
|
||||
})
|
||||
export class I18nModule {}
|
||||
@@ -0,0 +1,32 @@
|
||||
import { I18nService } from './i18n.service';
|
||||
|
||||
describe('I18nService', () => {
|
||||
const i18n = new I18nService();
|
||||
|
||||
it('يترجم للعربية افتراضياً ويستبدل المتغيّرات', () => {
|
||||
const m = i18n.t(null, 'trip.assigned', { driverName: 'أحمد' });
|
||||
expect(m.title).toBe('تم قبول رحلتك');
|
||||
expect(m.body).toBe('سائقك أحمد في الطريق إليك');
|
||||
});
|
||||
|
||||
it('يحترم لغة المستخدم الإنجليزية', () => {
|
||||
const m = i18n.t('en', 'trip.driver_arrived');
|
||||
expect(m.title).toBe('Driver has arrived');
|
||||
});
|
||||
|
||||
it('يطبّع وسوم اللغة المركّبة (ar-JO, EN_us)', () => {
|
||||
expect(i18n.normalize('ar-JO')).toBe('ar');
|
||||
expect(i18n.normalize('EN_us')).toBe('en');
|
||||
expect(i18n.normalize('fr')).toBe('ar'); // غير مدعومة → الافتراضية
|
||||
expect(i18n.normalize(undefined)).toBe('ar');
|
||||
});
|
||||
|
||||
it('يترك المتغيّر كما هو إذا لم تُمرَّر قيمته', () => {
|
||||
const m = i18n.t('ar', 'trip.completed', { currency: 'JOD' });
|
||||
expect(m.body).toBe('وصلت إلى وجهتك. الأجرة {fare} JOD');
|
||||
});
|
||||
|
||||
it('لا يرمي عند مفتاح مفقود', () => {
|
||||
expect(() => i18n.t('ar', 'does.not.exist')).not.toThrow();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,34 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { CATALOGS, DEFAULT_LANG, Lang, Message, SUPPORTED_LANGS } from './messages';
|
||||
|
||||
/** ترجمة نصوص الإشعارات (docs/17 — A2). العربية هي الافتراض. */
|
||||
@Injectable()
|
||||
export class I18nService {
|
||||
private readonly logger = new Logger('I18n');
|
||||
|
||||
/** يطبّع أي وسم لغة (ar-JO, AR, en_US) إلى لغة مدعومة، وإلا الافتراضية. */
|
||||
normalize(lang?: string | null): Lang {
|
||||
const base = (lang ?? '').toLowerCase().split(/[-_]/)[0];
|
||||
return (SUPPORTED_LANGS as string[]).includes(base) ? (base as Lang) : DEFAULT_LANG;
|
||||
}
|
||||
|
||||
/** يترجم مفتاحاً مع استبدال المتغيّرات. المفتاح المفقود يرجع كما هو (لا يكسر الإرسال). */
|
||||
t(lang: string | null | undefined, key: string, params: Record<string, any> = {}): Message {
|
||||
const l = this.normalize(lang);
|
||||
const msg = CATALOGS[l][key] ?? CATALOGS[DEFAULT_LANG][key];
|
||||
if (!msg) {
|
||||
this.logger.warn(`missing i18n key "${key}"`);
|
||||
return { title: key, body: '' };
|
||||
}
|
||||
return {
|
||||
title: this.interpolate(msg.title, params),
|
||||
body: this.interpolate(msg.body, params),
|
||||
};
|
||||
}
|
||||
|
||||
private interpolate(tpl: string, params: Record<string, any>): string {
|
||||
return tpl.replace(/\{(\w+)\}/g, (match, name) =>
|
||||
params[name] == null ? match : String(params[name]),
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
/**
|
||||
* نصوص الإشعارات لكل لغة. المفتاح يُشتق من نوع الحدث (راجع docs/17 — A2).
|
||||
* كل مدخل: عنوان + نص. المتغيّرات بصيغة {name}.
|
||||
*/
|
||||
export type Lang = 'ar' | 'en';
|
||||
|
||||
export const DEFAULT_LANG: Lang = 'ar';
|
||||
export const SUPPORTED_LANGS: Lang[] = ['ar', 'en'];
|
||||
|
||||
export interface Message {
|
||||
title: string;
|
||||
body: string;
|
||||
}
|
||||
|
||||
type Catalog = Record<string, Message>;
|
||||
|
||||
const ar: Catalog = {
|
||||
'trip.assigned': { title: 'تم قبول رحلتك', body: 'سائقك {driverName} في الطريق إليك' },
|
||||
'trip.driver_arriving': { title: 'السائق في الطريق', body: 'سائقك متوجّه إلى نقطة الانطلاق' },
|
||||
'trip.driver_arrived': { title: 'وصل السائق', body: 'سائقك ينتظرك في نقطة الانطلاق' },
|
||||
'trip.in_progress': { title: 'بدأت الرحلة', body: 'رحلتك جارية الآن — رحلة موفقة' },
|
||||
'trip.completed': { title: 'انتهت الرحلة', body: 'وصلت إلى وجهتك. الأجرة {fare} {currency}' },
|
||||
'trip.paid': { title: 'تم الدفع', body: 'تم استلام مبلغ {fare} {currency}' },
|
||||
'trip.cancelled': { title: 'أُلغيت الرحلة', body: 'تم إلغاء الرحلة' },
|
||||
'trip.cancelled_fee': { title: 'أُلغيت الرحلة', body: 'تم إلغاء الرحلة ورسم الإلغاء {fee} {currency}' },
|
||||
'trip.no_drivers': { title: 'لا يوجد سائقون', body: 'لم نجد سائقاً متاحاً قريباً منك. حاول مجدداً' },
|
||||
'trip.expired': { title: 'انتهت مهلة الطلب', body: 'انتهت مهلة طلب الرحلة دون قبول' },
|
||||
'trip.offer': { title: 'طلب رحلة جديد', body: 'راكب على بعد {distanceKm} كم — الأجرة {fare} {currency}' },
|
||||
'trip.offer_taken': { title: 'الرحلة لم تعد متاحة', body: 'قبِل الطلبَ سائق آخر' },
|
||||
'chat.message': { title: 'رسالة جديدة', body: '{preview}' },
|
||||
};
|
||||
|
||||
const en: Catalog = {
|
||||
'trip.assigned': { title: 'Your ride is accepted', body: 'Your driver {driverName} is on the way' },
|
||||
'trip.driver_arriving': { title: 'Driver on the way', body: 'Your driver is heading to the pickup point' },
|
||||
'trip.driver_arrived': { title: 'Driver has arrived', body: 'Your driver is waiting at the pickup point' },
|
||||
'trip.in_progress': { title: 'Trip started', body: 'Your trip is now in progress — have a good ride' },
|
||||
'trip.completed': { title: 'Trip finished', body: 'You have arrived. Fare {fare} {currency}' },
|
||||
'trip.paid': { title: 'Payment received', body: 'Payment of {fare} {currency} received' },
|
||||
'trip.cancelled': { title: 'Trip cancelled', body: 'The trip was cancelled' },
|
||||
'trip.cancelled_fee': { title: 'Trip cancelled', body: 'The trip was cancelled with a {fee} {currency} cancellation fee' },
|
||||
'trip.no_drivers': { title: 'No drivers', body: 'No driver available nearby. Please try again' },
|
||||
'trip.expired': { title: 'Request expired', body: 'The ride request expired without being accepted' },
|
||||
'trip.offer': { title: 'New ride request', body: 'Rider {distanceKm} km away — fare {fare} {currency}' },
|
||||
'trip.offer_taken': { title: 'Ride no longer available', body: 'Another driver accepted the request' },
|
||||
'chat.message': { title: 'New message', body: '{preview}' },
|
||||
};
|
||||
|
||||
export const CATALOGS: Record<Lang, Catalog> = { ar, en };
|
||||
@@ -0,0 +1,12 @@
|
||||
import { MigrationInterface, QueryRunner } from 'typeorm';
|
||||
|
||||
/** لغة المستخدم — تُستخدم لترجمة الإشعارات (docs/17 — A2). */
|
||||
export class UserLanguage1721800000000 implements MigrationInterface {
|
||||
public async up(q: QueryRunner): Promise<void> {
|
||||
await q.query(`ALTER TABLE tripz_users ADD COLUMN IF NOT EXISTS language varchar NOT NULL DEFAULT 'ar'`);
|
||||
}
|
||||
|
||||
public async down(q: QueryRunner): Promise<void> {
|
||||
await q.query(`ALTER TABLE tripz_users DROP COLUMN IF EXISTS language`);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
import { MigrationInterface, QueryRunner } from 'typeorm';
|
||||
|
||||
/**
|
||||
* دفتر المحفظة (docs/17 — I1):
|
||||
* - `balance_after` لتدقيق كل حركة ومطابقتها بالرصيد.
|
||||
* - قيد `balance >= 0` — شبكة أمان أخيرة على مستوى القاعدة: أي مسار كتابة
|
||||
* يتجاوز الخصم الذرّي سيفشل بدل أن يُنتج رصيداً سالباً بصمت.
|
||||
*/
|
||||
export class WalletLedger1721810000000 implements MigrationInterface {
|
||||
public async up(q: QueryRunner): Promise<void> {
|
||||
await q.query(
|
||||
`ALTER TABLE tripz_wallet_txns ADD COLUMN IF NOT EXISTS balance_after numeric(12,3)`,
|
||||
);
|
||||
await q.query(`
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE tripz_wallets ADD CONSTRAINT tripz_wallets_balance_non_negative CHECK (balance >= 0);
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
`);
|
||||
}
|
||||
|
||||
public async down(q: QueryRunner): Promise<void> {
|
||||
await q.query(
|
||||
`ALTER TABLE tripz_wallets DROP CONSTRAINT IF EXISTS tripz_wallets_balance_non_negative`,
|
||||
);
|
||||
await q.query(`ALTER TABLE tripz_wallet_txns DROP COLUMN IF EXISTS balance_after`);
|
||||
}
|
||||
}
|
||||
@@ -3,10 +3,11 @@ import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
import { DeviceToken } from './entities/device-token.entity';
|
||||
import { NotificationsService } from './notifications.service';
|
||||
import { NotificationsController } from './notifications.controller';
|
||||
import { UsersModule } from '../users/users.module';
|
||||
|
||||
@Global()
|
||||
@Module({
|
||||
imports: [TypeOrmModule.forFeature([DeviceToken])],
|
||||
imports: [TypeOrmModule.forFeature([DeviceToken]), UsersModule],
|
||||
controllers: [NotificationsController],
|
||||
providers: [NotificationsService],
|
||||
exports: [NotificationsService],
|
||||
|
||||
@@ -1,11 +1,19 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { In, Repository } from 'typeorm';
|
||||
import { DeviceToken } from './entities/device-token.entity';
|
||||
import { I18nService } from '../../common/i18n/i18n.service';
|
||||
import { UsersService } from '../users/users.service';
|
||||
|
||||
export interface PushOptions {
|
||||
/** رسالة بيانات فقط — لا يعرضها النظام، يتولّاها التطبيق (overlay أندرويد، docs/17 A7). */
|
||||
dataOnly?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* إشعارات FCM. تسجيل التوكنات + إرسال أفضل جهد (best-effort).
|
||||
* النصوص تُترجَم حسب لغة المستخدم (docs/17 — A2).
|
||||
* بلا FCM_SERVER_KEY: يُسجَّل فقط دون إرسال.
|
||||
*/
|
||||
@Injectable()
|
||||
@@ -15,6 +23,8 @@ export class NotificationsService {
|
||||
constructor(
|
||||
@InjectRepository(DeviceToken) private readonly tokens: Repository<DeviceToken>,
|
||||
private readonly config: ConfigService,
|
||||
private readonly i18n: I18nService,
|
||||
private readonly users: UsersService,
|
||||
) {}
|
||||
|
||||
async register(tenantId: string, userId: string, token: string, platform = 'android') {
|
||||
@@ -30,30 +40,111 @@ export class NotificationsService {
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* الإرسال المترجَم — الواجهة المفضّلة. `key` مفتاح من كتالوج الترجمة
|
||||
* و`params` متغيّراته؛ `data` حمولة التطبيق (tripId, type, …).
|
||||
*/
|
||||
async sendLocalized(
|
||||
tenantId: string,
|
||||
userId: string,
|
||||
key: string,
|
||||
params: Record<string, any> = {},
|
||||
data: Record<string, any> = {},
|
||||
opts: PushOptions = {},
|
||||
) {
|
||||
const lang = await this.users.getLanguage(tenantId, userId);
|
||||
const { title, body } = this.i18n.t(lang, key, params);
|
||||
return this.sendToUser(tenantId, userId, title, body, data, opts);
|
||||
}
|
||||
|
||||
/** نفس الشيء لعدة مستخدمين — يُستخدم لبث «الرحلة لم تعد متاحة» (A5). */
|
||||
async sendLocalizedToMany(
|
||||
tenantId: string,
|
||||
userIds: string[],
|
||||
key: string,
|
||||
params: Record<string, any> = {},
|
||||
data: Record<string, any> = {},
|
||||
opts: PushOptions = {},
|
||||
) {
|
||||
await Promise.all(
|
||||
userIds.map((id) =>
|
||||
this.sendLocalized(tenantId, id, key, params, data, opts).catch((e: any) =>
|
||||
this.logger.warn(`push to ${id} failed: ${e?.message}`),
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/** إرسال بنص جاهز (غير مترجَم) — للحالات الخاصة فقط. */
|
||||
async sendToUser(
|
||||
tenantId: string,
|
||||
userId: string,
|
||||
title: string,
|
||||
body: string,
|
||||
data: Record<string, any> = {},
|
||||
opts: PushOptions = {},
|
||||
) {
|
||||
const rows = await this.tokens.find({ where: { tenant_id: tenantId, user_id: userId } });
|
||||
return this.dispatch(rows, title, body, data, opts);
|
||||
}
|
||||
|
||||
// ---- داخلي ----
|
||||
|
||||
private async dispatch(
|
||||
rows: DeviceToken[],
|
||||
title: string,
|
||||
body: string,
|
||||
data: Record<string, any>,
|
||||
opts: PushOptions,
|
||||
) {
|
||||
const key = this.config.get<string>('fcm.serverKey');
|
||||
const rows = await this.tokens.find({ where: { tenant_id: tenantId, user_id: userId } });
|
||||
if (!key || rows.length === 0) {
|
||||
this.logger.debug(`push skip (key=${!!key} tokens=${rows.length}) "${title}"`);
|
||||
return;
|
||||
}
|
||||
const endpoint = this.config.get<string>('fcm.endpoint')!;
|
||||
for (const r of rows) {
|
||||
try {
|
||||
await fetch(endpoint, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', Authorization: `key=${key}` },
|
||||
body: JSON.stringify({ to: r.token, notification: { title, body }, data }),
|
||||
});
|
||||
} catch (e: any) {
|
||||
this.logger.warn(`push failed: ${e?.message}`);
|
||||
}
|
||||
// FCM يقبل نصوصاً فقط في data — والـoverlay يقرأ العنوان/النص من هنا.
|
||||
const payload = this.stringifyData({ ...data, title, body });
|
||||
|
||||
const stale: string[] = [];
|
||||
await Promise.all(
|
||||
rows.map(async (r) => {
|
||||
try {
|
||||
const res = await fetch(endpoint, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', Authorization: `key=${key}` },
|
||||
body: JSON.stringify({
|
||||
to: r.token,
|
||||
// أولوية عالية: يوقظ التطبيق في الخلفية (شرط الـoverlay).
|
||||
priority: 'high',
|
||||
...(opts.dataOnly ? {} : { notification: { title, body } }),
|
||||
data: payload,
|
||||
}),
|
||||
});
|
||||
const json: any = await res.json().catch(() => null);
|
||||
if (json?.results?.[0]?.error === 'NotRegistered' ||
|
||||
json?.results?.[0]?.error === 'InvalidRegistration') {
|
||||
stale.push(r.token);
|
||||
}
|
||||
} catch (e: any) {
|
||||
this.logger.warn(`push failed: ${e?.message}`);
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
// توكن ميت = إرسال ضائع لكل رحلة لاحقة؛ نحذفه فور ما يخبرنا FCM.
|
||||
if (stale.length > 0) {
|
||||
await this.tokens.delete({ token: In(stale) });
|
||||
this.logger.debug(`removed ${stale.length} stale token(s)`);
|
||||
}
|
||||
}
|
||||
|
||||
private stringifyData(data: Record<string, any>): Record<string, string> {
|
||||
const out: Record<string, string> = {};
|
||||
for (const [k, v] of Object.entries(data)) {
|
||||
if (v == null) continue;
|
||||
out[k] = typeof v === 'object' ? JSON.stringify(v) : String(v);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
import RedisMock from 'ioredis-mock';
|
||||
import { TripStateService } from './trip-state.service';
|
||||
import { Trip } from './entities/trip.entity';
|
||||
|
||||
const TENANT = 't1';
|
||||
|
||||
function makeTrip(overrides: Partial<Trip> = {}): Trip {
|
||||
return {
|
||||
id: 'trip-1',
|
||||
tenant_id: TENANT,
|
||||
rider_id: 'rider-1',
|
||||
driver_id: null,
|
||||
service_class: 'economy',
|
||||
is_round_trip: false,
|
||||
city: 'amman',
|
||||
origin_lat: 31.95,
|
||||
origin_lng: 35.91,
|
||||
dest_lat: 31.98,
|
||||
dest_lng: 35.87,
|
||||
status: 'searching',
|
||||
distance_km: 4.2,
|
||||
duration_min: 11,
|
||||
quoted_fare: 3.75,
|
||||
currency: 'JOD',
|
||||
payment_method: 'cash',
|
||||
assigned_at: null,
|
||||
...overrides,
|
||||
} as Trip;
|
||||
}
|
||||
|
||||
describe('TripStateService', () => {
|
||||
let redis: any;
|
||||
let state: TripStateService;
|
||||
|
||||
beforeEach(() => {
|
||||
redis = new RedisMock({ keyPrefix: 'tripz:' });
|
||||
state = new TripStateService(redis);
|
||||
});
|
||||
|
||||
it('يحفظ الحالة ويقرأها بأنواعها الصحيحة', async () => {
|
||||
await state.save(makeTrip(), 'driver-user-1');
|
||||
const s = await state.get(TENANT, 'trip-1');
|
||||
|
||||
expect(s).not.toBeNull();
|
||||
expect(s!.status).toBe('searching');
|
||||
expect(s!.driver_user_id).toBe('driver-user-1');
|
||||
expect(s!.origin_lat).toBeCloseTo(31.95);
|
||||
expect(s!.quoted_fare).toBe(3.75);
|
||||
expect(s!.assigned_at).toBeNull();
|
||||
});
|
||||
|
||||
it('يرجع null لرحلة غير موجودة (فيرجع المُنادي للقاعدة)', async () => {
|
||||
expect(await state.get(TENANT, 'ghost')).toBeNull();
|
||||
});
|
||||
|
||||
describe('claim — القبول الذرّي', () => {
|
||||
it('أول سائق يفوز والثاني يخسر', async () => {
|
||||
await state.save(makeTrip());
|
||||
const at = new Date();
|
||||
|
||||
expect(await state.claim(TENANT, 'trip-1', 'd1', 'u1', at)).toBe('won');
|
||||
expect(await state.claim(TENANT, 'trip-1', 'd2', 'u2', at)).toBe('lost');
|
||||
|
||||
const s = await state.get(TENANT, 'trip-1');
|
||||
expect(s!.status).toBe('assigned');
|
||||
expect(s!.driver_id).toBe('d1');
|
||||
expect(s!.driver_user_id).toBe('u1');
|
||||
expect(s!.assigned_at?.toISOString()).toBe(at.toISOString());
|
||||
});
|
||||
|
||||
it('فائز واحد فقط عند تزامن عدة سائقين', async () => {
|
||||
await state.save(makeTrip());
|
||||
const at = new Date();
|
||||
|
||||
const results = await Promise.all(
|
||||
Array.from({ length: 20 }, (_, i) =>
|
||||
state.claim(TENANT, 'trip-1', `d${i}`, `u${i}`, at),
|
||||
),
|
||||
);
|
||||
|
||||
expect(results.filter((r) => r === 'won')).toHaveLength(1);
|
||||
expect(results.filter((r) => r === 'lost')).toHaveLength(19);
|
||||
});
|
||||
|
||||
it('يرجع unknown عند غياب المفتاح ليحسم الأمر على القاعدة', async () => {
|
||||
expect(await state.claim(TENANT, 'gone', 'd1', 'u1', new Date())).toBe('unknown');
|
||||
});
|
||||
});
|
||||
|
||||
it('setStatus لا يُنشئ مفتاحاً لرحلة غير جارية', async () => {
|
||||
await state.setStatus(TENANT, 'ghost', 'completed');
|
||||
expect(await state.get(TENANT, 'ghost')).toBeNull();
|
||||
});
|
||||
|
||||
it('clear يمسح الحالة والعروض معاً', async () => {
|
||||
await state.save(makeTrip());
|
||||
await state.addOffers(TENANT, 'trip-1', ['u1', 'u2']);
|
||||
await state.clear(TENANT, 'trip-1');
|
||||
|
||||
expect(await state.get(TENANT, 'trip-1')).toBeNull();
|
||||
expect(await state.takeLosingOffers(TENANT, 'trip-1', 'u1')).toEqual([]);
|
||||
});
|
||||
|
||||
it('takeLosingOffers يستثني الفائز ويستهلك المجموعة مرة واحدة', async () => {
|
||||
await state.addOffers(TENANT, 'trip-1', ['u1', 'u2', 'u3']);
|
||||
|
||||
const losers = await state.takeLosingOffers(TENANT, 'trip-1', 'u2');
|
||||
expect(losers.sort()).toEqual(['u1', 'u3']);
|
||||
// الاستدعاء الثاني فارغ — لا إشعارات مكرّرة لبقية السائقين
|
||||
expect(await state.takeLosingOffers(TENANT, 'trip-1', 'u2')).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,173 @@
|
||||
import { Inject, Injectable, Logger } from '@nestjs/common';
|
||||
import Redis from 'ioredis';
|
||||
import { REDIS } from '../../common/redis/redis.module';
|
||||
import { Trip, TripStatus } from './entities/trip.entity';
|
||||
|
||||
/** حالة الرحلة الجارية كما تعيش في Redis. القيم كلها نصوص (hash). */
|
||||
export interface TripState {
|
||||
tripId: string;
|
||||
status: TripStatus;
|
||||
rider_id: string;
|
||||
driver_id: string | null;
|
||||
/** user_id للسائق — مخزَّن ليُبَثّ له بلا استعلام drivers في كل انتقال. */
|
||||
driver_user_id: string | null;
|
||||
service_class: string;
|
||||
origin_lat: number;
|
||||
origin_lng: number;
|
||||
dest_lat: number;
|
||||
dest_lng: number;
|
||||
distance_km: number | null;
|
||||
quoted_fare: number | null;
|
||||
currency: string | null;
|
||||
assigned_at: Date | null;
|
||||
}
|
||||
|
||||
/** نتيجة محاولة القبول الذرّي. */
|
||||
export type ClaimResult = 'won' | 'lost' | 'unknown';
|
||||
|
||||
const ACTIVE_TTL_SEC = 6 * 60 * 60; // رحلة جارية لا تتجاوز 6 ساعات واقعياً
|
||||
const OFFERS_TTL_SEC = 15 * 60;
|
||||
|
||||
/**
|
||||
* حالة الرحلة الجارية في Redis (docs/17 — A3).
|
||||
* القاعدة تبقى مصدر الحقيقة الدائمة؛ هذه الطبقة تلغي القراءات المتكررة
|
||||
* في كل انتقال حالة. أي مفتاح مفقود = رجوع للقاعدة (لا يُعتمد عليه كحقيقة).
|
||||
*/
|
||||
@Injectable()
|
||||
export class TripStateService {
|
||||
private readonly logger = new Logger('TripState');
|
||||
|
||||
constructor(@Inject(REDIS) private readonly redis: Redis) {}
|
||||
|
||||
private key(tenantId: string, tripId: string) {
|
||||
return `trip:${tenantId}:${tripId}`;
|
||||
}
|
||||
|
||||
private offersKey(tenantId: string, tripId: string) {
|
||||
return `trip:offers:${tenantId}:${tripId}`;
|
||||
}
|
||||
|
||||
/** يكتب لقطة الرحلة عند الإنشاء أو بعد أي تغيير دائم. */
|
||||
async save(trip: Trip, driverUserId: string | null = null): Promise<void> {
|
||||
const k = this.key(trip.tenant_id, trip.id);
|
||||
await this.redis
|
||||
.multi()
|
||||
.hset(k, {
|
||||
tripId: trip.id,
|
||||
status: trip.status,
|
||||
rider_id: trip.rider_id,
|
||||
driver_id: trip.driver_id ?? '',
|
||||
driver_user_id: driverUserId ?? '',
|
||||
service_class: trip.service_class,
|
||||
origin_lat: String(trip.origin_lat),
|
||||
origin_lng: String(trip.origin_lng),
|
||||
dest_lat: String(trip.dest_lat),
|
||||
dest_lng: String(trip.dest_lng),
|
||||
distance_km: trip.distance_km == null ? '' : String(trip.distance_km),
|
||||
quoted_fare: trip.quoted_fare == null ? '' : String(trip.quoted_fare),
|
||||
currency: trip.currency ?? '',
|
||||
assigned_at: trip.assigned_at ? trip.assigned_at.toISOString() : '',
|
||||
})
|
||||
.expire(k, ACTIVE_TTL_SEC)
|
||||
.exec();
|
||||
}
|
||||
|
||||
async get(tenantId: string, tripId: string): Promise<TripState | null> {
|
||||
const h = await this.redis.hgetall(this.key(tenantId, tripId));
|
||||
if (!h || !h.status) return null;
|
||||
return {
|
||||
tripId,
|
||||
status: h.status as TripStatus,
|
||||
rider_id: h.rider_id,
|
||||
driver_id: h.driver_id || null,
|
||||
driver_user_id: h.driver_user_id || null,
|
||||
service_class: h.service_class,
|
||||
origin_lat: Number(h.origin_lat),
|
||||
origin_lng: Number(h.origin_lng),
|
||||
dest_lat: Number(h.dest_lat),
|
||||
dest_lng: Number(h.dest_lng),
|
||||
distance_km: h.distance_km ? Number(h.distance_km) : null,
|
||||
quoted_fare: h.quoted_fare ? Number(h.quoted_fare) : null,
|
||||
currency: h.currency || null,
|
||||
assigned_at: h.assigned_at ? new Date(h.assigned_at) : null,
|
||||
};
|
||||
}
|
||||
|
||||
async setStatus(tenantId: string, tripId: string, status: TripStatus): Promise<void> {
|
||||
const k = this.key(tenantId, tripId);
|
||||
// لا نُنشئ المفتاح إن كان مفقوداً — الحالة الجارية فقط تُحدَّث هنا.
|
||||
if ((await this.redis.exists(k)) === 0) return;
|
||||
await this.redis.multi().hset(k, 'status', status).expire(k, ACTIVE_TTL_SEC).exec();
|
||||
}
|
||||
|
||||
/** الرحلة انتهت (completed/paid/cancelled/…): لا داعي لإبقائها في الذاكرة. */
|
||||
async clear(tenantId: string, tripId: string): Promise<void> {
|
||||
await this.redis.del(this.key(tenantId, tripId), this.offersKey(tenantId, tripId));
|
||||
}
|
||||
|
||||
/**
|
||||
* يُسقط اللقطة وحدها (بلا العروض) عندما تتبيّن مخالفتها للقاعدة —
|
||||
* الاستدعاء التالي يعيد بناءها من الحقيقة الدائمة.
|
||||
*/
|
||||
async invalidate(tenantId: string, tripId: string): Promise<void> {
|
||||
await this.redis.del(this.key(tenantId, tripId));
|
||||
}
|
||||
|
||||
/**
|
||||
* قبول ذرّي (docs/17 — A4): يضبط الحالة إلى assigned فقط إذا كانت searching.
|
||||
* السائق الأول يربح؛ الباقي يحصلون على 'lost' بلا لمس القاعدة.
|
||||
* 'unknown' = المفتاح مفقود (انتهت صلاحيته/إعادة تشغيل) → على المُنادي أن
|
||||
* يحسم الأمر عبر UPDATE شرطي على القاعدة.
|
||||
*/
|
||||
async claim(
|
||||
tenantId: string,
|
||||
tripId: string,
|
||||
driverId: string,
|
||||
driverUserId: string,
|
||||
assignedAt: Date,
|
||||
): Promise<ClaimResult> {
|
||||
const script = `
|
||||
local status = redis.call('HGET', KEYS[1], 'status')
|
||||
if not status then return 'unknown' end
|
||||
if status ~= 'searching' then return 'lost' end
|
||||
redis.call('HSET', KEYS[1], 'status', 'assigned',
|
||||
'driver_id', ARGV[1], 'driver_user_id', ARGV[2],
|
||||
'assigned_at', ARGV[3])
|
||||
redis.call('EXPIRE', KEYS[1], ARGV[4])
|
||||
return 'won'
|
||||
`;
|
||||
try {
|
||||
const res = await this.redis.eval(
|
||||
script,
|
||||
1,
|
||||
this.key(tenantId, tripId),
|
||||
driverId,
|
||||
driverUserId,
|
||||
assignedAt.toISOString(),
|
||||
String(ACTIVE_TTL_SEC),
|
||||
);
|
||||
return res as ClaimResult;
|
||||
} catch (e: any) {
|
||||
// فشل Redis لا يوقف الرحلة — القاعدة تحسم عبر الـUPDATE الشرطي.
|
||||
this.logger.warn(`claim failed, falling back to DB: ${e?.message}`);
|
||||
return 'unknown';
|
||||
}
|
||||
}
|
||||
|
||||
// ---- تتبّع العروض (A5) ----
|
||||
|
||||
/** يسجّل السائقين الذين عُرضت عليهم الرحلة لإبلاغهم عند فوز غيرهم. */
|
||||
async addOffers(tenantId: string, tripId: string, driverUserIds: string[]): Promise<void> {
|
||||
if (driverUserIds.length === 0) return;
|
||||
const k = this.offersKey(tenantId, tripId);
|
||||
await this.redis.multi().sadd(k, ...driverUserIds).expire(k, OFFERS_TTL_SEC).exec();
|
||||
}
|
||||
|
||||
/** يرجع بقية السائقين المعروض عليهم باستثناء الفائز. */
|
||||
async takeLosingOffers(tenantId: string, tripId: string, winnerUserId: string): Promise<string[]> {
|
||||
const k = this.offersKey(tenantId, tripId);
|
||||
const all = await this.redis.smembers(k);
|
||||
await this.redis.del(k);
|
||||
return all.filter((id) => id !== winnerUserId);
|
||||
}
|
||||
}
|
||||
@@ -23,6 +23,12 @@ export class TripsController {
|
||||
return this.trips.listForRider(user.tenantId, user.userId);
|
||||
}
|
||||
|
||||
// السائق يسحب الطلبات المتاحة قربه (بديل/مكمّل للعرض المباشر)
|
||||
@Get('available')
|
||||
available(@CurrentUser() user: AuthUser) {
|
||||
return this.trips.availableForDriver(user.tenantId, user.userId);
|
||||
}
|
||||
|
||||
@Get(':id')
|
||||
get(@CurrentUser() user: AuthUser, @Param('id') id: string) {
|
||||
return this.trips.get(user.tenantId, id);
|
||||
|
||||
@@ -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 { TripStateService } from './trip-state.service';
|
||||
import { TripsController } from './trips.controller';
|
||||
import { MapsModule } from '../maps/maps.module';
|
||||
import { TariffModule } from '../tariff/tariff.module';
|
||||
@@ -11,6 +12,7 @@ import { DriversModule } from '../drivers/drivers.module';
|
||||
import { RealtimeModule } from '../../realtime/realtime.module';
|
||||
import { FraudModule } from '../fraud/fraud.module';
|
||||
import { WalletModule } from '../wallet/wallet.module';
|
||||
import { UsersModule } from '../users/users.module';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
@@ -22,9 +24,10 @@ import { WalletModule } from '../wallet/wallet.module';
|
||||
RealtimeModule,
|
||||
FraudModule,
|
||||
WalletModule,
|
||||
UsersModule,
|
||||
],
|
||||
controllers: [TripsController],
|
||||
providers: [TripsService],
|
||||
providers: [TripsService, TripStateService],
|
||||
exports: [TripsService],
|
||||
})
|
||||
export class TripsModule {}
|
||||
|
||||
@@ -17,6 +17,8 @@ import { RealtimeGateway } from '../../realtime/realtime.gateway';
|
||||
import { FraudService } from '../fraud/fraud.service';
|
||||
import { WalletService } from '../wallet/wallet.service';
|
||||
import { NotificationsService } from '../notifications/notifications.service';
|
||||
import { TripState, TripStateService } from './trip-state.service';
|
||||
import { UsersService } from '../users/users.service';
|
||||
|
||||
/** رسوم الإلغاء حسب مرحلة الرحلة (docs/04). */
|
||||
const CANCEL_FEE_BY_STAGE: Partial<Record<TripStatus, number>> = {
|
||||
@@ -40,6 +42,9 @@ const TRANSITIONS: Record<TripStatus, TripStatus[]> = {
|
||||
expired: [],
|
||||
};
|
||||
|
||||
/** الحالات النهائية — تُمسح من Redis لأنها لم تعد "جارية". */
|
||||
const TERMINAL: TripStatus[] = ['paid', 'cancelled', 'no_drivers', 'expired'];
|
||||
|
||||
export interface RequestTripDto {
|
||||
origin: { lat: number; lng: number };
|
||||
destination: { lat: number; lng: number };
|
||||
@@ -64,6 +69,8 @@ export class TripsService {
|
||||
private readonly fraud: FraudService,
|
||||
private readonly wallet: WalletService,
|
||||
private readonly notifications: NotificationsService,
|
||||
private readonly state: TripStateService,
|
||||
private readonly users: UsersService,
|
||||
) {}
|
||||
|
||||
get(tenantId: string, id: string): Promise<Trip | null> {
|
||||
@@ -137,61 +144,59 @@ export class TripsService {
|
||||
payment_method: (dto as any).payment_method ?? 'cash',
|
||||
});
|
||||
trip = await this.trips.save(trip);
|
||||
await this.state.save(trip);
|
||||
await this.recordEvent(trip, null, 'searching', 'rider');
|
||||
|
||||
// مطابقة وبث عروض للسائقين القريبين من نفس فئة الخدمة
|
||||
const nearby = await this.matching.findNearby(
|
||||
tenantId,
|
||||
serviceClass,
|
||||
dto.origin.lat,
|
||||
dto.origin.lng,
|
||||
);
|
||||
for (const n of nearby) {
|
||||
const driver = await this.drivers.findById(tenantId, n.driverId);
|
||||
if (driver && driver.verification_status === 'approved') {
|
||||
this.gateway.offerToDriver(tenantId, driver.user_id, {
|
||||
tripId: trip.id,
|
||||
pickup: { lat: trip.origin_lat, lng: trip.origin_lng },
|
||||
dropoff: { lat: trip.dest_lat, lng: trip.dest_lng },
|
||||
distanceKm: n.distanceKm,
|
||||
quotedFare,
|
||||
currency,
|
||||
});
|
||||
}
|
||||
}
|
||||
const offered = await this.offerToNearbyDrivers(tenantId, trip, dto.origin);
|
||||
this.gateway.dispatch(tenantId, 'trip:new', { tripId: trip.id });
|
||||
|
||||
return { trip, offeredDrivers: nearby.length };
|
||||
return { trip, offeredDrivers: offered };
|
||||
}
|
||||
|
||||
/** قبول سائق للرحلة (أول قبول يفوز). */
|
||||
/** قبول سائق للرحلة (أول قبول يفوز — docs/17 A4). */
|
||||
async accept(tenantId: string, tripId: string, driverUserId: string) {
|
||||
const trip = await this.getOr404(tenantId, tripId);
|
||||
if (trip.status !== 'searching') {
|
||||
throw new BadRequestException('Trip is no longer available');
|
||||
}
|
||||
const driver = await this.drivers.findByUser(tenantId, driverUserId);
|
||||
if (!driver) throw new ForbiddenException('Not a driver');
|
||||
if (driver.verification_status !== 'approved') {
|
||||
throw new ForbiddenException('Driver not approved');
|
||||
}
|
||||
|
||||
trip.driver_id = driver.id;
|
||||
trip.assigned_at = new Date();
|
||||
await this.applyTransition(trip, 'assigned', 'driver');
|
||||
const assignedAt = new Date();
|
||||
// 1) حسم سريع في Redis: الخاسر يخرج بلا لمس القاعدة إطلاقاً.
|
||||
const claim = await this.state.claim(tenantId, tripId, driver.id, driverUserId, assignedAt);
|
||||
if (claim === 'lost') throw new BadRequestException('Trip is no longer available');
|
||||
|
||||
this.gateway.tripUpdate(tenantId, trip.rider_id, {
|
||||
tripId: trip.id,
|
||||
status: 'assigned',
|
||||
driverId: driver.id,
|
||||
});
|
||||
await this.notifications.sendToUser(
|
||||
tenantId,
|
||||
trip.rider_id,
|
||||
'تم قبول رحلتك',
|
||||
'سائقك في الطريق إليك',
|
||||
{ tripId: trip.id, type: 'trip_assigned' },
|
||||
// 2) القاعدة هي الحَكَم النهائي: UPDATE شرطي — لا يمرّ إلا إذا كانت searching.
|
||||
const res = await this.trips.update(
|
||||
{ tenant_id: tenantId, id: tripId, status: 'searching' },
|
||||
{ status: 'assigned', driver_id: driver.id, assigned_at: assignedAt },
|
||||
);
|
||||
if (!res.affected) {
|
||||
// Redis سمح والقاعدة رفضت = المفتاح كان متأخّراً عن الحقيقة. نمسحه بدل
|
||||
// إعادة كتابته: هنا لا نعرف user_id للسائق الفائز، وحفظ حالة ناقصة يجعل
|
||||
// snapshot() يثق بها فيحرم الفائز من إشعارات بقية الرحلة. والعروض تبقى
|
||||
// كما هي — الفائز وحده من يستهلكها ليُبلغ بقية السائقين.
|
||||
await this.state.invalidate(tenantId, tripId);
|
||||
throw new BadRequestException('Trip is no longer available');
|
||||
}
|
||||
|
||||
const trip = await this.getOr404(tenantId, tripId);
|
||||
await this.recordEvent(trip, 'searching', 'assigned', 'driver');
|
||||
const driverUser = await this.users.findById(tenantId, driverUserId);
|
||||
await this.notifyParties(tenantId, trip, 'assigned', driverUserId, {
|
||||
driverName: driverUser?.name ?? '',
|
||||
vehicle: {
|
||||
make: driver.vehicle_make,
|
||||
model: driver.vehicle_model,
|
||||
plate: driver.vehicle_plate,
|
||||
color: driver.vehicle_color,
|
||||
},
|
||||
});
|
||||
|
||||
// 3) بقية السائقين: العرض مات — أزِله من شاشتهم فوراً (A5).
|
||||
await this.cancelLosingOffers(tenantId, tripId, driverUserId);
|
||||
|
||||
return trip;
|
||||
}
|
||||
|
||||
@@ -202,57 +207,71 @@ export class TripsService {
|
||||
toStatus: TripStatus,
|
||||
actor: 'rider' | 'driver',
|
||||
) {
|
||||
const trip = await this.getOr404(tenantId, tripId);
|
||||
const snapshot = await this.snapshot(tenantId, tripId);
|
||||
this.assertTransition(snapshot.status, toStatus);
|
||||
|
||||
const patch: Partial<Trip> = { status: toStatus };
|
||||
|
||||
// كشف احتيال عند نقاط حسّاسة
|
||||
if (toStatus === 'driver_arrived' && trip.driver_id) {
|
||||
const drv = await this.drivers.findById(tenantId, trip.driver_id);
|
||||
if (toStatus === 'driver_arrived' && snapshot.driver_id) {
|
||||
const drv = await this.drivers.findById(tenantId, snapshot.driver_id);
|
||||
if (drv) {
|
||||
await this.fraud.checkArrivedProximity(
|
||||
tenantId,
|
||||
drv.user_id,
|
||||
{ last_lat: drv.last_lat, last_lng: drv.last_lng },
|
||||
{ lat: trip.origin_lat, lng: trip.origin_lng },
|
||||
trip.id,
|
||||
{ lat: snapshot.origin_lat, lng: snapshot.origin_lng },
|
||||
tripId,
|
||||
);
|
||||
}
|
||||
}
|
||||
if (toStatus === 'completed') {
|
||||
trip.completed_at = new Date();
|
||||
if (trip.quoted_fare != null) trip.final_fare = trip.quoted_fare;
|
||||
if (trip.driver_id) {
|
||||
const drv = await this.drivers.findById(tenantId, trip.driver_id);
|
||||
if (drv) {
|
||||
await this.fraud.checkFastCompletion(
|
||||
tenantId,
|
||||
drv.user_id,
|
||||
trip.assigned_at,
|
||||
trip.distance_km == null ? null : Number(trip.distance_km),
|
||||
trip.id,
|
||||
);
|
||||
}
|
||||
patch.completed_at = new Date();
|
||||
if (snapshot.quoted_fare != null) patch.final_fare = snapshot.quoted_fare;
|
||||
if (snapshot.driver_user_id) {
|
||||
await this.fraud.checkFastCompletion(
|
||||
tenantId,
|
||||
snapshot.driver_user_id,
|
||||
snapshot.assigned_at,
|
||||
snapshot.distance_km,
|
||||
tripId,
|
||||
);
|
||||
}
|
||||
}
|
||||
await this.applyTransition(trip, toStatus, actor);
|
||||
|
||||
// UPDATE شرطي على الحالة السابقة: يمنع تسابق تحديثين على نفس الرحلة.
|
||||
const res = await this.trips.update(
|
||||
{ tenant_id: tenantId, id: tripId, status: snapshot.status },
|
||||
patch,
|
||||
);
|
||||
if (!res.affected) {
|
||||
throw new BadRequestException(`Invalid transition ${snapshot.status} -> ${toStatus}`);
|
||||
}
|
||||
|
||||
const trip = await this.getOr404(tenantId, tripId);
|
||||
await this.syncState(trip, snapshot.driver_user_id, toStatus);
|
||||
await this.recordEvent(trip, snapshot.status, toStatus, actor);
|
||||
|
||||
// تسوية المحفظة عند الدفع (payment_method === wallet)
|
||||
if (toStatus === 'paid' && trip.payment_method === 'wallet' && trip.final_fare != null) {
|
||||
const fare = Number(trip.final_fare);
|
||||
const drv = trip.driver_id ? await this.drivers.findById(tenantId, trip.driver_id) : null;
|
||||
try {
|
||||
await this.wallet.debit(tenantId, trip.rider_id, fare, 'trip_fare', trip.id);
|
||||
if (drv) await this.wallet.credit(tenantId, drv.user_id, fare, 'trip_earning', trip.id);
|
||||
if (snapshot.driver_user_id) {
|
||||
await this.wallet.credit(
|
||||
tenantId,
|
||||
snapshot.driver_user_id,
|
||||
fare,
|
||||
'trip_earning',
|
||||
trip.id,
|
||||
);
|
||||
}
|
||||
} catch (e: any) {
|
||||
this.logger.warn(`wallet settle failed: ${e?.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
// بث لطرفَي الرحلة
|
||||
this.gateway.tripUpdate(tenantId, trip.rider_id, { tripId: trip.id, status: toStatus });
|
||||
if (trip.driver_id) {
|
||||
const drv = await this.drivers.findById(tenantId, trip.driver_id);
|
||||
if (drv) this.gateway.tripUpdate(tenantId, drv.user_id, { tripId: trip.id, status: toStatus });
|
||||
}
|
||||
await this.notifyParties(tenantId, trip, toStatus, snapshot.driver_user_id);
|
||||
return trip;
|
||||
}
|
||||
|
||||
@@ -263,36 +282,71 @@ export class TripsService {
|
||||
actor: 'rider' | 'driver',
|
||||
actorUserId: string,
|
||||
) {
|
||||
const trip = await this.getOr404(tenantId, tripId);
|
||||
const snapshot = await this.snapshot(tenantId, tripId);
|
||||
this.assertTransition(snapshot.status, 'cancelled');
|
||||
|
||||
// كشف إساءة الإلغاء (قد يرمي عند الحد الصارم)
|
||||
await this.fraud.recordCancellation(tenantId, actor, actorUserId, trip.id);
|
||||
await this.fraud.recordCancellation(tenantId, actor, actorUserId, tripId);
|
||||
|
||||
const fee = CANCEL_FEE_BY_STAGE[trip.status] ?? 0;
|
||||
trip.cancelled_by = actor;
|
||||
trip.cancel_fee = fee;
|
||||
await this.applyTransition(trip, 'cancelled', actor);
|
||||
const fee = CANCEL_FEE_BY_STAGE[snapshot.status] ?? 0;
|
||||
const res = await this.trips.update(
|
||||
{ tenant_id: tenantId, id: tripId, status: snapshot.status },
|
||||
{ status: 'cancelled', cancelled_by: actor, cancel_fee: fee },
|
||||
);
|
||||
if (!res.affected) throw new BadRequestException('Trip is no longer cancellable');
|
||||
|
||||
// أبلغ الطرف الآخر
|
||||
this.gateway.tripUpdate(tenantId, trip.rider_id, {
|
||||
tripId: trip.id,
|
||||
status: 'cancelled',
|
||||
cancelledBy: actor,
|
||||
cancelFee: fee,
|
||||
});
|
||||
if (trip.driver_id) {
|
||||
const drv = await this.drivers.findById(tenantId, trip.driver_id);
|
||||
if (drv) {
|
||||
this.gateway.tripUpdate(tenantId, drv.user_id, {
|
||||
tripId: trip.id,
|
||||
status: 'cancelled',
|
||||
cancelledBy: actor,
|
||||
});
|
||||
}
|
||||
const trip = await this.getOr404(tenantId, tripId);
|
||||
await this.syncState(trip, snapshot.driver_user_id, 'cancelled');
|
||||
await this.recordEvent(trip, snapshot.status, 'cancelled', actor);
|
||||
|
||||
// الإلغاء وهي searching: العرض معلّق عند سائقين — أزِله.
|
||||
if (snapshot.status === 'searching') {
|
||||
await this.cancelLosingOffers(tenantId, tripId, actorUserId);
|
||||
}
|
||||
|
||||
await this.notifyParties(tenantId, trip, 'cancelled', snapshot.driver_user_id, {
|
||||
fee,
|
||||
currency: trip.currency ?? '',
|
||||
cancelledBy: actor,
|
||||
});
|
||||
return trip;
|
||||
}
|
||||
|
||||
/**
|
||||
* الرحلات المتاحة قرب السائق (docs/17 — A6): يسحبها بنفسه بدل انتظار العرض.
|
||||
* تُقرأ الحالة من القاعدة (searching) ثم تُصفّى بالمسافة.
|
||||
*/
|
||||
async availableForDriver(tenantId: string, driverUserId: string, radiusKm = 5) {
|
||||
const driver = await this.drivers.findByUser(tenantId, driverUserId);
|
||||
if (!driver) throw new ForbiddenException('Not a driver');
|
||||
if (driver.verification_status !== 'approved') {
|
||||
throw new ForbiddenException('Driver not approved');
|
||||
}
|
||||
if (driver.last_lat == null || driver.last_lng == null) return [];
|
||||
|
||||
const open = await this.trips.find({
|
||||
where: {
|
||||
tenant_id: tenantId,
|
||||
status: 'searching',
|
||||
service_class: driver.service_class,
|
||||
},
|
||||
order: { requested_at: 'DESC' },
|
||||
take: 50,
|
||||
});
|
||||
|
||||
return open
|
||||
.map((t) => ({
|
||||
trip: t,
|
||||
distanceKm: haversineKm(driver.last_lat!, driver.last_lng!, t.origin_lat, t.origin_lng),
|
||||
}))
|
||||
.filter((r) => r.distanceKm <= radiusKm)
|
||||
.sort((a, b) => a.distanceKm - b.distanceKm)
|
||||
.map(({ trip, distanceKm }) => ({
|
||||
...this.offerPayload(trip, Number(distanceKm.toFixed(3))),
|
||||
requestedAt: trip.requested_at,
|
||||
}));
|
||||
}
|
||||
|
||||
// ---- داخلي ----
|
||||
|
||||
private async getOr404(tenantId: string, id: string): Promise<Trip> {
|
||||
@@ -301,17 +355,151 @@ export class TripsService {
|
||||
return trip;
|
||||
}
|
||||
|
||||
private async applyTransition(trip: Trip, to: TripStatus, source: string) {
|
||||
const allowed = TRANSITIONS[trip.status] ?? [];
|
||||
if (!allowed.includes(to)) {
|
||||
throw new BadRequestException(
|
||||
`Invalid transition ${trip.status} -> ${to}`,
|
||||
);
|
||||
/**
|
||||
* حالة الرحلة الجارية — من Redis أولاً (docs/17 A3)، ومن القاعدة عند غيابها.
|
||||
* الغرض: إلغاء قراءة الرحلة + قراءة السائق في كل انتقال.
|
||||
*/
|
||||
private async snapshot(tenantId: string, tripId: string): Promise<TripState> {
|
||||
const cached = await this.state.get(tenantId, tripId);
|
||||
if (cached) return cached;
|
||||
|
||||
const trip = await this.getOr404(tenantId, tripId);
|
||||
const drv = trip.driver_id ? await this.drivers.findById(tenantId, trip.driver_id) : null;
|
||||
await this.state.save(trip, drv?.user_id ?? null);
|
||||
return {
|
||||
tripId: trip.id,
|
||||
status: trip.status,
|
||||
rider_id: trip.rider_id,
|
||||
driver_id: trip.driver_id,
|
||||
driver_user_id: drv?.user_id ?? null,
|
||||
service_class: trip.service_class,
|
||||
origin_lat: trip.origin_lat,
|
||||
origin_lng: trip.origin_lng,
|
||||
dest_lat: trip.dest_lat,
|
||||
dest_lng: trip.dest_lng,
|
||||
distance_km: trip.distance_km == null ? null : Number(trip.distance_km),
|
||||
quoted_fare: trip.quoted_fare == null ? null : Number(trip.quoted_fare),
|
||||
currency: trip.currency ?? null,
|
||||
assigned_at: trip.assigned_at,
|
||||
};
|
||||
}
|
||||
|
||||
private assertTransition(from: TripStatus, to: TripStatus) {
|
||||
if (!(TRANSITIONS[from] ?? []).includes(to)) {
|
||||
throw new BadRequestException(`Invalid transition ${from} -> ${to}`);
|
||||
}
|
||||
const from = trip.status;
|
||||
trip.status = to;
|
||||
await this.trips.save(trip);
|
||||
await this.recordEvent(trip, from, to, source);
|
||||
}
|
||||
|
||||
private async syncState(trip: Trip, driverUserId: string | null, to: TripStatus) {
|
||||
if (TERMINAL.includes(to)) {
|
||||
await this.state.clear(trip.tenant_id, trip.id);
|
||||
} else {
|
||||
await this.state.save(trip, driverUserId);
|
||||
}
|
||||
}
|
||||
|
||||
/** حمولة العرض/الإشعار — كل ما يحتاجه الـoverlay ليقرّر بلا نداء إضافي (A7). */
|
||||
private offerPayload(trip: Trip, distanceKm: number) {
|
||||
return {
|
||||
tripId: trip.id,
|
||||
pickup: { lat: trip.origin_lat, lng: trip.origin_lng },
|
||||
dropoff: { lat: trip.dest_lat, lng: trip.dest_lng },
|
||||
distanceKm,
|
||||
tripDistanceKm: trip.distance_km == null ? null : Number(trip.distance_km),
|
||||
tripDurationMin: trip.duration_min == null ? null : Number(trip.duration_min),
|
||||
quotedFare: trip.quoted_fare == null ? null : Number(trip.quoted_fare),
|
||||
currency: trip.currency ?? null,
|
||||
serviceClass: trip.service_class,
|
||||
paymentMethod: trip.payment_method,
|
||||
isRoundTrip: trip.is_round_trip,
|
||||
};
|
||||
}
|
||||
|
||||
/** يبثّ العرض عبر WebSocket + FCM ويسجّل من عُرض عليهم (A1/A5/A7). */
|
||||
private async offerToNearbyDrivers(
|
||||
tenantId: string,
|
||||
trip: Trip,
|
||||
origin: { lat: number; lng: number },
|
||||
): Promise<number> {
|
||||
const nearby = await this.matching.findNearby(
|
||||
tenantId,
|
||||
trip.service_class,
|
||||
origin.lat,
|
||||
origin.lng,
|
||||
);
|
||||
const offeredUserIds: string[] = [];
|
||||
|
||||
for (const n of nearby) {
|
||||
const driver = await this.drivers.findById(tenantId, n.driverId);
|
||||
if (!driver || driver.verification_status !== 'approved') continue;
|
||||
|
||||
const payload = this.offerPayload(trip, n.distanceKm);
|
||||
this.gateway.offerToDriver(tenantId, driver.user_id, payload);
|
||||
offeredUserIds.push(driver.user_id);
|
||||
|
||||
// data-only: التطبيق يرسم الـoverlay بنفسه حتى وهو في الخلفية.
|
||||
await this.notifications
|
||||
.sendLocalized(
|
||||
tenantId,
|
||||
driver.user_id,
|
||||
'trip.offer',
|
||||
{
|
||||
distanceKm: n.distanceKm,
|
||||
fare: trip.quoted_fare ?? '—',
|
||||
currency: trip.currency ?? '',
|
||||
},
|
||||
{ type: 'trip_offer', ...payload },
|
||||
{ dataOnly: true },
|
||||
)
|
||||
.catch((e: any) => this.logger.warn(`offer push failed: ${e?.message}`));
|
||||
}
|
||||
|
||||
await this.state.addOffers(tenantId, trip.id, offeredUserIds);
|
||||
return offeredUserIds.length;
|
||||
}
|
||||
|
||||
/** «الرحلة لم تعد متاحة» لبقية السائقين — WebSocket + FCM (A5). */
|
||||
private async cancelLosingOffers(tenantId: string, tripId: string, winnerUserId: string) {
|
||||
const losers = await this.state.takeLosingOffers(tenantId, tripId, winnerUserId);
|
||||
if (losers.length === 0) return;
|
||||
for (const userId of losers) {
|
||||
this.gateway.offerTaken(tenantId, userId, { tripId });
|
||||
}
|
||||
await this.notifications.sendLocalizedToMany(
|
||||
tenantId,
|
||||
losers,
|
||||
'trip.offer_taken',
|
||||
{},
|
||||
{ type: 'trip_offer_taken', tripId },
|
||||
{ dataOnly: true },
|
||||
);
|
||||
}
|
||||
|
||||
/** بث + FCM لطرفَي الرحلة على كل انتقال (docs/17 — A1). */
|
||||
private async notifyParties(
|
||||
tenantId: string,
|
||||
trip: Trip,
|
||||
status: TripStatus,
|
||||
driverUserId: string | null,
|
||||
extra: Record<string, any> = {},
|
||||
) {
|
||||
const payload = { tripId: trip.id, status, driverId: trip.driver_id, ...extra };
|
||||
this.gateway.tripUpdate(tenantId, trip.rider_id, payload);
|
||||
if (driverUserId) this.gateway.tripUpdate(tenantId, driverUserId, payload);
|
||||
|
||||
const key =
|
||||
status === 'cancelled' && Number(extra.fee ?? 0) > 0
|
||||
? 'trip.cancelled_fee'
|
||||
: `trip.${status}`;
|
||||
const params = {
|
||||
fare: trip.final_fare ?? trip.quoted_fare ?? '—',
|
||||
currency: trip.currency ?? '',
|
||||
...extra,
|
||||
};
|
||||
const data = { type: `trip_${status}`, ...payload };
|
||||
|
||||
const targets = [trip.rider_id, ...(driverUserId ? [driverUserId] : [])];
|
||||
await this.notifications.sendLocalizedToMany(tenantId, targets, key, params, data);
|
||||
}
|
||||
|
||||
private async recordEvent(
|
||||
@@ -331,3 +519,15 @@ export class TripsService {
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/** المسافة بين نقطتين بالكيلومترات (لتصفية الرحلات المتاحة). */
|
||||
function haversineKm(lat1: number, lng1: number, lat2: number, lng2: number): number {
|
||||
const R = 6371;
|
||||
const toRad = (d: number) => (d * Math.PI) / 180;
|
||||
const dLat = toRad(lat2 - lat1);
|
||||
const dLng = toRad(lng2 - lng1);
|
||||
const a =
|
||||
Math.sin(dLat / 2) ** 2 +
|
||||
Math.cos(toRad(lat1)) * Math.cos(toRad(lat2)) * Math.sin(dLng / 2) ** 2;
|
||||
return 2 * R * Math.asin(Math.sqrt(a));
|
||||
}
|
||||
|
||||
@@ -30,6 +30,10 @@ export class User {
|
||||
@Column({ default: 'active' })
|
||||
status: string;
|
||||
|
||||
// لغة الواجهة والإشعارات (ar | en). العربية هي الافتراض.
|
||||
@Column({ default: 'ar' })
|
||||
language: string;
|
||||
|
||||
@CreateDateColumn()
|
||||
created_at: Date;
|
||||
|
||||
|
||||
@@ -21,7 +21,13 @@ export class UsersController {
|
||||
}
|
||||
|
||||
@Patch('me')
|
||||
updateMe(@CurrentUser() user: AuthUser, @Body('name') name: string) {
|
||||
return this.users.updateProfile(user.tenantId, user.userId, { name });
|
||||
updateMe(
|
||||
@CurrentUser() user: AuthUser,
|
||||
@Body() body: { name?: string; language?: string },
|
||||
) {
|
||||
return this.users.updateProfile(user.tenantId, user.userId, {
|
||||
name: body.name,
|
||||
language: body.language,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,14 +35,26 @@ export class UsersService {
|
||||
async updateProfile(
|
||||
tenantId: string,
|
||||
id: string,
|
||||
data: Partial<Pick<User, 'name'>>,
|
||||
data: Partial<Pick<User, 'name' | 'language'>>,
|
||||
): Promise<User | null> {
|
||||
if (data.name !== undefined) {
|
||||
await this.userRepository.update({ tenant_id: tenantId, id }, { name: data.name });
|
||||
const patch: Partial<User> = {};
|
||||
if (data.name !== undefined) patch.name = data.name;
|
||||
if (data.language !== undefined) patch.language = data.language;
|
||||
if (Object.keys(patch).length > 0) {
|
||||
await this.userRepository.update({ tenant_id: tenantId, id }, patch);
|
||||
}
|
||||
return this.findById(tenantId, id);
|
||||
}
|
||||
|
||||
/** لغة الإشعارات للمستخدم — استعلام خفيف (عمود واحد) يُستدعى قبل كل push. */
|
||||
async getLanguage(tenantId: string, id: string): Promise<string | null> {
|
||||
const row = await this.userRepository.findOne({
|
||||
where: { tenant_id: tenantId, id },
|
||||
select: { language: true },
|
||||
});
|
||||
return row?.language ?? null;
|
||||
}
|
||||
|
||||
async setRole(tenantId: string, id: string, role: string): Promise<void> {
|
||||
await this.userRepository.update(
|
||||
{ tenant_id: tenantId, id },
|
||||
|
||||
@@ -22,6 +22,10 @@ export class WalletTxn {
|
||||
@Column({ type: 'numeric', precision: 12, scale: 3 })
|
||||
amount: number;
|
||||
|
||||
// الرصيد بعد تطبيق هذه الحركة — يجعل الدفتر قابلاً للتدقيق ومطابقة الرصيد.
|
||||
@Column({ type: 'numeric', precision: 12, scale: 3, nullable: true })
|
||||
balance_after: number | null;
|
||||
|
||||
@Column()
|
||||
type: string; // credit | debit
|
||||
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
import { randomUUID } from 'crypto';
|
||||
import { newDb } from 'pg-mem';
|
||||
import { DataSource } from 'typeorm';
|
||||
import { BadRequestException } from '@nestjs/common';
|
||||
import { Wallet } from './entities/wallet.entity';
|
||||
import { WalletTxn } from './entities/wallet-txn.entity';
|
||||
import { WalletService } from './wallet.service';
|
||||
|
||||
/**
|
||||
* يعمل على Postgres في الذاكرة (pg-mem) — يثبت أن SQL المولَّد صحيح وأن دلالة
|
||||
* الخصم المشروط تعمل. التزامن الحقيقي (عمليات متوازية على نفس الصف) لا يمكن
|
||||
* إثباته هنا لأن pg-mem أحادي الخيط — لذلك scripts/wallet-race-test.mjs
|
||||
* يُشغَّل على السيرفر مقابل Postgres حقيقي.
|
||||
*/
|
||||
const TENANT = '11111111-1111-1111-1111-111111111111';
|
||||
const USER = '22222222-2222-2222-2222-222222222222';
|
||||
|
||||
describe('WalletService', () => {
|
||||
let ds: DataSource;
|
||||
let wallet: WalletService;
|
||||
|
||||
beforeEach(async () => {
|
||||
const db = newDb({ autoCreateForeignKeyIndices: true });
|
||||
db.public.registerFunction({
|
||||
name: 'version',
|
||||
returns: 'text' as any,
|
||||
implementation: () => 'pg-mem',
|
||||
});
|
||||
db.public.registerFunction({
|
||||
name: 'current_database',
|
||||
returns: 'text' as any,
|
||||
implementation: () => 'tripz',
|
||||
});
|
||||
db.registerExtension('uuid-ossp', (schema) =>
|
||||
schema.registerFunction({
|
||||
name: 'uuid_generate_v4',
|
||||
returns: 'uuid' as any,
|
||||
implementation: () => randomUUID(),
|
||||
impure: true,
|
||||
}),
|
||||
);
|
||||
await db.public.none(`CREATE EXTENSION "uuid-ossp"`);
|
||||
|
||||
ds = (await db.adapters.createTypeormDataSource({
|
||||
type: 'postgres',
|
||||
entities: [Wallet, WalletTxn],
|
||||
entityPrefix: 'tripz_',
|
||||
})) as DataSource;
|
||||
await ds.initialize();
|
||||
await ds.synchronize();
|
||||
|
||||
// شبكة الأمان التي تضيفها الهجرة WalletLedger
|
||||
await ds.query(
|
||||
`ALTER TABLE tripz_wallets ADD CONSTRAINT tripz_wallets_balance_non_negative CHECK (balance >= 0)`,
|
||||
);
|
||||
|
||||
wallet = new WalletService(ds.getRepository(Wallet), ds.getRepository(WalletTxn));
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
if (ds?.isInitialized) await ds.destroy();
|
||||
});
|
||||
|
||||
it('ينشئ المحفظة عند أول إيداع ويضبط الرصيد', async () => {
|
||||
const w = await wallet.credit(TENANT, USER, 10, 'topup');
|
||||
expect(w.balance).toBe(10);
|
||||
});
|
||||
|
||||
it('يجمع الإيداعات المتتالية بلا فقدان', async () => {
|
||||
await wallet.credit(TENANT, USER, 10, 'topup');
|
||||
await wallet.credit(TENANT, USER, 5.5, 'topup');
|
||||
const w = await wallet.getOrCreate(TENANT, USER);
|
||||
expect(Number(w.balance)).toBe(15.5);
|
||||
});
|
||||
|
||||
it('يخصم عند توفّر الرصيد', async () => {
|
||||
await wallet.credit(TENANT, USER, 20, 'topup');
|
||||
const w = await wallet.debit(TENANT, USER, 8, 'trip_fare');
|
||||
expect(w.balance).toBe(12);
|
||||
});
|
||||
|
||||
it('يرفض الخصم عند نقص الرصيد ولا يغيّر شيئاً', async () => {
|
||||
await wallet.credit(TENANT, USER, 5, 'topup');
|
||||
await expect(wallet.debit(TENANT, USER, 9, 'trip_fare')).rejects.toThrow(BadRequestException);
|
||||
|
||||
const w = await wallet.getOrCreate(TENANT, USER);
|
||||
expect(Number(w.balance)).toBe(5);
|
||||
// القيد المرفوض لا يُسجَّل في الدفتر
|
||||
const txns = await wallet.history(TENANT, w.id);
|
||||
expect(txns.filter((t) => t.type === 'debit')).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('يرفض الخصم من محفظة غير موجودة', async () => {
|
||||
await expect(wallet.debit(TENANT, USER, 1, 'trip_fare')).rejects.toThrow(BadRequestException);
|
||||
});
|
||||
|
||||
it('يرفض المبالغ غير الصالحة', async () => {
|
||||
for (const bad of [0, -5, NaN, Infinity]) {
|
||||
await expect(wallet.credit(TENANT, USER, bad, 'topup')).rejects.toThrow(BadRequestException);
|
||||
await expect(wallet.debit(TENANT, USER, bad, 'payout')).rejects.toThrow(BadRequestException);
|
||||
}
|
||||
});
|
||||
|
||||
it('الدفتر يطابق الرصيد ويسجّل balance_after', async () => {
|
||||
await wallet.credit(TENANT, USER, 30, 'topup');
|
||||
await wallet.debit(TENANT, USER, 12, 'trip_fare');
|
||||
await wallet.credit(TENANT, USER, 2, 'refund');
|
||||
|
||||
const w = await wallet.getOrCreate(TENANT, USER);
|
||||
const txns = await wallet.history(TENANT, w.id);
|
||||
expect(txns).toHaveLength(3);
|
||||
|
||||
const sum = txns.reduce(
|
||||
(acc, t) => acc + (t.type === 'credit' ? 1 : -1) * Number(t.amount),
|
||||
0,
|
||||
);
|
||||
expect(sum).toBe(Number(w.balance));
|
||||
expect(sum).toBe(20);
|
||||
|
||||
const last = txns.find((t) => t.reason === 'refund')!;
|
||||
expect(Number(last.balance_after)).toBe(20);
|
||||
});
|
||||
|
||||
it('محافظ مستأجرين مختلفين معزولة رغم نفس user_id', async () => {
|
||||
const other = '33333333-3333-3333-3333-333333333333';
|
||||
await wallet.credit(TENANT, USER, 10, 'topup');
|
||||
await wallet.credit(other, USER, 7, 'topup');
|
||||
|
||||
expect(Number((await wallet.getOrCreate(TENANT, USER)).balance)).toBe(10);
|
||||
expect(Number((await wallet.getOrCreate(other, USER)).balance)).toBe(7);
|
||||
});
|
||||
});
|
||||
@@ -1,9 +1,20 @@
|
||||
import { BadRequestException, Injectable } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { EntityManager, Repository } from 'typeorm';
|
||||
import { Wallet } from './entities/wallet.entity';
|
||||
import { WalletTxn } from './entities/wallet-txn.entity';
|
||||
|
||||
type TxnType = 'credit' | 'debit';
|
||||
|
||||
/**
|
||||
* المحفظة — دفتر قيود (wallet_txns) + رصيد مادّي (wallets.balance).
|
||||
*
|
||||
* قاعدة صارمة (docs/17 — I1): **لا يُقرأ الرصيد ثم يُكتب**. كل تغيير يمرّ بعبارة
|
||||
* `UPDATE … SET balance = balance ± :delta` واحدة، والخصم مشروط بـ
|
||||
* `balance >= :amount` داخل نفس العبارة. القاعدة تسلسل التحديثات على الصف،
|
||||
* فخصمان متزامنان لا يكتبان فوق بعضهما.
|
||||
* القيد والرصيد يُكتبان في **معاملة واحدة** — لا خصم بلا قيد ولا قيد بلا خصم.
|
||||
*/
|
||||
@Injectable()
|
||||
export class WalletService {
|
||||
constructor(
|
||||
@@ -12,40 +23,16 @@ export class WalletService {
|
||||
) {}
|
||||
|
||||
async getOrCreate(tenantId: string, userId: string, currency = 'JOD'): Promise<Wallet> {
|
||||
let w = await this.wallets.findOne({ where: { tenant_id: tenantId, user_id: userId } });
|
||||
if (!w) {
|
||||
w = await this.wallets.save(
|
||||
this.wallets.create({ tenant_id: tenantId, user_id: userId, balance: 0, currency }),
|
||||
);
|
||||
}
|
||||
return w;
|
||||
await this.ensureWallet(this.wallets.manager, tenantId, userId, currency);
|
||||
return (await this.wallets.findOne({ where: { tenant_id: tenantId, user_id: userId } }))!;
|
||||
}
|
||||
|
||||
private num(v: number | string): number {
|
||||
return typeof v === 'string' ? Number(v) : v;
|
||||
credit(tenantId: string, userId: string, amount: number, reason: string, ref?: string) {
|
||||
return this.apply(tenantId, userId, amount, 'credit', reason, ref);
|
||||
}
|
||||
|
||||
async credit(tenantId: string, userId: string, amount: number, reason: string, ref?: string) {
|
||||
if (amount <= 0) throw new BadRequestException('amount must be > 0');
|
||||
const w = await this.getOrCreate(tenantId, userId);
|
||||
w.balance = Number((this.num(w.balance) + amount).toFixed(3));
|
||||
await this.wallets.save(w);
|
||||
await this.txns.save(
|
||||
this.txns.create({ tenant_id: tenantId, wallet_id: w.id, amount, type: 'credit', reason, ref }),
|
||||
);
|
||||
return w;
|
||||
}
|
||||
|
||||
async debit(tenantId: string, userId: string, amount: number, reason: string, ref?: string) {
|
||||
if (amount <= 0) throw new BadRequestException('amount must be > 0');
|
||||
const w = await this.getOrCreate(tenantId, userId);
|
||||
if (this.num(w.balance) < amount) throw new BadRequestException('Insufficient balance');
|
||||
w.balance = Number((this.num(w.balance) - amount).toFixed(3));
|
||||
await this.wallets.save(w);
|
||||
await this.txns.save(
|
||||
this.txns.create({ tenant_id: tenantId, wallet_id: w.id, amount, type: 'debit', reason, ref }),
|
||||
);
|
||||
return w;
|
||||
debit(tenantId: string, userId: string, amount: number, reason: string, ref?: string) {
|
||||
return this.apply(tenantId, userId, amount, 'debit', reason, ref);
|
||||
}
|
||||
|
||||
history(tenantId: string, walletId: string) {
|
||||
@@ -55,4 +42,72 @@ export class WalletService {
|
||||
take: 100,
|
||||
});
|
||||
}
|
||||
|
||||
// ---- داخلي ----
|
||||
|
||||
/** إنشاء المحفظة إن غابت. ON CONFLICT DO NOTHING — إنشاءان متزامنان لا يتصادمان. */
|
||||
private async ensureWallet(
|
||||
em: EntityManager,
|
||||
tenantId: string,
|
||||
userId: string,
|
||||
currency = 'JOD',
|
||||
): Promise<void> {
|
||||
await em
|
||||
.createQueryBuilder()
|
||||
.insert()
|
||||
.into(Wallet)
|
||||
.values({ tenant_id: tenantId, user_id: userId, balance: 0, currency })
|
||||
.orIgnore()
|
||||
.execute();
|
||||
}
|
||||
|
||||
private async apply(
|
||||
tenantId: string,
|
||||
userId: string,
|
||||
amount: number,
|
||||
type: TxnType,
|
||||
reason: string,
|
||||
ref?: string,
|
||||
): Promise<Wallet> {
|
||||
if (!Number.isFinite(amount) || amount <= 0) {
|
||||
throw new BadRequestException('amount must be > 0');
|
||||
}
|
||||
const delta = type === 'credit' ? amount : -amount;
|
||||
|
||||
return this.wallets.manager.transaction(async (em) => {
|
||||
await this.ensureWallet(em, tenantId, userId);
|
||||
|
||||
const qb = em
|
||||
.createQueryBuilder()
|
||||
.update(Wallet)
|
||||
.set({ balance: () => 'balance + :delta' })
|
||||
.where('tenant_id = :tenantId AND user_id = :userId')
|
||||
.setParameters({ delta, tenantId, userId });
|
||||
|
||||
// شرط الرصيد داخل نفس العبارة: لا فجوة بين الفحص والخصم.
|
||||
if (type === 'debit') qb.andWhere('balance >= :amount', { amount });
|
||||
|
||||
const res = await qb.returning('*').execute();
|
||||
if (!res.raw?.length) throw new BadRequestException('Insufficient balance');
|
||||
|
||||
const wallet = this.hydrate(res.raw[0]);
|
||||
|
||||
await em.getRepository(WalletTxn).insert({
|
||||
tenant_id: tenantId,
|
||||
wallet_id: wallet.id,
|
||||
amount,
|
||||
type,
|
||||
reason,
|
||||
ref,
|
||||
balance_after: wallet.balance,
|
||||
});
|
||||
|
||||
return wallet;
|
||||
});
|
||||
}
|
||||
|
||||
/** Postgres يرجّع numeric كنص — نوحّده رقماً كما تتوقّعه بقية الخدمات. */
|
||||
private hydrate(row: any): Wallet {
|
||||
return { ...row, balance: Number(row.balance) } as Wallet;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -108,6 +108,11 @@ export class RealtimeGateway implements OnGatewayConnection {
|
||||
this.server.to(`tenant:${tenantId}:user:${driverUserId}`).emit('trip:offer', payload);
|
||||
}
|
||||
|
||||
/** العرض مات (قبِله غيره أو أُلغي) — يزيله من شاشة السائق والـoverlay. */
|
||||
offerTaken(tenantId: string, driverUserId: string, payload: any) {
|
||||
this.server.to(`tenant:${tenantId}:user:${driverUserId}`).emit('trip:offer_taken', payload);
|
||||
}
|
||||
|
||||
tripUpdate(tenantId: string, userId: string, payload: any) {
|
||||
this.server.to(`tenant:${tenantId}:user:${userId}`).emit('trip:update', payload);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user