import { ERPObjType } from "../../types/erp-types.js";
import {
  IERPLaborTicketHandler,
  IERPTimesheetHandler,
  IERPWorkOrderReconciliationHandler,
} from "../../types/erp-connector.js";
import {
  WorkOrderReconciler,
  type ReconcileDeletedWorkOrdersOptions,
  type ReconcileDeletedWorkOrdersResult,
} from "./work-order-reconciler.js";
import type { FlattenedWorkOrderRow } from "../../types/flattened-work-order.js";
import { BatchCacheManager } from "../../services/caching-service/batch-cache-manager.js";
import {
  IToRESTApiObject,
  MMSendWorkOrderBatch,
  MMSendPart,
  MMSendPartOperation,
  MMSendWorkOrder,
  MMSendWorkOrderOperation,
  MMSendLaborTicket,
} from "../../services/mm-api-service/index.js";
import { LaborTicketERPSynchronizer } from "./labor-ticket-erp-synchronizer.js";
import { TimesheetERPSynchronizer } from "./timesheet-erp-synchronizer.js";
import { MMEntityProcessor } from "./mm-entity-processor.js";
import {
  filterLaborTicketsForMmImport,
  type FilterLaborTicketsForMmImportOptions,
  type FilterLaborTicketsForMmImportResult,
  type SyncedWorkOrderOperationRow,
} from "../labor-import-from-erp.js";
import {
  DEFAULT_GAP_FILL_FETCH_WARN_THRESHOLD,
  DEFAULT_MAX_TRACKED_WORK_ORDERS,
  LaborWorkOrderGapFillTracker,
  computeWorkOrderGapFillWriteScope,
  findLaborWorkOrderIdsMissingOps,
  findMissingWorkOrderOpKeys,
  findOpenLaborWorkOrderIdsMissingOps,
  isTerminalWorkOrderForGapFill,
  mmPartIdentityKey,
  planWorkOrderGapFillFetch,
  updateTrackedAfterWorkOrderGapFillFetch,
  type LaborTicketWorkOrderOpRef,
  type TrackedLaborReferencedWorkOrder,
} from "../labor-work-order-gap-fill.js";
import { MMApiClient } from "../../services/mm-api-service/mm-api-service.js";
import { EntityTransformer } from "../../services/mm-api-service/types/entity-transformer.js";
import { sdkLogger as logger } from "../../services/reporting-service/logger.js";

/**
 * The result of writing entities to the MachineMetrics API on complete success.
 */
export interface WriteEntitiesToMMResult {
  message: string;
  upsertedEntities: number;
  localDedupeCount: number;
  apiDedupeCount: number;
}

/**
 * The result of {@link StandardProcessDrivers.retryRejectedIndividually}.
 */
export interface RetryRejectedIndividuallyResult {
  /** Entities that imported successfully when retried alone (collateral rejects of a bad sub-batch neighbor). */
  succeeded: IToRESTApiObject[];
  /** Residual failures: genuinely rejected entities (or entities with no matching original record) with their validation message. Must not be silently dropped. */
  failed: Array<{ entity: IToRESTApiObject; message: string }>;
}

export class MMBatchValidationError extends Error {
  public readonly upsertedEntities: number;
  public readonly localDedupeCount: number;
  public readonly apiDedupeCount: number;
  public readonly errorCount: number;
  public readonly httpStatus: number;
  public readonly batchErrors: Array<{
    message: string;
    affectedEntities: IToRESTApiObject[]; // All entities in the failing batch
  }>;

  constructor(options: {
    message: string;
    upsertedEntities: number;
    localDedupeCount: number;
    apiDedupeCount: number;
    errorCount: number;
    httpStatus: number;
    batchErrors: Array<{
      message: string;
      affectedEntities: IToRESTApiObject[]; // All entities in the failing batch
    }>;
  }) {
    super(options.message);
    this.name = "MMBatchValidationError";
    this.upsertedEntities = options.upsertedEntities;
    this.localDedupeCount = options.localDedupeCount;
    this.apiDedupeCount = options.apiDedupeCount;
    this.errorCount = options.errorCount;
    this.httpStatus = options.httpStatus;
    this.batchErrors = options.batchErrors;
  }
}

/**
 * A collection of standardized process drivers that orchestrate various synchronization
 * and integration processes between systems. Use of these drivers is optional,
 * and useful as a reference implementation for how to implement the various processes.
 */
export interface FillWorkOrderGapsFromLaborTicketsOptions {
  /** Namespaces the persisted watch list (one per connector type). */
  connectorType: string;
  /** This cycle's from-ERP labor tickets (any state), before the labor filter. */
  laborTickets: ReadonlyArray<LaborTicketWorkOrderOpRef>;
  /** Work order operation rows the bulk work-order sync wrote this cycle. */
  bulkSyncedWorkOrderOperations: ReadonlyArray<SyncedWorkOrderOperationRow>;
  /**
   * Connector-owned: fetch the given work orders from the ERP and map them to a full
   * {@link MMSendWorkOrderBatch} (every operation of each work order, plus its part, part
   * operations and work order rows). Ids the ERP does not know are simply absent from the
   * result (reported as `unresolvedWorkOrderIds`; they stay on the watch list until the
   * TTL). Must throw on an ERP failure rather than return a partial batch.
   */
  fetchWorkOrders: (workOrderIds: string[]) => Promise<MMSendWorkOrderBatch>;
  /**
   * Let CLOSED tickets drive a fetch too (default false: OPEN tickets only, which bounds
   * the fetch by concurrent clock-ins). Off, a work order observed only through closed
   * tickets is never fetched and those tickets stay rejected; the driver reports such ids
   * as `closedOnlyLaborWorkOrderIds` and warns. On, the bound becomes the closed tickets
   * in the connector's labor window, so enable it only when that window is narrow.
   */
  includeClosedTickets?: boolean;
  /** Default {@link isTerminalWorkOrderForGapFill} (terminal status or a closed date). */
  isTerminalWorkOrder?: (workOrder: MMSendWorkOrder) => boolean;
  /**
   * Replaces {@link StandardProcessDrivers.writeWorkOrderBatchToMM} as the writer, for
   * connectors whose bulk writer does something the SDK writer does not yet (e.g. routing
   * RELEASED rows through `?clearColumns=closedDate`). The batch is already FK-complete
   * and in referential order.
   *
   * Must throw unless every entity in the batch was written, as
   * {@link StandardProcessDrivers.writeWorkOrderBatchToMM} does on partial success. A
   * resolved promise is taken as "the close-out landed" and removes the terminal work
   * orders from the watch list; they sit outside the bulk window, so one removed after a
   * write that did not land is never fetched again and keeps a stale status in MM.
   */
  writeBatch?: (batch: MMSendWorkOrderBatch) => Promise<unknown>;
  /** Forwarded to the default writer; ignored when `writeBatch` is given. */
  batchCacheManager?: BatchCacheManager | null;
  /** Default: a {@link LaborWorkOrderGapFillTracker} for `connectorType`. */
  tracker?: LaborWorkOrderGapFillTracker;
  /** Clock override (tests). */
  now?: Date;
  /** Default {@code DEFAULT_TRACKED_WORK_ORDER_TTL_DAYS}. */
  trackedTtlDays?: number;
  /**
   * Default {@code DEFAULT_MAX_TRACKED_WORK_ORDERS}. Caps the open entries; pending
   * close-outs are never evicted to honor it.
   */
  maxTrackedWorkOrders?: number;
  /** Default {@link DEFAULT_GAP_FILL_FETCH_WARN_THRESHOLD}. */
  fetchWarnThreshold?: number;
}

