fix(core/orders): stop the order upsert resetting fulfillmentState to NULL

apps/api/test/integration/orders/order-record-fulfillment-state.int-spec.ts

@@ -0,0 +1,154 @@
+/**
+ * Order Record Fulfillment-State Persistence Int-Spec (#2101)
+ *
+ * Proves that a re-ingestion of the same order cannot reset the fulfillment
+ * rollup the shipping context wrote out-of-band. A mocked repository spec can
+ * only assert "the property was never set on the ORM entity"; only a real
+ * `save()` against Postgres proves TypeORM actually omits the column from the
+ * generated `UPDATE` and the committed `'dispatched'` value survives.
+ *
+ * Also covers the operator-visible consequence: the re-persisted order must not
+ * re-enter the not-shipped list filter or its ship-by SLA bucket.
+ *
+ * @module apps/api/test/integration/orders
+ */
+import {
+  ORDER_RECORD_SERVICE_TOKEN,
+  deriveSlaState,
+  type IOrderRecordService,
+  type Order,
+} from '@openlinker/core/orders';
+import { OrderRecordOrmEntity } from '@openlinker/core/orders/orm-entities';
+import {
+  getTestHarness,
+  resetTestHarness,
+  teardownTestHarness,
+  type IntegrationTestHarness,
+} from '../setup';
+import { createTestConnection } from '../helpers/test-connection.helper';
+
+const PAGE = { limit: 50, offset: 0 };
+
+function makeOrder(overrides: Partial<Order> = {}): Order {
+  return {
+    id: 'ol_order_fulfillment_state_test',
+    orderNumber: 'ORD-FULFILL-1',
+    status: 'pending',
+    customerId: null,
+    items: [],
+    totals: { subtotal: 0, tax: 0, shipping: 0, total: 0, currency: 'PLN' },
+    shippingAddress: {
+      firstName: 'Jan',
+      lastName: 'Kowalski',
+      address1: 'ul. Testowa 1',
+      city: 'Warszawa',
+      postalCode: '00-001',
+      country: 'PL',
+    },
+    billingAddress: {
+      firstName: 'Jan',
+      lastName: 'Kowalski',
+      address1: 'ul. Testowa 1',
+      city: 'Warszawa',
+      postalCode: '00-001',
+      country: 'PL',
+    },
+    createdAt: new Date('2026-08-01T00:00:00Z'),
+    updatedAt: new Date('2026-08-01T00:00:00Z'),
+    ...overrides,
+  } as Order;
+}
+
+describe('Order record fulfillment state survives re-ingestion (#2101)', () => {
+  let harness: IntegrationTestHarness;
+  let orderRecordService: IOrderRecordService;
+
+  beforeAll(async () => {
+    harness = await getTestHarness();
+    orderRecordService = harness.getApp().get<IOrderRecordService>(ORDER_RECORD_SERVICE_TOKEN);
+  });
+
+  afterEach(async () => {
+    await resetTestHarness();
+  });
+
+  afterAll(async () => {
+    await teardownTestHarness();
+  });
+
+  it('keeps a dispatched rollup after a later re-poll of the same order (persistOrder)', async () => {
+    const dataSource = harness.getDataSource();
+    const recordRepo = dataSource.getRepository(OrderRecordOrmEntity);
+    const connection = await createTestConnection(dataSource, {
+      platformType: 'allegro',
+      name: 'Allegro source',
+      adapterKey: 'allegro.test.unused',
+    });
+
+    const order = makeOrder();
+    await orderRecordService.persistOrder(order, connection.id, 'evt-1');
+
+    // The shipping context dispatches the order and pushes the rollup.
+    await orderRecordService.updateFulfillmentState(order.id, 'dispatched');
+    const afterDispatch = await recordRepo.findOne({ where: { internalOrderId: order.id } });
+    expect(afterDispatch!.fulfillmentState).toBe('dispatched');
+
+    // A reconciliation poll re-pulls the same order. Its full-object upsert()
+    // must not reset the rollup - the ingestion path never carries one.
+    await orderRecordService.persistOrder(
+      makeOrder({ id: order.id, updatedAt: new Date('2026-08-02T00:00:00Z') }),
+      connection.id,
+      'evt-2'
+    );
+
+    const row = await recordRepo.findOne({ where: { internalOrderId: order.id } });
+    expect(row!.fulfillmentState).toBe('dispatched');
+
+    const found = await orderRecordService.getOrderRecord(order.id);
+    expect(found!.fulfillmentState).toBe('dispatched');
+    // The re-pull still refreshes the columns it does own.
+    expect(found!.sourceEventId).toBe('evt-2');
+  });
+
+  it('does not reclassify a re-polled dispatched order as not-shipped or SLA-pressured', async () => {
+    const dataSource = harness.getDataSource();
+    const connection = await createTestConnection(dataSource, {
+      platformType: 'allegro',
+      name: 'Allegro source',
+      adapterKey: 'allegro.test.unused',
+    });
+
+    // A past ship-by deadline: were the rollup reset to NULL, this order would
+    // read `not-shipped` and re-enter the `overdue` bucket.
+    const dispatchTo = new Date(Date.now() - 60 * 60 * 1000).toISOString();
+    const order = makeOrder({
+      id: 'ol_order_fulfillment_sla_test',
+      dispatchTime: { to: dispatchTo },
+    });
+
+    await orderRecordService.persistOrder(order, connection.id, 'evt-1');
+    await orderRecordService.updateFulfillmentState(order.id, 'dispatched');
+    await orderRecordService.persistOrder(
+      makeOrder({
+        id: order.id,
+        dispatchTime: { to: dispatchTo },
+        updatedAt: new Date('2026-08-02T00:00:00Z'),
+      }),
+      connection.id,
+      'evt-2'
+    );
+
+    const notShipped = await orderRecordService.findMany({ fulfillmentState: 'not-shipped' }, PAGE);
+    expect(notShipped.items.map((o) => o.internalOrderId)).not.toContain(order.id);
+
+    const overdue = await orderRecordService.findMany({ slaState: 'overdue' }, PAGE);
+    expect(overdue.items.map((o) => o.internalOrderId)).not.toContain(order.id);
+
+    const dispatched = await orderRecordService.findMany({ fulfillmentState: 'dispatched' }, PAGE);
+    expect(dispatched.items.map((o) => o.internalOrderId)).toContain(order.id);
+
+    // The domain derivation the API response mapper uses agrees with the SQL.
+    const found = await orderRecordService.getOrderRecord(order.id);
+    expect(deriveSlaState(found!.dispatchByAt, found!.fulfillmentState, new Date())).toBe('none');
+  });
+});

