diff --git a/src/controllers/deliveryEventController.ts b/src/controllers/deliveryEventController.ts new file mode 100644 index 0000000..3103672 --- /dev/null +++ b/src/controllers/deliveryEventController.ts @@ -0,0 +1,58 @@ +import type { NextFunction, Request, Response } from 'express'; +import { deliveryService } from '../services/delivery.service'; +import { AppError } from '../utils/AppError'; + +const DEFAULT_LIMIT = 50; +const MAX_LIMIT = 100; + +type AuthedRequest = Request & { + user?: { _id?: unknown; id?: unknown; role?: unknown }; +}; + +function parsePositiveInt(value: unknown, fallback: number): number { + const parsed = typeof value === 'string' ? Number.parseInt(value, 10) : Number.NaN; + return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; +} + +/** + * GET /api/v1/deliveries/:id/events — ordered milestone log for a delivery. + * + * Layering: Controller -> Service -> Repository. Authorization (participant + * or admin) is enforced in the service so every caller gets the same rule. + */ +export const listDeliveryMilestones = async ( + req: Request, + res: Response, + next: NextFunction, +): Promise => { + try { + const user = (req as AuthedRequest).user; + const userId = user?._id ?? user?.id; + if (!user || !userId) { + throw new AppError('Authentication required.', 401); + } + + const page = parsePositiveInt(req.query.page, 1); + const limit = Math.min(parsePositiveInt(req.query.limit, DEFAULT_LIMIT), MAX_LIMIT); + + const result = await deliveryService.listMilestones( + req.params.id, + { id: String(userId), role: String(user.role ?? '') }, + page, + limit, + ); + + res.status(200).json({ + success: true, + data: result.data, + meta: { + page: result.page, + limit: result.limit, + total: result.total, + totalPages: result.totalPages, + }, + }); + } catch (error) { + next(error); + } +}; diff --git a/src/controllers/deliveryStatusController.ts b/src/controllers/deliveryStatusController.ts index 3f62b85..62b109a 100644 --- a/src/controllers/deliveryStatusController.ts +++ b/src/controllers/deliveryStatusController.ts @@ -57,7 +57,13 @@ export const updateDeliveryStatus = async ( return next(new AppError('Invalid status value', 400)); } - const delivery = await deliveryService.updateStatus(id, nextStatus as DeliveryStatus); + const user = (req as Request & { user?: { _id?: unknown; id?: unknown } }).user; + const userId = user?._id ?? user?.id; + const actor = userId + ? { type: 'user' as const, id: String(userId) } + : { type: 'system' as const }; + + const delivery = await deliveryService.updateStatus(id, nextStatus as DeliveryStatus, actor); sendSuccess(res, delivery, 'Delivery status updated successfully'); } catch (error) { diff --git a/src/models/DeliveryEvent.ts b/src/models/DeliveryEvent.ts new file mode 100644 index 0000000..4ac409f --- /dev/null +++ b/src/models/DeliveryEvent.ts @@ -0,0 +1,80 @@ +import mongoose, { Schema, Document, Types } from 'mongoose'; +import { DeliveryStatus } from './Delivery'; + +/** Milestones on the delivery lifecycle: created -> funded -> assigned -> in_progress -> completed/cancelled. */ +export enum DeliveryMilestone { + CREATED = 'created', + FUNDED = 'funded', + ASSIGNED = 'assigned', + IN_PROGRESS = 'in_progress', + COMPLETED = 'completed', + CANCELLED = 'cancelled', +} + +/** The milestone raised when a delivery enters each status. */ +export const MILESTONE_BY_STATUS: Record = { + [DeliveryStatus.PENDING]: DeliveryMilestone.CREATED, + [DeliveryStatus.FUNDED]: DeliveryMilestone.FUNDED, + [DeliveryStatus.ASSIGNED]: DeliveryMilestone.ASSIGNED, + [DeliveryStatus.IN_PROGRESS]: DeliveryMilestone.IN_PROGRESS, + [DeliveryStatus.COMPLETED]: DeliveryMilestone.COMPLETED, + [DeliveryStatus.CANCELLED]: DeliveryMilestone.CANCELLED, +}; + +export type MilestoneActorType = 'user' | 'system' | 'indexer'; + +export interface IMilestoneActor { + type: MilestoneActorType; + /** User id for `user` actors; omitted for system/indexer actors. */ + id?: string; +} + +export interface IDeliveryEvent extends Document { + delivery: Types.ObjectId; + event: DeliveryMilestone; + /** `null` for the initial `created` milestone. */ + fromStatus: DeliveryStatus | null; + toStatus: DeliveryStatus; + payload: Record; + actor: IMilestoneActor; + occurredAt: Date; + /** 1-based, gapless, strictly increasing per delivery. */ + seq: number; + createdAt: Date; +} + +const ActorSchema = new Schema( + { + type: { type: String, enum: ['user', 'system', 'indexer'], required: true }, + id: { type: String }, + }, + { _id: false }, +); + +const DeliveryEventSchema = new Schema( + { + delivery: { type: Schema.Types.ObjectId, ref: 'Delivery', required: true }, + event: { type: String, enum: Object.values(DeliveryMilestone), required: true }, + fromStatus: { type: String, enum: [...Object.values(DeliveryStatus), null], default: null }, + toStatus: { type: String, enum: Object.values(DeliveryStatus), required: true }, + payload: { type: Schema.Types.Mixed, default: {} }, + actor: { type: ActorSchema, required: true }, + occurredAt: { type: Date, required: true, default: Date.now }, + seq: { + type: Number, + required: true, + min: [1, 'seq starts at 1'], + validate: { validator: Number.isInteger, message: 'seq must be an integer' }, + }, + }, + // Append-only log: no updatedAt, no version key. + { timestamps: { createdAt: true, updatedAt: false }, versionKey: false }, +); + +// Ordering is enforced by the database: one seq per delivery, ever. +DeliveryEventSchema.index({ delivery: 1, seq: 1 }, { unique: true }); +DeliveryEventSchema.index({ delivery: 1, occurredAt: 1 }); + +const DeliveryEvent = mongoose.model('DeliveryEvent', DeliveryEventSchema); + +export default DeliveryEvent; diff --git a/src/repositories/DeliveryEventRepository.ts b/src/repositories/DeliveryEventRepository.ts new file mode 100644 index 0000000..8357500 --- /dev/null +++ b/src/repositories/DeliveryEventRepository.ts @@ -0,0 +1,104 @@ +import { Types } from 'mongoose'; +import DeliveryEvent, { + DeliveryMilestone, + IDeliveryEvent, + IMilestoneActor, +} from '../models/DeliveryEvent'; +import { DeliveryStatus } from '../models/Delivery'; +import { BaseRepository } from './BaseRepository'; +import { Page, WriteOptions } from './types'; + +/** Everything needed to append a milestone; `seq` is assigned by the repository. */ +export interface AppendDeliveryEventInput { + delivery: Types.ObjectId | string; + event: DeliveryMilestone; + fromStatus: DeliveryStatus | null; + toStatus: DeliveryStatus; + payload?: Record; + actor: IMilestoneActor; + occurredAt?: Date; +} + +/** Attempts before giving up when concurrent writers keep colliding on `seq`. */ +const MAX_SEQ_ATTEMPTS = 5; + +const MONGO_DUPLICATE_KEY = 11000; + +function isDuplicateKeyError(error: unknown): boolean { + return ( + typeof error === 'object' && + error !== null && + (error as { code?: unknown }).code === MONGO_DUPLICATE_KEY + ); +} + +/** + * Persistence gateway for the append-only `DeliveryEvent` log. + * + * Ordering is guaranteed by the database: `(delivery, seq)` is a unique index, + * and `seq` is computed as the current maximum + 1. Two concurrent appends for + * the same delivery cannot both win a seq; the loser retries with a fresh one. + */ +export class DeliveryEventRepository extends BaseRepository { + constructor() { + super(DeliveryEvent); + } + + /** Highest seq recorded for a delivery, or 0 when it has no events yet. */ + async getLastSeq(deliveryId: Types.ObjectId | string): Promise { + const last = await this.model + .findOne({ delivery: deliveryId }) + .sort({ seq: -1 }) + .select('seq') + .lean() + .exec(); + return last?.seq ?? 0; + } + + /** + * Append the next milestone for a delivery, assigning a gapless `seq`. + * + * @throws The last duplicate-key error if every attempt collided. + */ + async append(input: AppendDeliveryEventInput, options?: WriteOptions): Promise { + let lastError: unknown; + + for (let attempt = 1; attempt <= MAX_SEQ_ATTEMPTS; attempt += 1) { + const seq = (await this.getLastSeq(input.delivery)) + 1; + try { + return await this.create( + { + delivery: new Types.ObjectId(String(input.delivery)), + event: input.event, + fromStatus: input.fromStatus, + toStatus: input.toStatus, + payload: input.payload ?? {}, + actor: input.actor, + occurredAt: input.occurredAt ?? new Date(), + seq, + }, + options, + ); + } catch (error) { + if (!isDuplicateKeyError(error)) throw error; + lastError = error; + } + } + + throw lastError; + } + + /** One page of a delivery's milestones, oldest first (by seq). */ + async listByDelivery( + deliveryId: string, + page: number, + limit: number, + ): Promise> { + if (!this.isValidId(deliveryId)) { + return { data: [], total: 0, page: 1, limit, totalPages: 0 }; + } + return this.paginate({ delivery: deliveryId }, page, limit, { sort: { seq: 1 } }); + } +} + +export const deliveryEventRepository = new DeliveryEventRepository(); diff --git a/src/repositories/index.ts b/src/repositories/index.ts index 08995e5..6057ec7 100644 --- a/src/repositories/index.ts +++ b/src/repositories/index.ts @@ -22,3 +22,6 @@ export { } from './NotificationPreferenceRepository'; export { NotificationRepository, notificationRepository } from './NotificationRepository'; export { ChatMessageRepository, chatMessageRepository } from './ChatMessageRepository'; + +export { DeliveryEventRepository, deliveryEventRepository } from './DeliveryEventRepository'; +export type { AppendDeliveryEventInput } from './DeliveryEventRepository'; diff --git a/src/routes/deliveryEventRoutes.ts b/src/routes/deliveryEventRoutes.ts new file mode 100644 index 0000000..9c198da --- /dev/null +++ b/src/routes/deliveryEventRoutes.ts @@ -0,0 +1,47 @@ +import { Router } from 'express'; +import { authenticate } from '../middleware/auth'; +import { listDeliveryMilestones } from '../controllers/deliveryEventController'; + +const router = Router(); + +/** + * @openapi + * /v1/deliveries/{id}/events: + * get: + * tags: [Delivery Events] + * summary: List a delivery's ordered milestone events + * description: > + * Returns the append-only milestone log (created, funded, assigned, + * in_progress, completed/cancelled) ordered by `seq`. Restricted to the + * delivery's participants and admins. + * security: + * - bearerAuth: [] + * parameters: + * - in: path + * name: id + * required: true + * schema: + * type: string + * - in: query + * name: page + * schema: + * type: integer + * - in: query + * name: limit + * schema: + * type: integer + * responses: + * 200: + * description: Milestone events, oldest first + * 400: + * description: Invalid delivery id + * 401: + * $ref: '#/components/responses/Unauthorized' + * 403: + * $ref: '#/components/responses/Forbidden' + * 404: + * $ref: '#/components/responses/NotFound' + */ +router.get('/:id/events', authenticate, listDeliveryMilestones); + +export default router; diff --git a/src/routes/index.ts b/src/routes/index.ts index d866373..0119877 100644 --- a/src/routes/index.ts +++ b/src/routes/index.ts @@ -4,6 +4,7 @@ import bulkDeliveryRoutes from './bulkDeliveryRoutes'; import deliveryCrudRoutes from './delivery.routes'; import deliveryEtaRoutes from './deliveryRoutes'; import deliveryStatusRoutes from './deliveries'; +import deliveryEventRoutes from './deliveryEventRoutes'; import adminRoutes from './adminRoutes'; import driverRoutes from './driverRoutes'; import fleetRoutes from './fleetRoutes'; @@ -35,6 +36,7 @@ router.use('/v1/deliveries', bulkDeliveryRoutes); router.use('/v1/deliveries', deliveryCrudRoutes); router.use('/v1/deliveries', deliveryEtaRoutes); router.use('/v1/deliveries', deliveryStatusRoutes); +router.use('/v1/deliveries', deliveryEventRoutes); router.use('/v1/deliveries', assignmentRoutes); router.use('/v1/deliveries', proofOfDeliveryRoutes); router.use('/v1/admin', adminRoutes); diff --git a/src/services/delivery.service.ts b/src/services/delivery.service.ts index 44740df..dc4732d 100644 --- a/src/services/delivery.service.ts +++ b/src/services/delivery.service.ts @@ -9,6 +9,28 @@ import { deliveryRepository } from '../repositories/DeliveryRepository'; import { notificationService } from './notificationService'; import { webhookService } from './webhookService'; import { proofOfDeliveryService } from './proofOfDeliveryService'; +import { deliveryEventRepository } from '../repositories/DeliveryEventRepository'; +import { + DeliveryMilestone, + IDeliveryEvent, + IMilestoneActor, + MILESTONE_BY_STATUS, +} from '../models/DeliveryEvent'; +import { emitDeliveryStatusUpdated } from '../sockets/connectionHandler'; + +/** Describes the milestone being raised for a delivery that has just committed a status change. */ +export interface MilestoneInput { + name: DeliveryMilestone; + /** Status the delivery left; `null` for the initial `created` milestone. */ + fromStatus: DeliveryStatus | null; + payload?: Record; +} + +/** Authenticated caller asking to read a delivery's milestone log. */ +export interface MilestoneRequester { + id: string; + role: string; +} export interface CreateDeliveryInput { trackingNumber: string; @@ -83,6 +105,116 @@ const ALLOWED_TRANSITIONS: Record = { }; export class DeliveryService { + /** + * Record a delivery milestone and fan it out to every consumer. + * + * Call this only AFTER the status change has committed. It never throws: + * the event is appended to the ordered `DeliveryEvent` log (gapless `seq` + * per delivery), then consumers run in a fixed order — + * notification -> webhook -> socket broadcast — each isolated so a failing + * consumer is logged and cannot roll back or fail the transition. + * + * @returns The persisted event, or `null` if it could not be recorded. + */ + async emitMilestone( + delivery: IDelivery, + event: MilestoneInput, + actor: IMilestoneActor, + ): Promise { + const deliveryId = String(delivery._id); + const toStatus = delivery.status; + + let record: IDeliveryEvent | null = null; + try { + record = await deliveryEventRepository.append({ + delivery: deliveryId, + event: event.name, + fromStatus: event.fromStatus, + toStatus, + payload: event.payload, + actor, + }); + } catch (error) { + logger.error( + `[DeliveryService] Failed to record milestone '${event.name}' for delivery=${deliveryId}: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + } + + await this.runConsumer('notification', deliveryId, () => + notificationService.notifyDeliveryTransition(delivery, toStatus), + ); + await this.runConsumer('webhook', deliveryId, () => + webhookService.dispatchDeliveryEvent(delivery, toStatus), + ); + await this.runConsumer('socket', deliveryId, async () => { + emitDeliveryStatusUpdated(deliveryId, { + contractId: delivery.contractId, + deliveryId, + status: toStatus, + event: event.name, + seq: record?.seq, + occurredAt: (record?.occurredAt ?? new Date()).toISOString(), + }); + }); + + return record; + } + + /** + * List a delivery's milestone events, oldest first (by `seq`). + * Only the delivery's participants (sender/customer account, assigned + * driver) and admins may read it. + */ + async listMilestones( + deliveryId: string, + requester: MilestoneRequester, + page = 1, + limit = 50, + ): Promise> { + if (!Types.ObjectId.isValid(deliveryId)) { + throw new AppError('Invalid delivery ID', httpStatus.BAD_REQUEST); + } + + const delivery = await deliveryRepository.findById(deliveryId); + if (!delivery) { + throw new AppError('Delivery not found', httpStatus.NOT_FOUND); + } + + const isAdmin = requester.role === 'admin'; + const participants = [ + delivery.userId, + delivery.driverId, + delivery.sender ? String(delivery.sender) : undefined, + ]; + const isParticipant = participants.some((id) => id !== undefined && id === requester.id); + + if (!isAdmin && !isParticipant) { + throw new AppError( + "You are not allowed to view this delivery's events.", + httpStatus.FORBIDDEN, + ); + } + + return deliveryEventRepository.listByDelivery(deliveryId, page, limit); + } + + /** Run one fan-out consumer; failures are logged, never propagated. */ + private async runConsumer( + consumer: 'notification' | 'webhook' | 'socket', + deliveryId: string, + run: () => Promise, + ): Promise { + try { + await run(); + } catch (error) { + logger.error( + `[DeliveryService] Milestone ${consumer} consumer failed for delivery=${deliveryId}: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + } + } + /** * Generates a QR code for secure delivery handoff verification (merged * capability from the legacy deliveryService module). @@ -103,6 +235,13 @@ export class DeliveryService { escrowAsset: validateAsset(input.escrowAsset ?? { code: 'XLM' }), }); logger.info(`Delivery created: ${delivery.trackingNumber}`); + + // Fire-after-commit: the delivery is already persisted at this point. + await this.emitMilestone( + delivery, + { name: MILESTONE_BY_STATUS[delivery.status], fromStatus: null }, + { type: 'system' }, + ); return delivery; } @@ -235,7 +374,11 @@ export class DeliveryService { * @throws {AppError} 404 — delivery not found. * @throws {AppError} 409 — the delivery changed status concurrently. */ - async updateStatus(id: string, nextStatus: DeliveryStatus): Promise { + async updateStatus( + id: string, + nextStatus: DeliveryStatus, + actor: IMilestoneActor = { type: 'system' }, + ): Promise { if (!Types.ObjectId.isValid(id)) { throw new AppError('Invalid delivery ID', httpStatus.BAD_REQUEST); } @@ -284,8 +427,11 @@ export class DeliveryService { // Fire-and-forget by design: notification/webhook failures are recorded // inside their own services and must not roll back a committed transition. - await notificationService.notifyDeliveryTransition(updated, nextStatus); - await webhookService.dispatchDeliveryEvent(updated, nextStatus); + await this.emitMilestone( + updated, + { name: MILESTONE_BY_STATUS[nextStatus], fromStatus: current.status }, + actor, + ); return updated; } @@ -390,6 +536,7 @@ export class DeliveryService { } // ── All guards passed — perform the assignment ────────────────────────── + const previousStatus = delivery.status; delivery.driverId = driverId; delivery.status = DeliveryStatus.ASSIGNED; const updated = await delivery.save(); @@ -399,6 +546,16 @@ export class DeliveryService { `driver=${driverId} escrow=${String(escrow._id)}`, ); + await this.emitMilestone( + updated, + { + name: MILESTONE_BY_STATUS[DeliveryStatus.ASSIGNED], + fromStatus: previousStatus, + payload: { driverId: String(driverId) }, + }, + { type: 'system' }, + ); + return updated; } } diff --git a/src/services/escrow.service.ts b/src/services/escrow.service.ts index 347ad59..ca9be48 100644 --- a/src/services/escrow.service.ts +++ b/src/services/escrow.service.ts @@ -12,6 +12,8 @@ import { proofOfDeliveryService } from './proofOfDeliveryService'; import { sorobanService } from '../blockchain/soroban.service'; import env from '../config/env'; import { nowUTC } from '../utils/dateUtils'; +import { deliveryService } from './delivery.service'; +import { MILESTONE_BY_STATUS } from '../models/DeliveryEvent'; /** Data extracted from an on-chain `escrow_funded` contract event. */ export interface EscrowFundedInput { @@ -172,10 +174,22 @@ export class EscrowService { } if (delivery.status !== DeliveryStatus.FUNDED) { + const previousStatus = delivery.status; delivery.status = DeliveryStatus.FUNDED; await deliveryRepository.updateById(String(delivery._id), { $set: { status: DeliveryStatus.FUNDED }, }); + + // Fire-after-commit: the status write above has landed. + await deliveryService.emitMilestone( + delivery, + { + name: MILESTONE_BY_STATUS[DeliveryStatus.FUNDED], + fromStatus: previousStatus, + payload: { contractId: input.contractId, transactionHash: input.transactionHash }, + }, + { type: 'system' }, + ); } logger.info( diff --git a/src/sockets/socket.types.ts b/src/sockets/socket.types.ts index 9a4e18d..ad54d54 100644 --- a/src/sockets/socket.types.ts +++ b/src/sockets/socket.types.ts @@ -5,12 +5,18 @@ import { Socket } from 'socket.io'; * status change on-chain (see `emitDeliveryStatusUpdated`). */ export interface DeliveryStatusUpdatedPayload { - /** Stellar contract id whose event carried the status change. */ - contractId: string; + /** Stellar contract id whose event carried the status change (absent for off-chain milestones). */ + contractId?: string; /** Delivery the status change applies to. */ deliveryId: string; /** New canonical status (see `DeliveryStatus`). */ status: string; + /** Milestone name, present when raised by `emitMilestone`. */ + event?: string; + /** Per-delivery ordering number of the milestone. */ + seq?: number; + /** ISO timestamp the milestone occurred. */ + occurredAt?: string; } /** diff --git a/tests/deliveryMilestones.test.ts b/tests/deliveryMilestones.test.ts new file mode 100644 index 0000000..418b330 --- /dev/null +++ b/tests/deliveryMilestones.test.ts @@ -0,0 +1,292 @@ +/** + * Delivery milestone event log (#245). + * + * Real in-memory MongoDB for the DeliveryEvent log; the three fan-out + * consumers (notification, webhook, socket) are mocked so call order and + * failure isolation can be asserted. + */ +import mongoose, { Types } from 'mongoose'; +import { MongoMemoryServer } from 'mongodb-memory-server'; + +jest.mock('../src/services/notificationService', () => ({ + notificationService: { notifyDeliveryTransition: jest.fn() }, +})); +jest.mock('../src/services/webhookService', () => ({ + webhookService: { dispatchDeliveryEvent: jest.fn() }, +})); +jest.mock('../src/services/proofOfDeliveryService', () => ({ + proofOfDeliveryService: { assertProofOfDeliveryExists: jest.fn() }, +})); +jest.mock('../src/sockets/connectionHandler', () => ({ + emitDeliveryStatusUpdated: jest.fn(), +})); + +import DeliveryEvent, { DeliveryMilestone } from '../src/models/DeliveryEvent'; +import { IDelivery, DeliveryStatus } from '../src/models/Delivery'; +import { deliveryService } from '../src/services/delivery.service'; +import { deliveryEventRepository } from '../src/repositories/DeliveryEventRepository'; +import { deliveryRepository } from '../src/repositories/DeliveryRepository'; +import { notificationService } from '../src/services/notificationService'; +import { webhookService } from '../src/services/webhookService'; +import { emitDeliveryStatusUpdated } from '../src/sockets/connectionHandler'; + +const notify = notificationService.notifyDeliveryTransition as unknown as jest.Mock; +const webhook = webhookService.dispatchDeliveryEvent as unknown as jest.Mock; +const socket = emitDeliveryStatusUpdated as unknown as jest.Mock; + +const makeDelivery = (status: DeliveryStatus, extra: Record = {}): IDelivery => + ({ _id: new Types.ObjectId(), status, contractId: 'CTEST', ...extra }) as unknown as IDelivery; + +describe('delivery milestone events', () => { + let mongod: MongoMemoryServer; + + beforeAll(async () => { + mongod = await MongoMemoryServer.create(); + await mongoose.connect(mongod.getUri()); + await DeliveryEvent.init(); + }, 60_000); + + afterAll(async () => { + await mongoose.disconnect(); + await mongod.stop(); + }, 30_000); + + beforeEach(() => { + jest.restoreAllMocks(); + notify.mockReset().mockResolvedValue(undefined); + webhook.mockReset().mockResolvedValue(undefined); + socket.mockReset(); + }); + + afterEach(async () => { + await DeliveryEvent.deleteMany({}); + }); + + describe('ordering', () => { + it('assigns a gapless, increasing seq per delivery', async () => { + const delivery = makeDelivery(DeliveryStatus.PENDING); + const actor = { type: 'system' as const }; + + await deliveryService.emitMilestone( + delivery, + { name: DeliveryMilestone.CREATED, fromStatus: null }, + actor, + ); + delivery.status = DeliveryStatus.FUNDED; + await deliveryService.emitMilestone( + delivery, + { name: DeliveryMilestone.FUNDED, fromStatus: DeliveryStatus.PENDING }, + actor, + ); + delivery.status = DeliveryStatus.ASSIGNED; + await deliveryService.emitMilestone( + delivery, + { name: DeliveryMilestone.ASSIGNED, fromStatus: DeliveryStatus.FUNDED }, + actor, + ); + + const page = await deliveryEventRepository.listByDelivery(String(delivery._id), 1, 10); + expect(page.data.map((e) => e.seq)).toEqual([1, 2, 3]); + expect(page.data.map((e) => e.event)).toEqual([ + DeliveryMilestone.CREATED, + DeliveryMilestone.FUNDED, + DeliveryMilestone.ASSIGNED, + ]); + }); + + it('keeps seq independent between deliveries', async () => { + const a = makeDelivery(DeliveryStatus.PENDING); + const b = makeDelivery(DeliveryStatus.PENDING); + const actor = { type: 'system' as const }; + const created = { name: DeliveryMilestone.CREATED, fromStatus: null }; + + await deliveryService.emitMilestone(a, created, actor); + await deliveryService.emitMilestone(a, created, actor); + const first = await deliveryService.emitMilestone(b, created, actor); + + expect(first?.seq).toBe(1); + }); + + it('never produces duplicate seqs under concurrent writers', async () => { + const delivery = makeDelivery(DeliveryStatus.PENDING); + const actor = { type: 'system' as const }; + const created = { name: DeliveryMilestone.CREATED, fromStatus: null }; + + await Promise.all([ + deliveryService.emitMilestone(delivery, created, actor), + deliveryService.emitMilestone(delivery, created, actor), + deliveryService.emitMilestone(delivery, created, actor), + ]); + + const events = await DeliveryEvent.find({ delivery: delivery._id }).sort({ seq: 1 }); + expect(events.map((e) => e.seq)).toEqual([1, 2, 3]); + }); + }); + + describe('fan-out', () => { + it('runs notification -> webhook -> socket, after the record is persisted', async () => { + const delivery = makeDelivery(DeliveryStatus.FUNDED); + const order: string[] = []; + notify.mockImplementation(async () => { + order.push('notification'); + }); + webhook.mockImplementation(async () => { + order.push('webhook'); + }); + socket.mockImplementation(() => { + order.push('socket'); + }); + + const record = await deliveryService.emitMilestone( + delivery, + { name: DeliveryMilestone.FUNDED, fromStatus: DeliveryStatus.PENDING }, + { type: 'system' }, + ); + + expect(order).toEqual(['notification', 'webhook', 'socket']); + expect(record?.seq).toBe(1); + expect(socket).toHaveBeenCalledWith( + String(delivery._id), + expect.objectContaining({ + event: DeliveryMilestone.FUNDED, + seq: 1, + status: DeliveryStatus.FUNDED, + }), + ); + }); + + it('still delivers to consumers when the record cannot be written', async () => { + const delivery = makeDelivery(DeliveryStatus.FUNDED); + jest.spyOn(deliveryEventRepository, 'append').mockRejectedValue(new Error('db down')); + + const record = await deliveryService.emitMilestone( + delivery, + { name: DeliveryMilestone.FUNDED, fromStatus: DeliveryStatus.PENDING }, + { type: 'system' }, + ); + + expect(record).toBeNull(); + expect(notify).toHaveBeenCalledTimes(1); + expect(webhook).toHaveBeenCalledTimes(1); + expect(socket).toHaveBeenCalledTimes(1); + }); + }); + + describe('fire-after-commit isolation', () => { + it('a throwing consumer neither rejects emitMilestone nor skips later consumers', async () => { + const delivery = makeDelivery(DeliveryStatus.FUNDED); + notify.mockRejectedValue(new Error('notify boom')); + webhook.mockRejectedValue(new Error('webhook boom')); + socket.mockImplementation(() => { + throw new Error('socket boom'); + }); + + const record = await deliveryService.emitMilestone( + delivery, + { name: DeliveryMilestone.FUNDED, fromStatus: DeliveryStatus.PENDING }, + { type: 'system' }, + ); + + expect(record?.seq).toBe(1); + expect(await DeliveryEvent.countDocuments({ delivery: delivery._id })).toBe(1); + expect(webhook).toHaveBeenCalledTimes(1); + expect(socket).toHaveBeenCalledTimes(1); + }); + + it('updateStatus emits only after the transition commits, and consumer failures cannot break it', async () => { + const current = makeDelivery(DeliveryStatus.ASSIGNED); + const updated = { ...current, status: DeliveryStatus.IN_PROGRESS } as unknown as IDelivery; + const id = String(current._id); + const order: string[] = []; + + jest.spyOn(deliveryRepository, 'findById').mockResolvedValue(current); + jest.spyOn(deliveryRepository, 'transitionStatus').mockImplementation(async () => { + order.push('commit'); + return updated; + }); + notify.mockImplementation(async () => { + order.push('notification'); + }); + webhook.mockImplementation(async () => { + order.push('webhook'); + throw new Error('webhook boom'); + }); + socket.mockImplementation(() => { + order.push('socket'); + }); + + const result = await deliveryService.updateStatus(id, DeliveryStatus.IN_PROGRESS, { + type: 'user', + id: 'user-1', + }); + + expect(result).toBe(updated); + expect(order).toEqual(['commit', 'notification', 'webhook', 'socket']); + + const [event] = await DeliveryEvent.find({ delivery: id }); + expect(event.event).toBe(DeliveryMilestone.IN_PROGRESS); + expect(event.fromStatus).toBe(DeliveryStatus.ASSIGNED); + expect(event.toStatus).toBe(DeliveryStatus.IN_PROGRESS); + expect(event.actor).toMatchObject({ type: 'user', id: 'user-1' }); + }); + + it('does not emit anything when the transition is rejected', async () => { + const current = makeDelivery(DeliveryStatus.COMPLETED); + jest.spyOn(deliveryRepository, 'findById').mockResolvedValue(current); + + await expect( + deliveryService.updateStatus(String(current._id), DeliveryStatus.IN_PROGRESS), + ).rejects.toMatchObject({ statusCode: 400 }); + + expect(await DeliveryEvent.countDocuments({})).toBe(0); + expect(notify).not.toHaveBeenCalled(); + }); + }); + + describe('listMilestones authorization', () => { + const setup = async (): Promise => { + const delivery = makeDelivery(DeliveryStatus.ASSIGNED, { + userId: 'owner-1', + driverId: 'driver-1', + }); + jest.spyOn(deliveryRepository, 'findById').mockResolvedValue(delivery); + await deliveryService.emitMilestone( + delivery, + { name: DeliveryMilestone.ASSIGNED, fromStatus: DeliveryStatus.FUNDED }, + { type: 'system' }, + ); + return String(delivery._id); + }; + + it('lets the owner, the driver and an admin read the log', async () => { + const id = await setup(); + for (const requester of [ + { id: 'owner-1', role: 'customer' }, + { id: 'driver-1', role: 'driver' }, + { id: 'someone-else', role: 'admin' }, + ]) { + const page = await deliveryService.listMilestones(id, requester); + expect(page.total).toBe(1); + expect(page.data[0].seq).toBe(1); + } + }); + + it('rejects a non-participant with 403', async () => { + const id = await setup(); + await expect( + deliveryService.listMilestones(id, { id: 'stranger', role: 'driver' }), + ).rejects.toMatchObject({ statusCode: 403 }); + }); + + it('rejects a malformed id with 400 and an unknown delivery with 404', async () => { + await expect( + deliveryService.listMilestones('not-an-id', { id: 'x', role: 'admin' }), + ).rejects.toMatchObject({ statusCode: 400 }); + + jest.spyOn(deliveryRepository, 'findById').mockResolvedValue(null); + await expect( + deliveryService.listMilestones(String(new Types.ObjectId()), { id: 'x', role: 'admin' }), + ).rejects.toMatchObject({ statusCode: 404 }); + }); + }); +});