export interface FillWorkOrderGapsFromLaborTicketsResult {
  /**
   * Every work order operation of the fetched work orders, in the shape of
   * `FilterLaborTicketsForMmImportOptions.syncedWorkOrderOperations`. Concatenate with the
   * bulk rows before the labor write. Empty when nothing was fetched.
   */
  keySetWorkOrderOps: SyncedWorkOrderOperationRow[];
  /** Work order ids referenced by OPEN tickets with no bulk-synced operation. */
  openLaborWorkOrderIds: string[];
  /**
   * Work order ids referenced only by non-OPEN tickets with no bulk-synced operation.
   * Fetched only when `includeClosedTickets` is set; otherwise their tickets will be
   * rejected by the labor filter this cycle.
   */
  closedOnlyLaborWorkOrderIds: string[];
  /** Work order ids handed to `fetchWorkOrders` (open labor ∪ watch list). */
  fetchedWorkOrderIds: string[];
  /**
   * Requested ids for which `fetchWorkOrders` returned no work order row (deleted in the
   * ERP, or outside the connector's query scope). They stay on the watch list until the TTL.
   */
  unresolvedWorkOrderIds: string[];
  /** The batch written to MM, or null when nothing needed writing. */
  writtenBatch: MMSendWorkOrderBatch | null;
  /** The writer's return value (the SDK writer's per-entity results by default), or null. */
  writeResult: unknown;
  /**
   * Terminal work orders with rows outside the bulk window; those rows and their full
   * operation sets were written this cycle.
   */
  closingWorkOrderIds: string[];
  /**
   * The watch list as persisted when this call returned: work orders to re-fetch until
   * their terminal state is imported. Work orders fetched terminal this cycle are still
   * listed when their close-out write did not happen (nothing to write yet), and gone once
   * it has; they are counted in `dropped.terminal` either way.
   */
  trackedWorkOrderIds: string[];
  dropped: {
    expired: number;
    /** Tracked work orders whose every fetched row the bulk sync covered this cycle. */
    coveredByBulkSync: number;
    /** Fetched terminal this cycle (leave the list once their close-out write lands). */
    terminal: number;
    evicted: number;
  };
}

export class StandardProcessDrivers {
  /**
   * Synchronizes updated labor tickets from MachineMetrics to an ERP system:
   *
   * Initializes a checkpoint that tracks the last time labor tickets were synced,
   * fetches labor tickets from MM and converts them to the company timezone,
   * creates or updates the labor ticket in the ERP via connector-specific implementations,
   * updates the labor ticket id in MM on newly created labor tickets,
   * and saves the checkpoint
   */
  static async syncLaborTicketsToERP(
    connectorType: string,
    connector: IERPLaborTicketHandler,
    options?: {
      onlyClosed?: boolean;   // true: process only CLOSED labor tickets from MM; false (default): process all labor tickets
    }
  ): Promise<void> {
    return LaborTicketERPSynchronizer.syncToERP(connectorType, connector, options);
  }

  /**
   * Synchronizes timesheet rows (employee clock in/out) from MachineMetrics to an ERP system.
   * Run after {@link syncLaborTicketsToERP} in the same job so labor is written/closed before clock-out.
   */
  static async syncTimesheetsToERP(
    connectorType: string,
    handler: IERPTimesheetHandler
  ): Promise<void> {
    return TimesheetERPSynchronizer.syncToERP(connectorType, handler);
  }

  /**
   * Marks MM work orders — and their operations — as DELETED when they no longer exist /
   * are active in the ERP. Intended to run rarely: the `reconcile-work-orders` job fires
   * 1 hour after process start, then every `RECONCILE_WORK_ORDERS_INTERVAL` (default
   * 24 hours). Runs are idempotent, so the occasional extra post-restart run is harmless.
   *
   * Division of labor: the SDK fetches all work orders from MM and restricts candidates
   * to status RELEASED or ACTIVE (case-insensitive); the connector's
   * {@link IERPWorkOrderReconciliationHandler} selects which candidates to delete
   * (commonly by delegating to {@code selectWorkOrdersMissingFromErp()}, plus any
   * ERP/site-specific rules); the SDK then validates the selection and writes a
   * status-only patch (identity keys + status DELETED) for each — the MM import derives
   * its upsert columns from the fields present, so no other column is touched.
   *
   * The operation phase follows: open operations of DELETED work orders are cascaded to
   * DELETED (MM has no server-side cascade), and — when the handler implements
   * `selectWorkOrderOperationsToDelete()` — open operations of surviving jobs are offered
   * to it so an operation deleted from a surviving job is caught too. Same validation
   * and guard as work orders.
   *
   * Safety behavior:
   * - The handler MUST throw on ERP fetch failure — never return a guessed selection.
   * - The selection is validated as a subset of the candidates; foreign rows reject the
   *   whole run.
   * - A blast-radius guard throws {@link ReconcileDeletedWorkOrdersAbortedError} (nothing
   *   written) when the deletion set exceeds `maxDeleteFraction` of candidates
   *   (default 20%, only enforced above `deleteGuardMinCount` deletions).
   * - `dryRun` logs the would-be deletions without writing to MM, including whether a
   *   real run would have tripped the guard.
   *
   * @throws ReconcileDeletedWorkOrdersAbortedError when the blast-radius guard trips
   * @throws Error on handler contract violations or underlying MM API failures
   */
  static async reconcileDeletedWorkOrders(
    connectorType: string,
    handler: IERPWorkOrderReconciliationHandler,
    options?: ReconcileDeletedWorkOrdersOptions
  ): Promise<ReconcileDeletedWorkOrdersResult> {
    return WorkOrderReconciler.reconcileDeleted(connectorType, handler, options);
  }

  /**
   * Retries labor tickets that have failed to be created or updated in the ERP during the sync:
   *
   * Fetches failed labor tickets from MM, processes them, and updates the labor ticket id in MM.
   * If the labor ticket is successfully created or updated, it is added to the list of successful labor ticket ids,
   * which are then deleted from the list of failed labor ticket ids.
   */
  static async retryFailedLaborTickets(
    connectorType: string,
    connector: IERPLaborTicketHandler
  ): Promise<void> {
    return LaborTicketERPSynchronizer.retryFailed(connectorType, connector);
  }

  /**
   * Filters ERP-mapped labor tickets for MM FK/import rules, then writes them via {@link writeEntitiesToMM}.
   */
  static async writeLaborTicketsToMmFromErpImport(
    tickets: MMSendLaborTicket[],
    filterOptions: FilterLaborTicketsForMmImportOptions,
    batchCacheManager?: BatchCacheManager | null,
    writeOptions?: { connectorType?: string; chunkSize?: number }
  ): Promise<WriteEntitiesToMMResult & { filter: FilterLaborTicketsForMmImportResult }> {
    const filter = await filterLaborTicketsForMmImport(tickets, filterOptions);
    if (filter.accepted.length === 0) {
      return {
        message: "No labor tickets passed FK/import filters",
        upsertedEntities: 0,
        localDedupeCount: 0,
        apiDedupeCount: 0,
        filter,
      };
    }

    // Pre-write the labor sync ledger BEFORE the MM POST. The post-write marker in
    // _handleResponse is kept as a safety net but cannot be relied on if the POST hangs
    // or the worker dies. Pre-marking guarantees the next to-erp cycle sees these tickets.
    if (writeOptions?.connectorType) {
      await MMEntityProcessor.markLaborTicketsInLedger(
        filter.accepted,
        writeOptions.connectorType,
        "labor-ledger-mark:pre-write"
      );
    } else {
      logger.warn(
        "writeLaborTicketsToMmFromErpImport: no connectorType provided; skipping labor sync ledger pre-marking. The next to-erp cycle will reprocess these tickets."
      );
    }

    logger.info(
      "writeLaborTicketsToMmFromErpImport: sending accepted labor tickets to MM API",
      { count: filter.accepted.length }
    );
    const write = await this.writeEntitiesToMM(
      ERPObjType.LABOR_TICKETS,
      filter.accepted,
      batchCacheManager,
      writeOptions
    );
    // Capture "now" right after the MM POST returns. Anything MM updated as a side-effect
    // of our writes carries an updatedAt <= this instant, so advancing the labor_tickets
    // export checkpoint to it lets the next to-erp cycle skip the tickets we just wrote
    // (the dedupe-by-signature path can drift due to MM-side normalization). Real-time
    // edits to those tickets in MM still arrive via NATS, so the skipped polling window
    // is acceptable.
    const checkpointAfterWriteIso = new Date().toISOString();
    logger.info("writeLaborTicketsToMmFromErpImport: MM API write finished", {
      upsertedEntities: write.upsertedEntities,
      localDedupeCount: write.localDedupeCount,
      apiDedupeCount: write.apiDedupeCount,
    });

    if (writeOptions?.connectorType) {
      await this.advanceLaborTicketsExportCheckpoint(
        writeOptions.connectorType,
        checkpointAfterWriteIso
      );
    } else {
      logger.warn(
        "writeLaborTicketsToMmFromErpImport: no connectorType provided; skipping labor_tickets export-checkpoint advance. The next to-erp cycle will still see these tickets in its updates feed."
      );
    }

    return { ...write, filter };
  }