docs/lessons.md

@@ -23,6 +23,13 @@ When a lesson hardens into a rule, **graduate it** to the canonical doc and leav
 
 ---
 
+## A column written by a narrow out-of-band UPDATE must be excluded from the full-row upsert's write set
+
+**Context**: `order_records` carries denormalized columns that no ingestion payload supplies and that a different context pushes in with a narrow `UPDATE` - `fulfillmentState` (a rollup over the order's shipments, written by `updateFulfillmentState`) and `cancelledAt` (written by `markCancelled`). `OrderRecordRepository.upsert` is a full-object TypeORM `save()`.
+**Problem**: `toOrm` mapped `fulfillmentState` unconditionally while `persistOrder` never populated it (it is the 12th constructor argument and only 11 were passed), so every re-ingestion - a poll re-read, a webhook-triggered sync, a manual re-sync - wrote `null` over a committed `'dispatched'`. A dispatched order silently reappeared as not-shipped in the ship-by SLA buckets and the not-shipped list filter. The same class of defect had already been fixed for `cancelledAt` in the same method (#1984) and the exclusion comment sat three lines below the offending assignment.
+**Rule**: When a column has a dedicated out-of-band writer, that writer must be its **only** writer: leave the ORM entity property unset in `toOrm` so TypeORM omits the column from the generated statement, and say so in a comment next to the columns that *are* mapped. Do not "fix" it by reading the row first and carrying the value onto the new instance - an unlocked upsert racing the out-of-band UPDATE still loses a value that commits between the read and the save. Pin it with a unit test asserting the property is `undefined` on the entity handed to `save()` **and** an integration test proving the committed value survives a second persist (a mocked spec cannot prove TypeORM really omits the column). Note the consequence: the record returned by the upsert reports such a column as `null` whatever the row holds, so a caller needing its live value must re-read.
+**Applies to**: any repository whose `upsert`/`save` coexists with a narrow `UPDATE` writer on the same table - today `libs/core/src/orders/infrastructure/persistence/repositories/order-record.repository.ts` (`fulfillmentState`, `cancelledAt`).
+**Source**: #2101 (surfaced reviewing #2050 / ADR-040, which adopts the narrow-conditional-UPDATE shape for its own columns); the `cancelledAt` precedent is #1984.
 ## Claim an ADR number from the "Reserved numbers" note, not from the last row of the index table
 
 **Context**: #2066 authored three ADRs and numbered them 039/040/041 by reading the index table in `docs/architecture/adrs/README.md` and taking "last merged row + 1".

libs/core/src/orders/application/services/__tests__/order-record.service.spec.ts

@@ -486,6 +486,27 @@ describe('OrderRecordService', () => {
     });
   });
 
