feat(deliveries): stream ordered milestone events to webhooks, socket…

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<void> => {
+  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);
+  }
+};

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) {

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, DeliveryMilestone> = {
+  [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<string, unknown>;
+  actor: IMilestoneActor;
+  occurredAt: Date;
+  /** 1-based, gapless, strictly increasing per delivery. */
+  seq: number;
+  createdAt: Date;
+}
+
+const ActorSchema = new Schema<IMilestoneActor>(
+  {
+    type: { type: String, enum: ['user', 'system', 'indexer'], required: true },
+    id: { type: String },
+  },
+  { _id: false },
+);
+
+const DeliveryEventSchema = new Schema<IDeliveryEvent>(
+  {
+    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<IDeliveryEvent>('DeliveryEvent', DeliveryEventSchema);
+
+export default DeliveryEvent;

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<string, unknown>;
+  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<IDeliveryEvent> {
+  constructor() {
+    super(DeliveryEvent);
+  }
+
+  /** Highest seq recorded for a delivery, or 0 when it has no events yet. */
+  async getLastSeq(deliveryId: Types.ObjectId | string): Promise<number> {
+    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<IDeliveryEvent> {
+    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<Page<IDeliveryEvent>> {
+    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();

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';

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;

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);