  /**
   * Best-effort advance of the {@code labor_tickets / export} checkpoint past the just-finished
   * import window. Failures here are logged and swallowed — the to-erp ledger/signature dedupe
   * is the real correctness guarantee; this is a belt-and-suspenders skip-window optimization.
   */
  static async advanceLaborTicketsExportCheckpoint(
    connectorType: string,
    timestampIso: string
  ): Promise<void> {
    const mmApiClient = new MMApiClient();
    try {
      await mmApiClient.saveCheckpoint({
        system: connectorType,
        table: "labor_tickets",
        checkpointType: "export",
        checkpointValue: { timestamp: timestampIso },
      });
      logger.info(
        "advanceLaborTicketsExportCheckpoint: labor_tickets export checkpoint advanced",
        { connectorType, checkpointTimestamp: timestampIso }
      );
    } catch (error) {
      logger.warn(
        "advanceLaborTicketsExportCheckpoint: failed to advance labor_tickets export checkpoint; to-erp will fall back to ledger/signature dedupe",
        { connectorType, checkpointTimestamp: timestampIso, error }
      );
    } finally {
      try {
        await mmApiClient.destroy();
      } catch (destroyError) {
        logger.warn(
          "advanceLaborTicketsExportCheckpoint: failed to destroy MMApiClient",
          { error: destroyError }
        );
      }
    }
  }

  /**
   * Writes a batch of records to MM for a specific entity type by:
   * 1. Deduplicating against cached records (localDedupeCount)
   * 2. Sending non-duplicate records to the MM API (apiDedupeCount)
   * 3. Storing successful records in cache
   *
   * An example usage pattern including comprehensive error handling:
   * ```
   * try {
   *   const result = await StandardProcessDrivers.writeEntitiesToMM(
   *     ERPObjType.WORK_ORDERS,
   *     workOrderRecords,
   *     batchCacheManager
   *   );
   *
   *   // HTTP 200 - Complete success
   *   console.log(`✅ Success: ${result.message}`);
   *   console.log(`📊 Metrics: ${result.upsertedEntities} upserted, ${result.localDedupeCount} locally deduplicated, ${result.apiDedupeCount} API deduplicated`);
   *
   * } catch (error) {
   *   if (error instanceof MMBatchValidationError) {
   *     // HTTP 207 - Partial success with some batches failing due to validation errors
   *     // HTTP 500 - A specific type representing complete failure due to validation issues;
   *     //            other 500 types represent failure due to other issues and are not converted to MMBatchValidationError
   *     //
   *     // MM imports in ATOMIC SUB-BATCHES: each batchErrors entry contains ALL entities of its
   *     // failing sub-batch — valid records rejected as collateral of a bad neighbor, not just the
   *     // offending record.
   *     //
   *     // REQUIRED: your connector must recover the entities in batchErrors[].affectedEntities —
   *     // never advance a record-tracking cursor/checkpoint past them, or they freeze in MM with
   *     // stale values until their ERP rows happen to change again. Pick one of:
   *     //   1. retry individually: StandardProcessDrivers.retryRejectedIndividually(entityType, error, mmRecords)
   *     //      (collateral records succeed; log the residual failures at ERROR level)
   *     //   2. persist the matching ORIGINAL records and re-queue them on a later sync cycle
   *     //      (never re-send affectedEntities themselves — they are lossy reconstructions)
   *     //   3. hold your cursor/checkpoint at/before the first rejected entity so they are re-fetched
   *     // See CONNECTOR_SDK_UTILITIES.md → "Partial-failure handling for from-ERP writes".
   *
   *     console.error(`Batch processing completed with errors (HTTP ${error.httpStatus}): ${error.errorCount} rejected`);
   *
   *     if (error.httpStatus === 207) {
   *       // Partial success — recover the rejected sub-batches (option 1 shown):
   *       const { failed } = await StandardProcessDrivers.retryRejectedIndividually(
   *         ERPObjType.WORK_ORDERS,
   *         error,
   *         workOrderRecords,
   *         batchCacheManager
   *       );
   *       for (const residual of failed) {
   *         // Genuinely invalid records: log loudly so the rejection is visible in production.
   *         console.error(`Work order permanently rejected: ${residual.message}`, residual.entity);
   *       }
   *     } else {
   *       // HTTP 500 — complete failure, nothing was written; safe to rethrow and retry the
   *       // whole batch next cycle (do not advance tracking state).
   *       throw error;
   *     }
   *
   *   } else {
   *     // Other underlying errors (network issues, authentication, etc.) — nothing is known about
   *     // individual entities; rethrow without advancing tracking state.
   *     console.error('🚨 Unexpected error during MM API call:', error);
   *     throw error;
   *   }
   * }
   * ```
   *
   * @param entityType The type of entity being processed
   * @param mmRecords The records to process
   * @param batchCacheManager The batch cache manager instance.
   * - Omit / pass `undefined` to use a default cache manager (recommended).
   * - Pass `null` to explicitly disable caching + local deduplication (generally not recommended; primarily useful for debugging/emergency bypass).
   * @param options Optional settings:
   * - `connectorType`: marks written labor tickets in the local sync ledger (labor tickets only).
   * - `chunkSize`: max records per MM API request. When set (> 0), large record sets are written
   *   in sequential chunks so each request stays within the API timeout. Chunking is OFF by
   *   default — connectors that chunk at their own layer are unaffected unless they opt in here
   *   or via env `MM_API_WRITE_CHUNK_SIZE`. On chunked writes, structured per-chunk failures do
   *   not stop the remaining chunks — one aggregated MMBatchValidationError is thrown at the end,
   *   and a structured 500 on one chunk surfaces as 207 when any other chunk succeeded.
   * - `queryParams`: extra query params forwarded to the MM API write request (work orders and
   *   work order operations today), e.g. `{ clearColumns: 'closedDate' }` to opt specific columns
   *   into null-clearing.
   *
   * @returns WriteEntitiesToMMResult on complete success (HTTP 200)
   * @throws MMBatchValidationError on partial success (HTTP 207) or complete failure (HTTP 500) with structured error details
   * @throws Error on other underlying issues (network, authentication, etc.)
   */
  static async writeEntitiesToMM(
    entityType: ERPObjType,
    mmRecords: IToRESTApiObject[],
    batchCacheManager?: BatchCacheManager | null,
    options?: {
      connectorType?: string;
      chunkSize?: number;
      queryParams?: Record<string, string>;
    }
  ): Promise<WriteEntitiesToMMResult> {
    const ownsCacheManager = batchCacheManager === undefined;
    const resolvedCacheManager =
      batchCacheManager === undefined ? new BatchCacheManager() : batchCacheManager;

    try {
      return await MMEntityProcessor.writeEntities(
        entityType,
        mmRecords,
        resolvedCacheManager,
        options
      );
    } finally {
      if (ownsCacheManager && resolvedCacheManager) {
        await resolvedCacheManager.destroy();
      }
    }
  }