+  describe('persistOrder - fulfillment rollup left to updateFulfillmentState (#2101)', () => {
+    beforeEach(() => {
+      process.env.OL_STORE_PII = 'true';
+      service = new OrderRecordService(repository);
+    });
+
+    it('never constructs the OrderRecord passed to upsert() with a fulfillment state', async () => {
+      // No order source reports a fulfillment state, so the ingestion path must
+      // leave the field at its null default. Passing a derived value here would
+      // let the upsert's full-object save() reset the rollup the shipping
+      // context committed out-of-band - see the toOrm comment in
+      // OrderRecordRepository.
+      repository.upsert.mockResolvedValue({} as OrderRecord);
+
+      await service.persistOrder(createMockOrder(), 'source-connection-123', 'event-456');
+
+      const callArg = repository.upsert.mock.calls[0][0];
+      expect(callArg.fulfillmentState).toBeNull();
+    });
+  });
+
   describe('persistIncomingSnapshot', () => {
     beforeEach(() => {
       process.env.OL_STORE_PII = 'true';

libs/core/src/orders/application/services/order-record.service.ts

@@ -125,6 +125,11 @@ export class OrderRecordService implements IOrderRecordService {
     // Initial sync status: pending for all destinations (will be updated as sync progresses)
     const syncStatus: OrderSyncStatus[] = [];
 
+    // `fulfillmentState` is intentionally left at its constructor default:
+    // it is a rollup over the order's shipments, not a source-payload field,
+    // and the upsert excludes the column so a re-ingestion can't reset the
+    // value the shipping context wrote out-of-band (#2101). Same for
+    // `cancelledAt` (#1984), recorded below via `recordCancellationIfNeeded`.
     const orderRecord = new OrderRecord(
       order.id,
       order.customerId || null,

libs/core/src/orders/domain/entities/order-record.entity.ts

@@ -56,6 +56,11 @@ export class OrderRecord {
      * `updateFulfillmentState`. `null` ≡ `not-shipped` (no backfill needed).
      * Lets the orders list show/filter "has this shipped?" without reaching
      * into the shipping context.
+     *
+     * Never sourced from an ingestion payload: no order source reports a
+     * fulfillment state, and `updateFulfillmentState` is the column's sole
+     * writer, so the ingestion path leaves this field at its `null` default
+     * and the upsert excludes the column entirely (#2101).
      */
     public readonly fulfillmentState: FulfillmentRollupState | null = null,
     /**

libs/core/src/orders/domain/ports/order-record-repository.port.ts

@@ -42,7 +42,15 @@ export interface OrderRecordRepositoryPort {
 
   /**
    * Upsert order record (create or update)
-   * Uses internalOrderId as the primary key
+   * Uses internalOrderId as the primary key.
+   *
+   * Writes only the columns the ingestion path owns. The two columns written
+   * out-of-band by a narrow, atomic UPDATE - `fulfillmentState` (#2101,
+   * {@link updateFulfillmentState}) and `cancelledAt` (#1984,
+   * {@link markCancelled}) - are NOT part of the write set, so a re-ingestion
+   * of the same order cannot reset them. The returned record therefore reports
+   * both as `null` whatever the row holds; re-read via {@link findById} when
+   * their live value matters.
    */
   upsert(orderRecord: OrderRecord): Promise<OrderRecord>;
 
@@ -102,6 +110,8 @@ export interface OrderRecordRepositoryPort {
    * from the shipping context after a shipment-status change (best-effort
    * projection). Idempotent absolute-set; a missing order row is a no-op (never
    * throws) so it can't fail the shipment operation.
+   *
+   * Sole writer of the column - {@link upsert} deliberately omits it (#2101).
    */
   updateFulfillmentState(
     internalOrderId: string,

libs/core/src/orders/infrastructure/persistence/repositories/__tests__/order-record.repository.spec.ts

@@ -282,6 +282,54 @@ describe('OrderRecordRepository', () => {
       const callArg = ormRepository.save.mock.calls[0][0] as OrderRecordOrmEntity;
       expect(callArg.cancelledAt).toBeUndefined();
     });
+
+    it('should NOT include fulfillmentState in the entity passed to save() (#2101)', async () => {
+      // The ingestion path never carries a fulfillment rollup, so writing the
+      // column here reset a `'dispatched'` order to NULL on every re-poll.
+      // Leaving the property unset lets TypeORM omit the column from the
+      // generated UPDATE - updateFulfillmentState is the sole writer.
+      ormRepository.save.mockResolvedValue(createOrmEntity());
+
+      await repository.upsert(createDomainEntity());
+
+      const callArg = ormRepository.save.mock.calls[0][0] as OrderRecordOrmEntity;
+      expect(callArg.fulfillmentState).toBeUndefined();
+    });
+
+    it('should NOT write fulfillmentState even when the domain record carries one', async () => {
+      // Guards against a future caller reintroducing the clobber by passing a
+      // rollup value through the ingestion path.
+      const domainEntity = new OrderRecord(
+        'order-123',
+        null,
+        'conn-123',
+        null,
+        {},
+        [],
+        'ready',
+        new Date('2025-01-01T10:00:00Z'),
+        new Date('2025-01-01T10:00:00Z'),
+        [],
+        null,
+        'dispatched'
+      );
+      ormRepository.save.mockResolvedValue(createOrmEntity());
+
+      await repository.upsert(domainEntity);
+
+      const callArg = ormRepository.save.mock.calls[0][0] as OrderRecordOrmEntity;
+      expect(callArg.fulfillmentState).toBeUndefined();
+    });
+
+    it('should read fulfillmentState back via toDomain when present on the ORM row', async () => {
+      const savedEntity = createOrmEntity();
+      savedEntity.fulfillmentState = 'dispatched';
+      ormRepository.save.mockResolvedValue(savedEntity);
+
+      const result = await repository.upsert(createDomainEntity());
+
+      expect(result.fulfillmentState).toBe('dispatched');
+    });
   });
 
   describe('findMany', () => {

libs/core/src/orders/infrastructure/persistence/repositories/order-record.repository.ts

@@ -633,6 +633,15 @@ export class OrderRecordRepository implements OrderRecordRepositoryPort {
     );
   }
 
+  /**
+   * Full-row upsert of the ingestion-owned columns, keyed on the primary key.
+   *
+   * `fulfillmentState` (#2101) and `cancelledAt` (#1984) are deliberately
+   * outside the write set - see the {@link toOrm} comments. A consequence is
+   * that the returned record reports both as `null` regardless of what the row
+   * holds, because neither column was part of the statement; callers needing
+   * their true value re-read via {@link findById}.
+   */
   async upsert(orderRecord: OrderRecord): Promise<OrderRecord> {
     const entity = this.toOrm(orderRecord);
     // TypeORM save() performs upsert on primary key (internalOrderId)
@@ -797,7 +806,17 @@ export class OrderRecordRepository implements OrderRecordRepositoryPort {
     entity.recordStatus = orderRecord.recordStatus;
     entity.mappingFailureReason = orderRecord.mappingFailureReason;
     entity.dispatchByAt = orderRecord.dispatchByAt;
-    entity.fulfillmentState = orderRecord.fulfillmentState;
+    // fulfillmentState is deliberately NOT mapped here (#2101), for the same
+    // reason as cancelledAt below: it is OL-owned state rolled up from the
+    // order's shipments, never re-derivable from the source payload, and
+    // {@link updateFulfillmentState} is its sole writer. Mapping it made every
+    // re-ingestion (a poll re-read, a webhook-triggered sync, a manual
+    // re-sync) write the ingestion path's in-memory `null` over a
+    // `'dispatched'` value the shipping context had already committed, so a
+    // dispatched order reappeared as not-shipped in the ship-by SLA buckets
+    // and the not-shipped list filter. Carrying the value forward with a
+    // read-before-write would still lose a rollup that commits between that
+    // read and this save; omitting the column is race-free.
     // cancelledAt is deliberately NOT mapped here (#1984 follow-up). upsert()
     // is a full-object save() with no per-order lock around it (two ingestion
     // paths — webhook + reconciliation poll — legitimately race for the same