  /**
   * Re-imports the entities rejected by a {@link writeEntitiesToMM} call one record at a time.
   *
   * MM imports in atomic sub-batches, so `error.batchErrors[].affectedEntities` contains valid
   * records rejected as collateral of a bad neighbor. The same entity may appear under several
   * `batchErrors` entries (MM reports one entry per validation message, each echoing the whole
   * sub-batch); each entity is retried only once. Retried individually the valid records succeed; the
   * returned `failed` list is only the genuinely-invalid residue, each paired with its validation
   * message. Callers must not silently drop the residue: log it at ERROR level, and never advance
   * a record-tracking cursor/checkpoint past a rejected entity that has not been recovered — see
   * CONNECTOR_SDK_UTILITIES.md → "Partial-failure handling for from-ERP writes".
   *
   * The entities inside `batchErrors` are reconstructions of the API's echoed data and default any
   * field absent from the echo, so they are never re-sent as-is. Instead each rejected entity is
   * matched back to `originalRecords` by primary key and the caller's ORIGINAL object is retried.
   * An entity with no matching original lands in `failed` unretried — re-sending its reconstruction
   * could overwrite real MM values (e.g. a work-order-operation labor-ticket patch would round-trip
   * into a full row with zeroed quantities).
   *
   * Individual retries bypass the local dedupe read entirely — a stale cache entry must never
   * turn a rejected record into a silent no-op "success" — and records that succeed are stored
   * in the cache exactly as a normal write would store them.
   *
   * Non-validation errors (network, auth, timeout) abort the retry loop and are rethrown as-is —
   * they say nothing about the entities themselves, so treating them as residual validation
   * failures would misclassify transient outages as permanently-bad records.
   *
   * @param entityType The entity type that was being written when the error was thrown
   * @param error The MMBatchValidationError thrown by {@link writeEntitiesToMM}
   * @param originalRecords The records passed to the failing {@link writeEntitiesToMM} call (a superset is fine); rejected entities are matched back to these by primary key and the originals are re-sent. 
   *        Composite drivers that build their MM records internally (e.g. {@link syncWorkOrderBatchFromFlattened}) never expose them, so this helper only fits call sites that hold the records array they wrote.
   * @param batchCacheManager Same semantics as {@link writeEntitiesToMM} (omit for a default manager, `null` to disable caching)
   * @param options Same semantics as {@link writeEntitiesToMM}: `connectorType` (labor-ledger marking)
   *        and `queryParams` (forwarded to each individual retry write, so opt-in behaviors like
   *        `clearColumns` survive the recovery path). `chunkSize` has no effect — retries are always
   *        single-record.
   *
   * @returns `{ succeeded, failed }` — see {@link RetryRejectedIndividuallyResult}
   * @throws Error on non-validation issues (network, authentication, etc.); entities not yet retried remain unrecovered
   */
  static async retryRejectedIndividually(
    entityType: ERPObjType,
    error: MMBatchValidationError,
    originalRecords: IToRESTApiObject[],
    batchCacheManager?: BatchCacheManager | null,
    options?: { connectorType?: string; queryParams?: Record<string, string> }
  ): Promise<RetryRejectedIndividuallyResult> {
    const succeeded: IToRESTApiObject[] = [];
    const failed: Array<{ entity: IToRESTApiObject; message: string }> = [];

    // MM emits one batchErrors entry per validation message, and each entry carries
    // the entire rejected sub-batch, so the same entity can appear under several
    // entries. Dedupe by serialized payload — identical echoes serialize identically,
    // while distinct records that happen to share a primary key stay distinct — and
    // collect every validation message each entity was rejected with.
    const rejectedByPayload = new Map<
      string,
      { entity: IToRESTApiObject; key: string | null; messages: string[] }
    >();
    const unserializable = new Set<IToRESTApiObject>();
    for (const batchError of error.batchErrors) {
      for (const entity of batchError.affectedEntities) {
        let payload: string;
        try {
          payload = JSON.stringify(entity.toRESTApiObject());
        } catch (serializeError) {
          // An entity that cannot serialize can be neither deduped, matched, nor
          // re-sent; report it instead of aborting the remaining retries.
          if (!unserializable.has(entity)) {
            unserializable.add(entity);
            failed.push({
              entity,
              message: `not retried: entity could not be serialized (${
                serializeError instanceof Error
                  ? serializeError.message
                  : String(serializeError)
              })`,
            });
          }
          continue;
        }
        const existing = rejectedByPayload.get(payload);
        if (existing) {
          existing.messages.push(batchError.message);
          continue;
        }
        let key: string | null = null;
        try {
          key = EntityTransformer.extractPrimaryKey(entityType, entity);
        } catch {
          // A keyless entity can never be matched to an original; reported below.
        }
        rejectedByPayload.set(payload, {
          entity,
          key,
          messages: [batchError.message],
        });
      }
    }
    const rejected = [...rejectedByPayload.values()];
    if (rejected.length === 0 && failed.length === 0) {
      return { succeeded, failed };
    }

    // Match rejected entities back to the caller's originals by primary key. Several
    // originals can legitimately share a key (e.g. two partial patches for the same
    // operation), so keep them all and hand out one per rejected entity.
    const originalsByKey = new Map<string, IToRESTApiObject[]>();
    for (const record of originalRecords) {
      try {
        const key = EntityTransformer.extractPrimaryKey(entityType, record);
        const sameKey = originalsByKey.get(key);
        if (sameKey) {
          sameKey.push(record);
        } else {
          originalsByKey.set(key, [record]);
        }
      } catch {
        // A record whose key cannot be extracted can never be matched to a
        // rejected entity; if it was rejected it surfaces as unmatched below.
      }
    }

    logger.info(
      "retryRejectedIndividually: retrying entities from failed sub-batches one at a time",
      { entityType, rejectedCount: rejected.length }
    );

    const ownsCacheManager = batchCacheManager === undefined;
    const resolvedCacheManager =
      batchCacheManager === undefined ? new BatchCacheManager() : batchCacheManager;

    try {
      for (const { entity, key, messages } of rejected) {
        const original =
          key === null ? undefined : originalsByKey.get(key)?.shift();
        if (original === undefined) {
          // The entities in batchErrors are reconstructions of MM's echoed data and
          // default any field absent from the echo (a partial work-order-operation
          // patch would become a full row with zeroed quantities). Re-sending one
          // could overwrite real MM values, so it is never retried.
          failed.push({
            entity,
            message: `not retried: no record in originalRecords matches this rejected entity's primary key (rejected with: ${
              messages.filter(Boolean).join("; ") || error.message
            })`,
          });
          continue;
        }
        try {
          // Bypass the dedupe read: a stale cache hash must never turn a rejected
          // record into a silent no-op "success". Successes are stored below, so
          // cache state ends up the same as a normal write.
          await MMEntityProcessor.writeEntities(
            entityType,
            [original],
            null,
            options
          );
          if (resolvedCacheManager) {
            await resolvedCacheManager.storeBatch(entityType, [original]);
          }
          succeeded.push(original);
        } catch (retryError) {
          if (!(retryError instanceof MMBatchValidationError)) {
            throw retryError;
          }
          // A 1-record request can still report multiple batchErrors entries
          // (multiple validation messages for the same record) — keep them all.
          failed.push({
            entity: original,
            message:
              retryError.batchErrors
                .map((e) => e.message)
                .filter(Boolean)
                .join("; ") || retryError.message,
          });
        }
      }
    } finally {
      if (ownsCacheManager && resolvedCacheManager) {
        await resolvedCacheManager.destroy();
      }
    }

    if (failed.length > 0) {
      logger.warn(
        "retryRejectedIndividually: some entities were rejected again when retried individually; the caller must recover or loudly skip them",
        { entityType, succeededCount: succeeded.length, failedCount: failed.length }
      );
    } else {
      logger.info(
        "retryRejectedIndividually: all rejected entities imported successfully on individual retry",
        { entityType, succeededCount: succeeded.length }
      );
    }

    return { succeeded, failed };
  }

  /**
   * Writes a batch of Work Order related entities to MM maintaining referential integrity by:
   * 1. Processing parts first (base entities)
   * 2. Processing part operations (depends on parts)
   * 3. Processing work orders (depends on parts)
   * 4. Processing work order operations (depends on work orders and part operations)
   *
   * This ensures all foreign key constraints are satisfied and prevents referential integrity errors.
   *
   * @param workOrderBatch The batch containing all related entities
   * @param batchCacheManager The batch cache manager instance.
   * - Omit / pass `undefined` to use a default cache manager (recommended).
   * - Pass `null` to explicitly disable caching + local deduplication (generally not recommended; primarily useful for debugging/emergency bypass).
   *
   * @returns Combined results from all entity processing
   * @throws MMBatchValidationError on partial success or complete failure with structured error details
   * @throws Error on other underlying issues (network, authentication, etc.)
   */
  static async writeWorkOrderBatchToMM(
    workOrderBatch: MMSendWorkOrderBatch,
    batchCacheManager?: BatchCacheManager | null
  ): Promise<{
    parts: WriteEntitiesToMMResult;
    partOperations: WriteEntitiesToMMResult;
    workOrders: WriteEntitiesToMMResult;
    workOrderOperations: WriteEntitiesToMMResult;
  }> {
    const ownsCacheManager = batchCacheManager === undefined;
    const resolvedCacheManager =
      batchCacheManager === undefined ? new BatchCacheManager() : batchCacheManager;

    try {
      // Process entities in order to maintain referential integrity

      // 1. Parts first (no dependencies)
      const partsResult = await MMEntityProcessor.writeEntities(
        ERPObjType.PARTS,
        workOrderBatch.parts,
        resolvedCacheManager
      );

      // 2. Part Operations (depends on parts)
      const partOperationsResult = await MMEntityProcessor.writeEntities(
        ERPObjType.PART_OPERATION,
        workOrderBatch.partOperations,
        resolvedCacheManager
      );

      // 3. Work Orders (depends on parts)
      const workOrdersResult = await MMEntityProcessor.writeEntities(
        ERPObjType.WORK_ORDERS,
        workOrderBatch.workOrders,
        resolvedCacheManager
      );

      // 4. Work Order Operations (depends on work orders and part operations)
      const workOrderOperationsResult = await MMEntityProcessor.writeEntities(
        ERPObjType.WORK_ORDER_OPERATIONS,
        workOrderBatch.workOrderOperations,
        resolvedCacheManager
      );

      return {
        parts: partsResult,
        partOperations: partOperationsResult,
        workOrders: workOrdersResult,
        workOrderOperations: workOrderOperationsResult,
      };
    } finally {
      if (ownsCacheManager && resolvedCacheManager) {
        await resolvedCacheManager.destroy();
      }
    }
  }

  /**
   * Groups flattened work-order rows into the four MM entity sets and constructs the MM
   * send objects, without writing anything. This is the mapping half of
   * {@link syncWorkOrderBatchFromFlattened}; connectors call it directly when they need the
   * batch for something other than the regular sync, e.g. as the return value of the
   * `fetchWorkOrders` callback of {@link fillWorkOrderGapsFromLaborTickets}.
   *
   * Throws when any work order or operation lacks a non-blank string status (a connector
   * mapping bug), before anything is constructed.
   */
  static buildWorkOrderBatchFromFlattened(
    flattenedData: ReadonlyArray<FlattenedWorkOrderRow>
  ): MMSendWorkOrderBatch {
    const toStringOrFallback = (
      value: string | number | null | undefined,
      fallback = ""
    ): string => (value === undefined || value === null ? fallback : String(value));
    const isNonBlankString = (value: unknown): value is string =>
      typeof value === "string" && value.trim().length > 0;
    const isAbsentOrBlank = (value: unknown): boolean =>
      value === undefined ||
      value === null ||
      (typeof value === "string" && value.trim().length === 0);

    // Deduplicated entities take their scalar fields from the first row seen. Custom fields
    // instead resolve across all of the entity's rows with precedence value > null > undefined,
    // so a row where an optional join produced nothing cannot mask (or, via null, clear) the
    // value another row carries, and row order only matters between differing non-null values.
    // The pair resolves atomically: the winning row supplies both the values and the schema ref
    // they were written against, so the two can never be assembled from rows that disagree.
    // Every row runs through this, including the one that creates the entity, so a row carrying
    // only a schema ref contributes nothing no matter where it sits in the batch.
    const customFieldsPrecedence = (
      customFields: Record<string, unknown> | null | undefined
    ): number => (customFields === undefined ? 0 : customFields === null ? 1 : 2);

    const mergeCustomFields = (
      target: {
        customFieldsSchemaRef?: number | null;
        customFields?: Record<string, unknown> | null;
      },
      schemaRef: number | null | undefined,
      customFields: Record<string, unknown> | null | undefined
    ): void => {
      // Strictly greater, so the first row wins ties and keeps its pair intact.
      if (
        customFieldsPrecedence(customFields) <=
        customFieldsPrecedence(target.customFields)
      ) {
        return;
      }
      target.customFields = customFields;
      target.customFieldsSchemaRef = schemaRef;
    };

    // Process the flattened data - each row contains both work order and operation info
    const uniqueParts = new Map();
    const uniquePartOperations = new Map();
    const uniqueWorkOrders = new Map();
    const workOrderOperations = [];

    // Process each flattened row to extract entities
    for (const row of flattenedData) {
      // Extract unique parts.
      //
      // Both part keys are built from the exact normalizers the constructors apply below, so
      // rows that construct the same MM part share one entry. Interpolating raw row values
      // let `method` NULL vs "Standard", or `partRevision` NULL vs "", split one part into
      // two entries. That was harmless while the two payloads were identical; with custom
      // fields resolved per entry it sends two payloads for one part key whose custom fields
      // disagree, and the later one (possibly null) wins on row order.
      const partIdentity = mmPartIdentityKey(row);
      const partKey = partIdentity;
      if (!uniqueParts.has(partKey)) {
        uniqueParts.set(partKey, {
          partNumber: row.partNumber, // → partNumber
          partRevision: row.partRevision, // → partRevision
          method: row.method, // → method
        });
      }
      mergeCustomFields(
        uniqueParts.get(partKey),
        row.partCustomFieldsSchemaRef,
        row.partCustomFields
      );

      // Extract unique part operations (same normalization as the part key; see above).
      const partOpKey = `${partIdentity}|${row.sequenceNumber?.toString() || ""}`;
      if (!uniquePartOperations.has(partOpKey)) {
        uniquePartOperations.set(partOpKey, {
          partNumber: row.partNumber, // → partNumber
          partRevision: row.partRevision, // → partRevision
          method: row.method, // → method
          sequenceNumber: row.sequenceNumber, // → sequenceNumber
          resourceId: row.resourceId, // → resourceId
          cycleTimeMs: row.cycleTimeMs, // → cycleTimeMs
          setupTimeMs: row.setupTimeMs, // → setupTimeMs
          operationDescription: row.operationDescription, // → description
          quantityPerPart: row.quantityPerPart ?? 1, // → quantityPerPart
          partName: row.partName, // → partName
        });
      }
      mergeCustomFields(
        uniquePartOperations.get(partOpKey),
        row.partOperationCustomFieldsSchemaRef,
        row.partOperationCustomFields
      );

      // Extract unique work orders.
      //
      // The key must be MM's full work order identity (workOrderId + lot + split + sub, see
      // EntityTransformer.extractWorkOrderKey), not the id alone. Keying on the id collapses
      // two lots of one job into a single entity: the second lot's work order is never sent,
      // while its operations are, and `erp.work_order_operation.work_order_fk` then rejects
      // them against a parent that does not exist. Retrying cannot clear that, because the
      // parent is never created on this path, so the batch fails permanently.
      //
      // Components are normalized exactly as the MMSendWorkOrder constructor normalizes them,
      // so rows that build the same MM entity share a key. Without that, an unset lot and an
      // empty-string lot (or a numeric and a string id) would split one entity into two
      // conflicting payloads.
      const workOrderKey = [
        toStringOrFallback(row.workOrderId),
        toStringOrFallback(row.lot),
        toStringOrFallback(row.split),
        toStringOrFallback(row.sub),
      ].join("|");
      if (!uniqueWorkOrders.has(workOrderKey)) {
        uniqueWorkOrders.set(workOrderKey, {
          workOrderId: row.workOrderId, // → workOrderId
          lot: row.lot, // → lot
          split: row.split, // → split
          sub: row.sub, // → sub
          status: row.status, // → status
          dueDate: row.dueDate, // → dueDate
          description: row.description, // → description
          scheduledStartDate: row.scheduledStartDate, // → scheduledStartDate
          scheduledEndDate: row.scheduledEndDate, // → scheduledEndDate
          closedDate: row.closedDate, // → closedDate
          quantityRequired: row.quantityRequired, // → quantityRequired
          partNumber: row.partNumber, // → partNumber
          partRevision: row.partRevision, // → partRevision
          method: row.method, // → method
        });
      }
      mergeCustomFields(
        uniqueWorkOrders.get(workOrderKey),
        row.workOrderCustomFieldsSchemaRef,
        row.workOrderCustomFields
      );

      // Add each work order operation (these are not deduplicated since each operation is unique)
      const opCustomFields: {
        customFieldsSchemaRef?: number | null;
        customFields?: Record<string, unknown> | null;
      } = {};
      mergeCustomFields(
        opCustomFields,
        row.opCustomFieldsSchemaRef,
        row.opCustomFields
      );

      workOrderOperations.push({
        workOrderId: row.workOrderId, // → workOrderId
        lot: row.lot, // → lot
        split: row.split, // → split
        sub: row.sub, // → sub
        sequenceNumber: row.sequenceNumber, // → sequenceNumber
        resourceId: row.resourceId, // → resourceId
        startQuantity: row.startQuantity, // → startQuantity
        finishQuantity: row.finishQuantity, // → finishQuantity
        expectedRejectRate: row.expectedRejectRate, // → expectedRejectRate
        scheduledStartDate: row.opScheduledStartDate, // → scheduledStartDate
        scheduledFinishDate: row.opScheduledFinishDate, // → scheduledFinishDate
        closedDate: row.opClosedDate, // → closedDate
        cycleTimeMs: row.cycleTimeMs, // → cycleTimeMs
        setupTimeMs: row.setupTimeMs, // → setupTimeMs
        productionburdenRateHourly: row.productionburdenRateHourly, // → productionburdenRateHourly
        setupburdenRatehourly: row.setupburdenRatehourly, // → setupburdenRatehourly
        operationType: row.operationType, // → operationType
        quantityPerPart: row.quantityPerPart, // → quantityPerPart
        // Operation status falls back to the parent status only when opStatus is absent
        // or a blank string (a padded CHAR column or an empty CASE branch must not mask
        // a valid parent value). Any other non-string value (e.g. a raw numeric ERP
        // code) is kept so the validation below fails the batch instead of masking
        // the mapping bug behind the parent's status.
        status: isAbsentOrBlank(row.opStatus) ? row.status : row.opStatus, // → status
        // One row is one operation, so there is nothing to merge, but the pair still goes
        // through the same helper: a row carrying only a schema ref must be a no-op here
        // exactly as it is on the deduplicated paths.
        ...opCustomFields, // → customFieldsSchemaRef / customFields
      });
    }

    // A row without a status is a connector mapping bug (NULL branch in a SQL CASE, a
    // renamed column, …), and rows come from one query, so it is systemic: fail the
    // batch before any write rather than invent a status. RELEASED/ACTIVE would make
    // phantom rows schedulable and reconciliation candidates; NOT RELEASED would be
    // harmless in MM but hide the bug indefinitely.
    // Require a non-blank *string*: a numeric or boolean status (e.g. a raw ERP code 0)
    // is as much a mapping bug as a missing one and must not reach MM as "0".
    const workOrdersMissingStatus = Array.from(uniqueWorkOrders.values()).filter(
      (item) => !isNonBlankString(item.status)
    );
    const operationsMissingStatus = workOrderOperations.filter(
      (item) => !isNonBlankString(item.status)
    );
    if (workOrdersMissingStatus.length > 0 || operationsMissingStatus.length > 0) {
      const sample = [
        ...workOrdersMissingStatus.map(
          (item) => `work order ${item.workOrderId}|${item.lot ?? ""}|${item.split ?? ""}|${item.sub ?? ""}`
        ),
        ...operationsMissingStatus.map(
          (item) => `operation ${item.workOrderId}|${item.lot ?? ""}|${item.split ?? ""}|${item.sub ?? ""}/${item.sequenceNumber ?? ""}`
        ),
      ].slice(0, 5);
      throw new Error(
        `buildWorkOrderBatchFromFlattened: ${workOrdersMissingStatus.length} work order(s) and ${operationsMissingStatus.length} work order operation(s) lack a non-blank string status (e.g. ${sample.join("; ")}); every row must carry a status / opStatus — see MMWorkOrderStatus and MMWorkOrderOperationStatus. Nothing was written.`
      );
    }

    // Transform all the data directly to MM entities
    const parts = Array.from(uniqueParts.values()).map(
      (item) =>
        new MMSendPart(
          toStringOrFallback(item.partNumber), // partNumber
          toStringOrFallback(item.partRevision), // partRevision
          item.method || "Standard", // method
          item.customFieldsSchemaRef, // customFieldsSchemaRef
          item.customFields // customFields
        )
    );

    const partOperations = Array.from(uniquePartOperations.values()).map(
      (item) =>
        new MMSendPartOperation(
          toStringOrFallback(item.partNumber), // partNumber
          toStringOrFallback(item.partRevision), // partRevision
          item.method || "Standard", // method
          item.sequenceNumber?.toString() || "", // sequenceNumber
          item.resourceId?.toString() || "", // resourceId
          item.cycleTimeMs || 0, // cycleTimeMs
          item.setupTimeMs || 0, // setupTimeMs
          item.operationDescription || "", // description
          item.quantityPerPart || 1, // quantityPerPart
          item.customFieldsSchemaRef, // customFieldsSchemaRef
          item.customFields, // customFields
          item.partName // partName
        )
    );

    const transformedWorkOrders = Array.from(uniqueWorkOrders.values()).map(
      (item) =>
        new MMSendWorkOrder(
          item.workOrderId?.toString() || "", // workOrderId
          toStringOrFallback(item.lot), // lot
          toStringOrFallback(item.split), // split
          toStringOrFallback(item.sub), // sub
          toStringOrFallback(item.status), // status (validated non-empty above; see MMWorkOrderStatus)
          // Datetimes are assumed to already be UTC (or include an explicit offset).
          // We reserialize via Date solely to validate the ISO format before sending to MM.
          item.dueDate ? new Date(item.dueDate).toISOString() : null, // dueDate
          item.description || "", // description
          item.scheduledStartDate
            ? new Date(item.scheduledStartDate).toISOString()
            : null, // scheduledStartDate
          item.scheduledEndDate
            ? new Date(item.scheduledEndDate).toISOString()
            : null, // scheduledEndDate
          item.closedDate ? new Date(item.closedDate).toISOString() : null, // closedDate
          item.quantityRequired || 0, // quantityRequired
          toStringOrFallback(item.partNumber), // partNumber
          toStringOrFallback(item.partRevision), // partRevision
          item.method || "Standard", // method
          item.customFieldsSchemaRef, // customFieldsSchemaRef
          item.customFields // customFields
        )
    );

    const transformedWorkOrderOperations = workOrderOperations.map(
      (item) =>
        new MMSendWorkOrderOperation(
          item.workOrderId?.toString() || "", // workOrderId
          toStringOrFallback(item.lot), // lot
          toStringOrFallback(item.split), // split
          toStringOrFallback(item.sub), // sub
          item.sequenceNumber?.toString() || "", // sequenceNumber
          item.resourceId?.toString() || "", // resourceId
          item.startQuantity || 0, // startQuantity
          item.finishQuantity || 0, // finishQuantity
          item.expectedRejectRate || 0, // expectedRejectRate
          // Same UTC expectation as above; reserialize to ensure valid ISO.
          item.scheduledStartDate
            ? new Date(item.scheduledStartDate).toISOString()
            : null, // scheduledStartDate
          item.scheduledFinishDate
            ? new Date(item.scheduledFinishDate).toISOString()
            : null, // scheduledFinishDate
          item.closedDate ? new Date(item.closedDate).toISOString() : null, // closedDate
          item.cycleTimeMs || 0, // cycleTimeMs
          item.setupTimeMs || 0, // setupTimeMs
          parseFloat(toStringOrFallback(item.productionburdenRateHourly, "0")), // productionburdenRateHourly
          parseFloat(toStringOrFallback(item.setupburdenRatehourly, "0")), // setupburdenRatehourly
          item.operationType || "Production", // operationType
          item.quantityPerPart || 1, // quantityPerPart
          toStringOrFallback(item.status), // status (validated non-empty above; see MMWorkOrderOperationStatus)
          item.customFieldsSchemaRef, // customFieldsSchemaRef
          item.customFields // customFields
        )
    );

    return {
      parts,
      partOperations,
      workOrders: transformedWorkOrders,
      workOrderOperations: transformedWorkOrderOperations,
    };
  }

  /**
   * Processes flattened work order data and syncs it to MM maintaining referential integrity.
   *
   * This method expects flattened data where each row contains both work order and operation information
   * with camelCase field names matching the GraphQL schema. It will automatically extract and deduplicate
   * parts, part operations, work orders, and work order operations, then process them in the correct order
   * to maintain referential integrity.
   *
   * Timezone expectation: all datetime fields provided in `flattenedData` MUST already be expressed
   * as ISO-8601 UTC strings (trailing `Z` or explicit offset). The SDK does not attempt to infer or
   * convert local ERP timestamps on this path; it simply validates and forwards the values to MM.
   * Connector implementations should convert ERP-local times to UTC before invoking this method.
   *
   * Custom fields: per-entity `*CustomFields` / `*CustomFieldsSchemaRef` row properties are
   * forwarded to the matching MM entity (see `FlattenedWorkOrderRow`). For deduplicated entities
   * (parts, part operations, work orders) a non-null value on any of the entity's rows beats
   * `null`, which beats `undefined`; between differing non-null values the first row wins, and
   * the schema ref is taken from the row that supplied the values.
   *
   * @param flattenedData Array of flattened rows containing both work order and operation data (see `FlattenedWorkOrderRow`)
   * @param batchCacheManager The batch cache manager instance.
   * - Omit / pass `undefined` to use a default cache manager (recommended).
   * - Pass `null` to explicitly disable caching + local deduplication (generally not recommended; primarily useful for debugging/emergency bypass).
   *
   * @returns Combined results from all entity processing with detailed logging information
   * @throws MMBatchValidationError on partial success or complete failure with structured error details
   * @throws Error on other underlying issues (network, authentication, etc.)
   */
  static async syncWorkOrderBatchFromFlattened(
    flattenedData: ReadonlyArray<FlattenedWorkOrderRow>,
    batchCacheManager?: BatchCacheManager | null
  ): Promise<{
    parts: WriteEntitiesToMMResult;
    partOperations: WriteEntitiesToMMResult;
    workOrders: WriteEntitiesToMMResult;
    workOrderOperations: WriteEntitiesToMMResult;
  }> {
    if (!flattenedData || flattenedData.length === 0) {
      throw new Error("No flattened work order data provided");
    }

    const batch = this.buildWorkOrderBatchFromFlattened(flattenedData);
    return this.writeWorkOrderBatchToMM(batch, batchCacheManager);
  }

  /**
   * Supplementary work-order import driven by labor tickets: fills the operations that the
   * cycle's labor tickets reference but the windowed bulk work-order sync did not carry, so
   * {@link writeLaborTicketsToMmFromErpImport} accepts those tickets instead of rejecting
   * them as `notInSyncedWorkOrderOperation` (or MM rejecting them on
   * `work_order_operation_fk`). See `labor-work-order-gap-fill.ts` for the policy.
   *
   * Per cycle, before the labor write:
   *  1. plan — work order ids referenced by OPEN tickets with no bulk-synced operation
   *     (every state with `includeClosedTickets`), plus the persisted watch list of
   *     previously filled work orders (re-fetched until their terminal state is imported);
   *  2. fetch — `options.fetchWorkOrders(ids)` (connector: ERP query + mapping to an
   *     {@link MMSendWorkOrderBatch}; `buildWorkOrderBatchFromFlattened` helps connectors
   *     with flattened rows);
   *  3. write — the FK-complete subset that has no competing writer this cycle
   *     ({@link computeWorkOrderGapFillWriteScope}), through `options.writeBatch` when given
   *     and {@link writeWorkOrderBatchToMM} otherwise;
   *  4. return every operation of the fetched work orders as `keySetWorkOrderOps`, for the
   *     connector to concatenate with the bulk rows in
   *     `FilterLaborTicketsForMmImportOptions.syncedWorkOrderOperations`.
   *
   * Throws on a fetch or write failure, or when the watch list cannot be read (before
   * fetching anything, so the persisted list is never overwritten from an empty one), like
   * every SDK driver. A connector that prefers the
   * labor write to proceed without the widened key set wraps the call and treats the
   * failure as best-effort; the watch list is persisted before the write, so a terminal
   * work order whose close-out write failed is fetched again next cycle either way.
   *
   * This path is additive to the connector's regular work-order sync: labor tickets do not
   * describe the shop schedule. Work orders it writes sit outside the bulk window by
   * definition, so a `IERPWorkOrderReconciliationHandler` must judge deletion by ERP
   * state, not by bulk-window membership (see the reconciliation section of
   * `CONNECTOR_SDK_UTILITIES.md`); the returned `trackedWorkOrderIds` are the ids to
   * re-check explicitly.
   */
  static async fillWorkOrderGapsFromLaborTickets(
    options: FillWorkOrderGapsFromLaborTicketsOptions
  ): Promise<FillWorkOrderGapsFromLaborTicketsResult> {
    const {
      connectorType,
      laborTickets,
      bulkSyncedWorkOrderOperations,
      fetchWorkOrders,
      batchCacheManager,
    } = options;
    if (typeof fetchWorkOrders !== "function") {
      throw new Error("fillWorkOrderGapsFromLaborTickets: options.fetchWorkOrders must be a function");
    }
    const isTerminal = options.isTerminalWorkOrder ?? isTerminalWorkOrderForGapFill;
    const tracker = options.tracker ?? new LaborWorkOrderGapFillTracker(connectorType);
    const now = options.now ?? new Date();
    const warnThreshold =
      options.fetchWarnThreshold ?? DEFAULT_GAP_FILL_FETCH_WARN_THRESHOLD;

    const bulkOps = bulkSyncedWorkOrderOperations ?? [];
    const openLaborWorkOrderIds = findOpenLaborWorkOrderIdsMissingOps(laborTickets ?? [], bulkOps);
    const anyStateLaborWorkOrderIds = findLaborWorkOrderIdsMissingOps(laborTickets ?? [], bulkOps);
    const openUpper = new Set(openLaborWorkOrderIds.map((id) => id.toUpperCase()));
    const closedOnlyLaborWorkOrderIds = anyStateLaborWorkOrderIds.filter(
      (id) => !openUpper.has(id.toUpperCase())
    );
    const laborWorkOrderIds = options.includeClosedTickets
      ? anyStateLaborWorkOrderIds
      : openLaborWorkOrderIds;
    if (closedOnlyLaborWorkOrderIds.length > 0 && !options.includeClosedTickets) {
      logger.warn(
        `fillWorkOrderGapsFromLaborTickets: ${closedOnlyLaborWorkOrderIds.length} work order(s) are referenced only by closed tickets with no bulk-synced operation and will not be fetched (includeClosedTickets is off), so those tickets will be rejected this cycle: ${previewIds(closedOnlyLaborWorkOrderIds)}`
      );
    }
    const missingOpKeys = findMissingWorkOrderOpKeys(laborTickets ?? [], bulkOps);

    const tracked = tracker.load();
    const plan = planWorkOrderGapFillFetch({
      tracked,
      laborWorkOrderIds,
      now,
      trackedTtlDays: options.trackedTtlDays,
    });

    const emptyResult = (
      trackedAfter: ReadonlyArray<TrackedLaborReferencedWorkOrder>,
      dropped: FillWorkOrderGapsFromLaborTicketsResult["dropped"],
      unresolvedWorkOrderIds: string[] = []
    ): FillWorkOrderGapsFromLaborTicketsResult => ({
      keySetWorkOrderOps: [],
      openLaborWorkOrderIds,
      closedOnlyLaborWorkOrderIds,
      fetchedWorkOrderIds: plan.fetchWorkOrderIds,
      unresolvedWorkOrderIds,
      writtenBatch: null,
      writeResult: null,
      closingWorkOrderIds: [],
      trackedWorkOrderIds: trackedAfter.map((t) => t.workOrderId),
      dropped,
    });

    if (plan.fetchWorkOrderIds.length === 0) {
      if (tracked.length > 0) {
        // Persist the pruning of expired entries.
        tracker.save(plan.carriedTracked);
      }
      return emptyResult(plan.carriedTracked, {
        ...plan.dropped,
        coveredByBulkSync: 0,
        terminal: 0,
        evicted: 0,
      });
    }

    if (plan.fetchWorkOrderIds.length > warnThreshold) {
      logger.warn(
        `fillWorkOrderGapsFromLaborTickets: fetching ${plan.fetchWorkOrderIds.length} work order(s) (over ${warnThreshold}); unusually many open labor tickets reference work orders outside the bulk sync window`
      );
    }
    logger.info("fillWorkOrderGapsFromLaborTickets: fetching work orders", {
      count: plan.fetchWorkOrderIds.length,
      fromOpenLabor: openLaborWorkOrderIds.length,
      fromClosedOnlyLabor: options.includeClosedTickets ? closedOnlyLaborWorkOrderIds.length : 0,
      trackedForCloseOut: plan.carriedTracked.length,
      workOrderIds: previewIds(plan.fetchWorkOrderIds),
    });

    const fetched = await fetchWorkOrders([...plan.fetchWorkOrderIds]);
    const fullBatch = requireWorkOrderBatch(fetched);

    // The watch list and the write scope work per work order id, but MM's work order
    // identity is id + lot + split + sub and the connector may have fetched several. An id
    // is terminal only when every fetched row of it is: one completed lot must not close
    // out (and rewrite) the sibling lots operators are still clocked into. Grouped
    // case-insensitively, as the watch list keys ids: a job spelled two ways across its
    // lots is terminal only when every row under every spelling is, and then each spelling
    // is handed to the write scope, which judges row coverage by exact id.
    const rowsByWorkOrderId = new Map<string, { rows: MMSendWorkOrder[]; spellings: Set<string> }>();
    for (const wo of fullBatch.workOrders) {
      const id = String(wo.workOrderId ?? "").trim();
      if (!id) continue;
      const key = id.toUpperCase();
      const group = rowsByWorkOrderId.get(key) ?? { rows: [], spellings: new Set<string>() };
      group.rows.push(wo);
      group.spellings.add(id);
      rowsByWorkOrderId.set(key, group);
    }
    const terminalWorkOrderIds = [...rowsByWorkOrderId.values()]
      .filter(({ rows }) => rows.every((wo) => isTerminal(wo)))
      .flatMap(({ spellings }) => [...spellings]);

    // Requested ids the connector returned no work order for. They stay on the watch list
    // until the TTL (open labor keeps refreshing them while it lasts): the ERP may hide a
    // job transiently, and dropping one on absence would let a job that closes during the
    // gap keep a stale status in MM for good. Surfaced so the connector can log or act.
    const unresolvedWorkOrderIds = plan.fetchWorkOrderIds.filter(
      (id) => !rowsByWorkOrderId.has(id.toUpperCase())
    );
    if (unresolvedWorkOrderIds.length > 0) {
      logger.warn(
        `fillWorkOrderGapsFromLaborTickets: fetchWorkOrders returned no work order for ${unresolvedWorkOrderIds.length} requested id(s): ${previewIds(unresolvedWorkOrderIds)}`
      );
    }

    if (fullBatch.workOrderOperations.length === 0) {
      // Not a short-circuit: a closing work order with no operations still has its work
      // order row to write, which the scope below handles.
      logger.warn(
        `fillWorkOrderGapsFromLaborTickets: fetchWorkOrders returned no work order operations for ${plan.fetchWorkOrderIds.length} work order(s): ${previewIds(plan.fetchWorkOrderIds)}`
      );
    }

    // Pure, and computed before the watch-list update because it also says which fetched
    // ids the bulk sync now covers row for row with nothing left for the gap fill to write.
    const scope = computeWorkOrderGapFillWriteScope({
      fullBatch,
      bulkSyncedWorkOrderOperations: bulkOps,
      missingOpKeys,
      terminalWorkOrderIds,
    });

    // Update the watch list before any early return: refresh open-labor entries, drop work
    // orders the bulk sync covers row for row again (unless a gap write for them is
    // pending this cycle, which the entry retries if it fails), close out work orders that came back
    // terminal, evict beyond the cap. Terminal entries stay persisted until their close-out
    // write lands: they sit outside the bulk window, so one dropped before a failed write is
    // never fetched again and its MM work order keeps a stale status permanently.
    const next = updateTrackedAfterWorkOrderGapFillFetch({
      carriedTracked: plan.carriedTracked,
      laborWorkOrderIds,
      terminalWorkOrderIds,
      bulkCoveredWorkOrderIds: scope.bulkCoveredWorkOrderIds,
      now,
      maxTrackedWorkOrders: options.maxTrackedWorkOrders,
    });
    // Persisted now, with the terminal entries retained (see above); `pendingTracked` is
    // what `trackedWorkOrderIds` reports until the close-out write lands. Pending close-outs
    // are exempt from the cap and take room from the open entries: losing one costs a
    // permanently stale status, while a dropped open entry is re-added by labor.
    const cap = options.maxTrackedWorkOrders ?? DEFAULT_MAX_TRACKED_WORK_ORDERS;
    const pendingTracked = capTrackedList(next.terminal, next.tracked, cap);
    tracker.save(pendingTracked);
    const dropped = {
      ...plan.dropped,
      ...next.dropped,
      evicted:
        next.dropped.evicted + (next.terminal.length + next.tracked.length - pendingTracked.length),
    };
    logger.info("fillWorkOrderGapsFromLaborTickets: watch list updated", {
      tracking: next.tracked.length,
      closingOut: next.dropped.terminal,
      ...dropped,
    });

    const nothingToWrite =
      scope.batchToWrite.workOrderOperations.length === 0 &&
      scope.batchToWrite.workOrders.length === 0 &&
      scope.batchToWrite.partOperations.length === 0 &&
      scope.batchToWrite.parts.length === 0;
    if (nothingToWrite) {
      // Not an error: the work orders were fetched because labor referenced them, and their
      // operations still widen this cycle's labor filter key set.
      logger.info(
        `fillWorkOrderGapsFromLaborTickets: fetched ${fullBatch.workOrders.length} work order(s) carrying ${fullBatch.workOrderOperations.length} operation(s); none needed writing (no missing key, no close-out). Their keys still widen this cycle's labor filter.`
      );
      return {
        ...emptyResult(pendingTracked, dropped, unresolvedWorkOrderIds),
        keySetWorkOrderOps: scope.keySetWorkOrderOps,
      };
    }

    const writeResult = options.writeBatch
      ? await options.writeBatch(scope.batchToWrite)
      : await this.writeWorkOrderBatchToMM(scope.batchToWrite, batchCacheManager);
    logger.info("fillWorkOrderGapsFromLaborTickets: wrote gap-fill batch", {
      workOrderOperations: scope.batchToWrite.workOrderOperations.length,
      workOrders: scope.batchToWrite.workOrders.length,
      parts: scope.batchToWrite.parts.length,
      partOperations: scope.batchToWrite.partOperations.length,
      ofFetchedOperations: fullBatch.workOrderOperations.length,
      closingOut: scope.closingWorkOrderIds.length,
      outsideBulkWindow: scope.uncoveredWorkOrderIds.length,
      partsSharedWithBulkSync: scope.sharedPartIdentities.length,
    });

    // The close-out landed, so the terminal work orders can finally leave the watch list.
    if (next.terminal.length > 0) {
      tracker.save(next.tracked);
    }

    return {
      keySetWorkOrderOps: scope.keySetWorkOrderOps,
      openLaborWorkOrderIds,
      closedOnlyLaborWorkOrderIds,
      fetchedWorkOrderIds: plan.fetchWorkOrderIds,
      unresolvedWorkOrderIds,
      writtenBatch: scope.batchToWrite,
      writeResult,
      closingWorkOrderIds: scope.closingWorkOrderIds,
      trackedWorkOrderIds: next.tracked.map((t) => t.workOrderId),
      dropped,
    };
  }
}

const ID_LOG_PREVIEW_LIMIT = 20;

/** Capped, log-friendly rendering of an id list: "A, B, C (+N more)". */
function previewIds(ids: ReadonlyArray<string>): string {
  if (ids.length <= ID_LOG_PREVIEW_LIMIT) return ids.join(", ");
  return `${ids.slice(0, ID_LOG_PREVIEW_LIMIT).join(", ")} (+${ids.length - ID_LOG_PREVIEW_LIMIT} more)`;
}

/**
 * The persisted watch list: every pending close-out, then the open entries (already
 * newest-first-capped by the update step) in whatever room the cap leaves. Pending
 * close-outs are never truncated — they are bounded by one cycle's fetch, and one dropped
 * before its write lands is never fetched again.
 */
function capTrackedList(
  terminal: ReadonlyArray<TrackedLaborReferencedWorkOrder>,
  tracked: ReadonlyArray<TrackedLaborReferencedWorkOrder>,
  cap: number
): TrackedLaborReferencedWorkOrder[] {
  if (terminal.length === 0) return [...tracked];
  return [...tracked.slice(0, Math.max(cap - terminal.length, 0)), ...terminal];
}

/** A `fetchWorkOrders` return value that is not a batch is a connector bug; fail loudly. */
function requireWorkOrderBatch(value: unknown): MMSendWorkOrderBatch {
  const batch = value as Partial<MMSendWorkOrderBatch> | null | undefined;
  const missing = (["parts", "partOperations", "workOrders", "workOrderOperations"] as const).filter(
    (field) => !Array.isArray(batch?.[field])
  );
  if (!batch || typeof batch !== "object" || missing.length > 0) {
    throw new Error(
      `fillWorkOrderGapsFromLaborTickets: fetchWorkOrders must resolve to an MMSendWorkOrderBatch with array fields parts, partOperations, workOrders, workOrderOperations (missing or non-array: ${missing.join(", ") || "all"})`
    );
  }
  return batch as MMSendWorkOrderBatch;
}