import { IERPLaborTicketHandler } from "../../types/erp-connector.js";
import { MMApiClient } from "../../services/mm-api-service/mm-api-service.js";
import { MMReceiveLaborTicket } from "../../services/mm-api-service/types/receive-types.js";
import { convertLaborTicketToLocalTimezone } from "../mm-labor-ticket-helpers.js";
import { sdkLogger as logger } from "../../services/reporting-service/logger.js";
import { summarizeError } from "../error-utils.js";
import { LaborTicketSyncLedgerManager } from "../../services/caching-service/labor-ticket-sync-ledger-manager.js";
import {
  normalizeTimestamp,
  normalizeString,
  normalizeNumber,
  normalizeLaborTicketRef,
  normalizeReasons,
  tryLaborTicketLedgerErpImportKey,
} from "./labor-ticket-normalization.js";
import { LaborTicketRecovery } from "../labor-ticket-recovery.js";
import { normalizeLaborTicketWarnings } from "../labor-ticket-warnings.js";

/**
 * Handles synchronization of labor tickets between MachineMetrics and ERP systems
 */
export class LaborTicketERPSynchronizer {
  // Small allowance to mitigate tiny clock offsets / timestamp granularity differences
  // between the connector host and MM's timestamps.
  private static readonly CHECKPOINT_SKEW_MS = 2_000;

  /**
   * Prefetch any labor tickets from the failed queue refs so we can remove them from the
   * retry queue if they are successfully processed in this run.
   */
  private static async prefetchFailedLaborTicketQueueRefs(
    mmApiClient: MMApiClient,
    connectorType: string
  ): Promise<Set<string> | null> {
    // In most syncToERP cases, a successfully processed ticket won't be on the retry queue.
    // Prefetch the queue once so we only attempt removals when necessary.
    try {
      const failedQueueTickets =
        await mmApiClient.fetchFailedLaborTickets(connectorType);
      return new Set(
        failedQueueTickets
          .map((t) => t.laborTicketRef)
          .filter(
            (ref) => ref !== null && ref !== undefined && String(ref) !== ""
          )
          .map((ref) => String(ref))
      );
    } catch (error) {
      // Non-fatal: continue syncing; we just can't be sure what's already queued.
      logger.warn(
        "syncLaborTicketsToERP: Unable to prefetch failed labor ticket queue; skipping retry-queue cleanup for this run.",
        { error: summarizeError(error) }
      );
      return null;
    }
  }

  /**
   * Compute the successful labor ticket refs to remove from the retry queue.
   */
  private static computeSuccessRefsToRemoveFromRetryQueue(options: {
    failedLaborTicketRefs: ReadonlySet<string>;
    prefetchedFailedQueueRefs: ReadonlySet<string> | null;
    outcomeByLaborTicketRef: ReadonlyMap<string, "success" | "failure">;
  }): string[] {
    const {
      failedLaborTicketRefs,
      prefetchedFailedQueueRefs,
      outcomeByLaborTicketRef,
    } = options;

    // If a ticket succeeded during syncToERP, it may already be on the retry queue from a prior run.
    // In most cases it won't be, so we "check first" using the prefetch.
    // We also include tickets that failed during this run as candidates (since we may have just queued them).
    const candidateRetryQueueRefs = new Set<string>([
      ...failedLaborTicketRefs,
      ...(prefetchedFailedQueueRefs
        ? Array.from(prefetchedFailedQueueRefs)
        : []),
    ]);

    return Array.from(outcomeByLaborTicketRef.entries())
      .filter(([, outcome]) => outcome === "success")
      .map(([ref]) => ref)
      .filter((ref) => candidateRetryQueueRefs.has(ref));
  }

  /**
   * Synchronizes updated labor tickets from MachineMetrics to an ERP system.
   *
   * Complications (why this is more than "fetch updates and post"):
   * - MM's /export/updates feed is driven by timestamps (e.g. updatedAt) which may change for reasons
   *   that are not "business-relevant changes" (internal touches or background processes, but especially
   *   our write-back).
   * - Our own MM write-back (esp. setting laborTicketId on creates) can bump MM timestamps and cause
   *   the same ticket to reappear on the updates feed.
   * - Connector host time and MM time can differ (clock skew / timestamp precision). A ticket can look
   *   "in the future" relative to the connector host.
   * - We still need bounded runs and safe checkpointing: updates can arrive mid-run, and the feed can
   *   be reordered, so checkpointing must avoid gaps.
   *
   * Solution approach (high level):
   * - Persist a local SQLite ledger keyed by (system, laborTicketRef) with the last attempted
   *   "business signature" (excludes updatedAt, laborTicketId, etc.). This prevents re-processing
   *   substantially identical tickets across cycles, even if MM timestamps would otherwise allow it.
   * - Also keep in-run dedupe by a stable business signature so a single run doesn't re-attempt the same
   *   substantive state.
   * - Use a barrier time + checkpoint clamp for bounded processing and to reduce checkpoint gaps; but
   *   do not skip tickets solely due to "future vs barrier" timestamps (we only log the comparison).
   * - Provide per-ticket decision logging to explain why each ticket was processed or skipped.
   */
  static async syncToERP(
    connectorType: string,
    connector: IERPLaborTicketHandler,
    options?: {
      onlyClosed?: boolean; // true: process only CLOSED labor tickets from MM; false (default): process all labor tickets from MM
    }
  ): Promise<void> {
    try {
      const mmApiClient = new MMApiClient();
      try {
        const ledger = new LaborTicketSyncLedgerManager();
        const failedLaborTicketRefs = new Set<string>();
        const attemptedSignatures = new Set<string>();
        const outcomeByLaborTicketRef = new Map<
          string,
          "success" | "failure"
        >();
        const decisionState = LaborTicketSyncDecisionTracker.init();

        try {
          const prefetchedFailedQueueRefs =
            await this.prefetchFailedLaborTicketQueueRefs(
              mmApiClient,
              connectorType
            );

          // Initialize and fetch checkpoint (quick operations, don't need prolonged lock)
          await mmApiClient.initializeCheckpoint({
            system: connectorType,
            table: "labor_tickets",
            checkpointType: "export",
            checkpointValue: {
              timestamp: new Date().toISOString(),
            },
          });

          /**
           * Checkpointing + dedupe notes:
           * - We cannot safely checkpoint to "now" without bounded draining, because updates can arrive mid-run.
           * - We cannot dedupe by (laborTicketRef, updatedAt) because MM timestamps can change for non-business
           *   reasons (including our own write-backs).
           * - We therefore:
           *   - bound each run using a barrier time and a checkpoint clamp; and
           *   - rely primarily on the persistent SQLite ledger + business signature to avoid reprocessing
           *     substantially identical tickets across cycles.
           */
          const maxPasses = 10;
          let pass = 0;

          // Barrier time: the latest time we’re willing to advance the checkpoint to for this pass
          let barrierTime = new Date().toISOString();
          let maxMmTimestampWithinBarrier: string | null = null;

          while (pass < maxPasses) {
            pass += 1;
            barrierTime = new Date().toISOString();

            const laborTicketsUpdates =
              await mmApiClient.fetchLaborTicketUpdates({
                system: connectorType,
                checkpointType: "export",
                ...(options?.onlyClosed ? { onlyClosed: "true" } : {}),
              });

            if (laborTicketsUpdates.length === 0) {
              if (pass === 1) {
                logger.info(
                  "syncLaborTicketsToERP:No updated labor tickets found"
                );
              }
              break;
            }

            maxMmTimestampWithinBarrier = this.trackMaxMmTimestampWithinBarrier(
              {
                laborTicketsUpdates,
                barrierTime,
                currentMax: maxMmTimestampWithinBarrier,
              }
            );

            if (pass === 1) {
              // IMPORTANT: This is the raw fetch size, not "to process". Actual processing is based on `pending`.
              logger.info(
                `ToERP: Fetched ${laborTicketsUpdates.length} labor ticket updates from MM`
              );
            }

            const pending = await this.buildPendingLaborTickets({
              laborTicketsUpdates,
              connectorType,
              ledger,
              barrierTime,
              pass,
              decisionState,
              attemptedSignatures,
            });

            if (pending.length === 0) {
              // Per-pass decision summary (helps debug pass 1 vs pass 2 behavior).
              LaborTicketSyncDecisionTracker.logPassSummary(
                logger,
                decisionState,
                {
                  pass,
                  barrierTime,
                  fetchedCount: laborTicketsUpdates.length,
                  pendingCount: 0,
                }
              );

              // Nothing new to do in this barrier window; safe to checkpoint and exit.
              break;
            }

            logger.info(
              `syncLaborTicketsToERP: pass=${pass}/${maxPasses}, barrier=${barrierTime}, pending=${pending.length}`
            );

            await this.processPendingLaborTickets({
              connectorType,
              connector,
              mmApiClient,
              ledger,
              pending,
              attemptedSignatures,
              failedLaborTicketRefs,
              outcomeByLaborTicketRef,
              decisionState,
            });

            // Per-pass decision summary (helps debug pass 1 vs pass 2 behavior).
            LaborTicketSyncDecisionTracker.logPassSummary(
              logger,
              decisionState,
              {
                pass,
                barrierTime,
                fetchedCount: laborTicketsUpdates.length,
                pendingCount: pending.length,
              }
            );
          }

          if (pass >= maxPasses) {
            logger.warn(
              `syncLaborTicketsToERP: Reached max passes (${maxPasses}). Checkpointing anyway to avoid infinite loops.`
            );
          }

          await this.finalizeRetryQueueForToErp({
            connectorType,
            mmApiClient,
            prefetchedFailedQueueRefs,
            failedLaborTicketRefs,
            outcomeByLaborTicketRef,
          });

          const checkpointTimestamp = this.computeCheckpointTimestamp({
            barrierTime,
            maxMmTimestampWithinBarrier,
            skewMs: this.CHECKPOINT_SKEW_MS,
          });

          await mmApiClient.saveCheckpoint({
            system: connectorType,
            table: "labor_tickets",
            checkpointType: "export",
            checkpointValue: {
              timestamp: checkpointTimestamp,
            },
          });
          logger.info("syncLaborTicketsToERP: Checkpoint saved:", {
            checkpointTimestamp,
          });

          LaborTicketSyncDecisionTracker.logSummary(logger, decisionState, {
            checkpointTimestamp,
          });
        } finally {
          try {
            await ledger.destroy();
          } catch (error) {
            logger.warn(
              "syncLaborTicketsToERP: Unable to destroy ledger manager",
              {
                error: summarizeError(error),
              }
            );
          }
        }
      } finally {
        await mmApiClient.destroy();
      }
    } catch (error) {
      logger.error("syncLaborTicketsToERP: Error", {
        error: summarizeError(error),
      });
    }
  }

  private static trackMaxMmTimestampWithinBarrier(options: {
    laborTicketsUpdates: MMReceiveLaborTicket[];
    barrierTime: string;
    currentMax: string | null;
  }): string | null {
    const { laborTicketsUpdates, barrierTime, currentMax } = options;
    let max = currentMax;
    for (const ticket of laborTicketsUpdates) {
      const ts = this.getComparableTicketTimestamp(ticket);
      if (!ts) continue;
      if (new Date(ts) > new Date(barrierTime)) continue;
      if (!max || new Date(ts) > new Date(max)) max = ts;
    }
    return max;
  }

  private static async buildPendingLaborTickets(options: {
    laborTicketsUpdates: MMReceiveLaborTicket[];
    connectorType: string;
    ledger: LaborTicketSyncLedgerManager;
    barrierTime: string;
    pass: number;
    decisionState: LaborTicketSyncDecisionState;
    attemptedSignatures: Set<string>;
  }): Promise<Array<{ sig: string; ticket: MMReceiveLaborTicket }>> {
    const {
      laborTicketsUpdates,
      connectorType,
      ledger,
      barrierTime,
      pass,
      decisionState,
      attemptedSignatures,
    } = options;

    // Build a unique pending list by a stable "business signature".
    // We do not skip tickets solely due to "future vs barrier" timestamps (clock skew / precision).
    // The barrier is used for bounding/checkpointing; duplicate prevention is handled by the persistent
    // ledger and signature dedupe.
    const pending: Array<{ sig: string; ticket: MMReceiveLaborTicket }> = [];
    const dedupeWithinFetch = new Set<string>();

    for (const ticket of laborTicketsUpdates) {
      if (!ticket.laborTicketRef) {
        logger.error(
          "syncLaborTicketsToERP: laborTicketRef is not set for laborTicket pulled from MM:",
          { laborTicket: ticket }
        );
        LaborTicketSyncDecisionTracker.recordMissingRef(decisionState, {
          updatedAt: ticket.updatedAt ?? null,
          barrierTime,
          pass,
        });
        continue;
      }

      // Normalize ref to a stable string for consistent ledger lookup + decision logs.
      // IMPORTANT: do not use raw (number|string) values as SQLite keys; always normalize to string.
      const laborTicketRef = String(ticket.laborTicketRef);

      const ts = this.getComparableTicketTimestamp(ticket);
      if (ts && new Date(ts) > new Date(barrierTime)) {
        // Diagnostic only: show timing oddities without affecting processing.
        logger.debug(
          "syncLaborTicketsToERP: Ticket comparable timestamp is after barrier; processing anyway",
          {
            laborTicketRef,
            updatedAt: ticket.updatedAt ?? null,
            comparableTs: ts,
            barrierTime,
            deltaMs: new Date(ts).getTime() - new Date(barrierTime).getTime(),
            pass,
          }
        );
      }

      const sig = this.laborTicketBusinessSignature(ticket);
      const signatureHash = LaborTicketSyncLedgerManager.hashSignature(sig);
      try {
        let existing = await ledger.getLatestByRef({
          system: connectorType,
          laborTicketRef,
        });
        if (!existing) {
          const erpImportKey = tryLaborTicketLedgerErpImportKey(
            ticket.workOrderId,
            ticket.laborTicketId
          );
          if (erpImportKey) {
            existing = await ledger.getLatestByRef({
              system: connectorType,
              laborTicketRef: erpImportKey,
            });
          }
        }
        if (existing && existing.signatureHash === signatureHash) {
          LaborTicketSyncDecisionTracker.recordPersistedSameSignatureSkip(
            decisionState,
            {
              laborTicketRef,
              updatedAt: ticket.updatedAt ?? null,
              barrierTime,
              pass,
              priorOutcome: existing.lastOutcome,
            }
          );
          continue;
        }
      } catch (error) {
        // Fail-open: if the local ledger is unavailable, proceed so we don't drop data.
        logger.warn(
          "syncLaborTicketsToERP: Unable to read local sync ledger; proceeding without persisted dedupe",
          { error: summarizeError(error) }
        );
      }

      if (attemptedSignatures.has(sig)) {
        LaborTicketSyncDecisionTracker.recordAttemptedSignatureSkip(
          decisionState,
          {
            laborTicketRef,
            updatedAt: ticket.updatedAt ?? null,
            barrierTime,
            pass,
          }
        );
        continue;
      }
      if (dedupeWithinFetch.has(sig)) {
        LaborTicketSyncDecisionTracker.recordDuplicateInFetchSkip(
          decisionState,
          {
            laborTicketRef,
            updatedAt: ticket.updatedAt ?? null,
            barrierTime,
            pass,
          }
        );
        continue;
      }

      dedupeWithinFetch.add(sig);
      pending.push({ sig, ticket });
      LaborTicketSyncDecisionTracker.recordPending(decisionState, {
        signature: sig,
        laborTicketRef,
        updatedAt: ticket.updatedAt ?? null,
        barrierTime,
        pass,
        operation: ticket.laborTicketId ? "update" : "create",
      });
    }

    return pending;
  }

  private static async processPendingLaborTickets(options: {
    connectorType: string;
    connector: IERPLaborTicketHandler;
    mmApiClient: MMApiClient;
    ledger: LaborTicketSyncLedgerManager;
    pending: Array<{ sig: string; ticket: MMReceiveLaborTicket }>;
    attemptedSignatures: Set<string>;
    failedLaborTicketRefs: Set<string>;
    outcomeByLaborTicketRef: Map<string, "success" | "failure">;
    decisionState: LaborTicketSyncDecisionState;
  }): Promise<void> {
    const {
      connectorType,
      connector,
      mmApiClient,
      ledger,
      pending,
      attemptedSignatures,
      failedLaborTicketRefs,
      outcomeByLaborTicketRef,
      decisionState,
    } = options;

    await Promise.all(
      pending.map(async ({ sig, ticket }) => {
        // Mark attempted up-front so we don't re-attempt within this run even if MM reorders.
        attemptedSignatures.add(sig);
        const laborTicketRef = String(ticket.laborTicketRef);

        try {
          const result = await this.processLaborTicket(
            connector,
            mmApiClient,
            ticket
          );

          // Final-outcome semantics: if we ever succeed for a given ref in this run,
          // it should not be treated as failed, even if earlier attempts threw.
          failedLaborTicketRefs.delete(laborTicketRef);
          outcomeByLaborTicketRef.set(laborTicketRef, "success");
          LaborTicketSyncDecisionTracker.markProcessed(decisionState, sig);

          try {
            await ledger.upsertAttempt({
              system: connectorType,
              laborTicketRef,
              signature: sig,
              outcome: "success",
              mmUpdatedAt: ticket.updatedAt ?? null,
              mmComparableTs: this.getComparableTicketTimestamp(ticket),
              errorSummary: null,
              attemptedAtIso: new Date().toISOString(),
            });
          } catch (error) {
            logger.warn(
              "syncLaborTicketsToERP: Unable to update local sync ledger after success",
              { error: summarizeError(error) }
            );
          }

          return result;
        } catch (error) {
          // Only record a ref as failed if we haven't already succeeded for it in this run.
          if (outcomeByLaborTicketRef.get(laborTicketRef) !== "success") {
            failedLaborTicketRefs.add(laborTicketRef);
            outcomeByLaborTicketRef.set(laborTicketRef, "failure");
          }
          LaborTicketSyncDecisionTracker.markFailed(decisionState, sig);
          const operation = ticket.laborTicketId ? "update" : "create";
          logger.error(
            `syncLaborTicketsToERP: Error processing laborTicketRef ${ticket.laborTicketRef}`,
            {
              laborTicketRef,
              laborTicketId: ticket.laborTicketId,
              operation,
              error: summarizeError(error),
            }
          );

          try {
            await ledger.upsertAttempt({
              system: connectorType,
              laborTicketRef,
              signature: sig,
              outcome: "failure",
              mmUpdatedAt: ticket.updatedAt ?? null,
              mmComparableTs: this.getComparableTicketTimestamp(ticket),
              errorSummary: JSON.stringify(summarizeError(error)),
              attemptedAtIso: new Date().toISOString(),
            });
          } catch (ledgerError) {
            logger.warn(
              "syncLaborTicketsToERP: Unable to update local sync ledger after failure",
              { error: summarizeError(ledgerError) }
            );
          }
          return undefined;
        }
      })
    );
  }

  private static async finalizeRetryQueueForToErp(options: {
    connectorType: string;
    mmApiClient: MMApiClient;
    prefetchedFailedQueueRefs: ReadonlySet<string> | null;
    failedLaborTicketRefs: ReadonlySet<string>;
    outcomeByLaborTicketRef: ReadonlyMap<string, "success" | "failure">;
  }): Promise<void> {
    const {
      connectorType,
      mmApiClient,
      prefetchedFailedQueueRefs,
      failedLaborTicketRefs,
      outcomeByLaborTicketRef,
    } = options;

    logger.info(
      `syncLaborTicketsToERP: ${failedLaborTicketRefs.size} failed labor ticket ids`
    );
    if (failedLaborTicketRefs.size > 0) {
      const failedTicketRefs = Array.from(failedLaborTicketRefs);
      logger.info(
        `syncLaborTicketsToERP: Reporting ${failedTicketRefs.length} labor ticket failures:`,
        {
          failedLaborTicketRefs: failedTicketRefs,
        }
      );
      const addFailedResult = await mmApiClient.addFailedLaborTicketRefs(
        connectorType,
        failedTicketRefs
      );
      logger.info("syncLaborTicketsToERP: addFailedResult:", {
        addFailedResult,
      });
    }

    const successRefsToRemove = this.computeSuccessRefsToRemoveFromRetryQueue({
      failedLaborTicketRefs,
      prefetchedFailedQueueRefs,
      outcomeByLaborTicketRef,
    });

    if (successRefsToRemove.length > 0) {
      logger.info(
        `syncLaborTicketsToERP: Removing ${successRefsToRemove.length} successfully processed labor tickets from retry queue`,
        { successRefsToRemove }
      );
      const deleteFailedResult = await mmApiClient.deleteFailedLaborTicketIds(
        connectorType,
        successRefsToRemove
      );
      logger.info("syncLaborTicketsToERP: deleteFailedResult:", {
        deleteFailedResult,
      });
    }
  }

  /**
   * Retries labor tickets that have failed to be created or updated in the ERP during the sync
   */
  static async retryFailed(
    connectorType: string,
    connector: IERPLaborTicketHandler
  ): Promise<void> {
    try {
      const mmApiClient = new MMApiClient();
      try {
        const successLaborTicketIds: string[] = [];
        const ledger = new LaborTicketSyncLedgerManager();

        try {
          const laborTickets =
            await mmApiClient.fetchFailedLaborTickets(connectorType);
          if (laborTickets.length === 0) {
            logger.info(
              "retryFailedLaborTickets: No failed labor tickets found"
            );
            return;
          }
          logger.info(
            "retryFailedLaborTickets: Failed Labor Tickets count:" +
              laborTickets.length
          );

          await Promise.all(
            laborTickets.map(async (laborTicket: MMReceiveLaborTicket) => {
              if (!laborTicket.laborTicketRef) {
                logger.error(
                  "retryFailedLaborTickets: laborTicketRef is not set for laborTicket pulled from MM:",
                  { laborTicket }
                );
                return undefined;
              }

              // Normalize ref to a stable string for consistent ledger keys and retry-queue operations.
              // IMPORTANT: do not use raw (number|string) values as SQLite keys; always normalize to string.
              const laborTicketRef = String(laborTicket.laborTicketRef);

              // IMPORTANT: compute signature from MM ticket as received (before timezone conversion).
              const signature = this.laborTicketBusinessSignature(laborTicket);
              const mmComparableTs =
                this.getComparableTicketTimestamp(laborTicket);
              const attemptedAtIso = new Date().toISOString();

              try {
                const laborTicketResult = await this.processLaborTicket(
                  connector,
                  mmApiClient,
                  laborTicket
                );
                successLaborTicketIds.push(laborTicketRef);

                try {
                  await ledger.upsertAttempt({
                    system: connectorType,
                    laborTicketRef,
                    signature,
                    outcome: "success",
                    mmUpdatedAt: laborTicket.updatedAt ?? null,
                    mmComparableTs,
                    errorSummary: null,
                    attemptedAtIso,
                  });
                } catch (ledgerError) {
                  logger.warn(
                    "retryFailedLaborTickets: Unable to update local sync ledger after success",
                    { error: summarizeError(ledgerError) }
                  );
                }

                return laborTicketResult;
              } catch (error) {
                logger.error(
                  "retryFailedLaborTickets: Error processing laborTicketRef:",
                  {
                    laborTicketRef,
                    laborTicketId: laborTicket.laborTicketId,
                    operation: laborTicket.laborTicketId ? "update" : "create",
                    error: summarizeError(error),
                  }
                );

                try {
                  await ledger.upsertAttempt({
                    system: connectorType,
                    laborTicketRef,
                    signature,
                    outcome: "failure",
                    mmUpdatedAt: laborTicket.updatedAt ?? null,
                    mmComparableTs,
                    errorSummary: JSON.stringify(summarizeError(error)),
                    attemptedAtIso,
                  });
                } catch (ledgerError) {
                  logger.warn(
                    "retryFailedLaborTickets: Unable to update local sync ledger after failure",
                    { error: summarizeError(ledgerError) }
                  );
                }

                return undefined;
              }
            })
          );

          if (successLaborTicketIds.length > 0) {
            logger.info("Deleting failed labor ticket ids:", {
              successLaborTicketIds,
            });
            const deleteFailedResult =
              await mmApiClient.deleteFailedLaborTicketIds(
                connectorType,
                successLaborTicketIds
              );
            logger.info("deleteFailedResult:", { deleteFailedResult });
          }
        } finally {
          try {
            await ledger.destroy();
          } catch (error) {
            logger.warn(
              "retryFailedLaborTickets: Unable to destroy ledger manager",
              {
                error: summarizeError(error),
              }
            );
          }
        }
      } finally {
        await mmApiClient.destroy();
      }
    } catch (error) {
      logger.error("retryFailedLaborTickets: Error", {
        error: summarizeError(error),
      });
    }
  }

  // ============================================================================
  // PRIVATE HELPER METHODS
  // ============================================================================

  private static async writeLaborTicketIdToMM(
    mmApiClient: MMApiClient,
    laborTicket: MMReceiveLaborTicket,
    laborTicketResult: MMReceiveLaborTicket
  ): Promise<void> {
    const updateRefAPIResponse = await mmApiClient.updateLaborTicketIdByRef(
      String(laborTicket.laborTicketRef),
      laborTicketResult.laborTicketId
    );
    logger.info(
      `Updated laborTicketId ${laborTicketResult.laborTicketId} for laborTicketRef ${laborTicket.laborTicketRef} in MM:`,
      { updateRefAPIResponse }
    );
  }

  private static async processLaborTicket(
    connector: IERPLaborTicketHandler,
    mmApiClient: MMApiClient,
    laborTicket: MMReceiveLaborTicket
  ): Promise<MMReceiveLaborTicket> {
    let laborTicketResult: MMReceiveLaborTicket & { warnings?: string[] };
    let createWarnings: string[] | undefined;

    // Convert MM's UTC timestamps into the ERP's local timezone before invoking connector code.
    // Connector implementations should treat the incoming values as localized wall time and avoid
    // applying any further timezone shifts.
    laborTicketResult = convertLaborTicketToLocalTimezone(laborTicket);

    logger.info(
      `processing laborTicket, id=${laborTicket.laborTicketId}, ref=${laborTicket.laborTicketRef}`
    );
    logger.debug({ laborTicket });

    // MLW TODO: Should we always swap out the resource ID for the machine group ERP id when relevant?
    // Or use a flag? Connector-specific mapping (MES / ERP behavior) replaces the old GraphQL-based helper.

    if (!laborTicket.laborTicketId) {
      const recoveredLaborTicketId = await LaborTicketRecovery.attemptBeforeCreate(
        connector,
        laborTicket
      );

      if (recoveredLaborTicketId) {
        logger.info(
          "[LABOR_RECOVERY] Recovered existing ERP laborTicketId; patching MM before ERP update",
          {
            laborTicketRef: laborTicket.laborTicketRef,
            laborTicketId: recoveredLaborTicketId,
          }
        );

        laborTicket.laborTicketId = recoveredLaborTicketId;
        await this.writeLaborTicketIdToMM(mmApiClient, laborTicket, laborTicket);

        laborTicketResult = await connector.updateLaborTicketInERP(laborTicket);
        laborTicketResult.laborTicketId =
          laborTicketResult.laborTicketId || recoveredLaborTicketId;
      } else {
        try {
          const createResult = await connector.createLaborTicketInERP(
            laborTicket
          );
          laborTicketResult = createResult.laborTicket;
          createWarnings = createResult.warnings;
          const erpUid = createResult.erpUid;
          laborTicketResult.laborTicketId = erpUid;
          await this.writeLaborTicketIdToMM(
            mmApiClient,
            laborTicket,
            laborTicketResult
          );
        } catch (error) {
          await LaborTicketRecovery.patchMmOnCreateFailure(
            mmApiClient,
            laborTicket,
            error
          );
          throw error;
        }
      }
    } else {
      laborTicketResult = await connector.updateLaborTicketInERP(laborTicket);
    }

    // The polling path has no operator session to surface non-fatal ERP warnings
    // to; log them so they are not silently lost.
    const warnings = normalizeLaborTicketWarnings(
      createWarnings ?? laborTicketResult.warnings
    );
    if (warnings.length) {
      logger.info("ToERP: ERP returned non-fatal warnings for labor ticket", {
        laborTicketRef: laborTicket.laborTicketRef,
        warnings,
      });
    }

    logger.info("ToERP: laborTicket update result:", {
      laborTicketResult:
        laborTicketResult || "Failed to create/update labor ticket",
      laborTicketRef: laborTicket.laborTicketRef,
      operation: laborTicket.laborTicketId ? "update" : "create",
    });

    return laborTicketResult;
  }

  /**
   * A stable identity for "did the business-relevant contents of this labor ticket change?"
   *
   * IMPORTANT: Excludes fields that the SDK itself mutates (e.g. laborTicketId, updatedAt)
   * AND identity fields the caller already keys the ledger by (laborTicketRef / erp-import key).
   * Including identity here would make from-ERP imports (no ref yet) and the next to-ERP
   * read (real MM ref) produce different hashes for the same business state, defeating dedupe.
   *
   * PUBLIC: Used by both polling and NATS handlers for consistent deduplication
   */
  public static laborTicketBusinessSignature(
    ticket: MMReceiveLaborTicket
  ): string {
    const signatureObject = {
      clockIn: normalizeTimestamp(ticket.clockIn),
      clockOut: normalizeTimestamp(ticket.clockOut),
      goodParts: normalizeNumber(ticket.goodParts),
      badParts: normalizeNumber(ticket.badParts),
      state: normalizeString(ticket.state),
      type: normalizeString(ticket.type),
      transactionDate: normalizeTimestamp(ticket.transactionDate),
      workOrderId: normalizeLaborTicketRef(ticket.workOrderId),
      sequenceNumber: normalizeNumber(ticket.sequenceNumber),
      personId: normalizeLaborTicketRef(ticket.personId),
      resourceId: normalizeLaborTicketRef(ticket.resourceId),
      lot: normalizeString(ticket.lot),
      split: normalizeString(ticket.split),
      sub: normalizeString(ticket.sub),
      comment: normalizeString(ticket.comment),
      workOrderOperationClosedDate: normalizeTimestamp(
        ticket.workOrderOperationClosedDate
      ),
      reasons: normalizeReasons(ticket.reasons),
    };

    return JSON.stringify(signatureObject);
  }

  /**
   * Selects a comparable timestamp for windowing decisions.
   * Prefer updatedAt; fall back to other stable timestamps if needed.
   */
  private static getComparableTicketTimestamp(
    ticket: MMReceiveLaborTicket
  ): string | null {
    return (
      ticket.updatedAt ??
      ticket.createdAt ??
      ticket.transactionDate ??
      ticket.clockOut ??
      ticket.clockIn ??
      null
    );
  }

  private static addMsToIso(iso: string, ms: number): string {
    return new Date(new Date(iso).getTime() + ms).toISOString();
  }

  /**
   * Computes the checkpoint timestamp for the next barrier by adding the skew milliseconds to the maximum MM timestamp within the barrier.
   */
  private static computeCheckpointTimestamp(options: {
    barrierTime: string;
    maxMmTimestampWithinBarrier: string | null;
    skewMs: number;
  }): string {
    const { barrierTime, maxMmTimestampWithinBarrier, skewMs } = options;

    if (!maxMmTimestampWithinBarrier) return barrierTime;

    const mmPlusSkew = this.addMsToIso(maxMmTimestampWithinBarrier, skewMs);
    return new Date(mmPlusSkew) < new Date(barrierTime)
      ? mmPlusSkew
      : barrierTime;
  }
}

type LaborTicketSyncDecision =
  | "processed"
  | "failed"
  | "pending"
  | "skipped_dedupe_attempted_signature"
  | "skipped_duplicate_in_fetch"
  | "skipped_persisted_same_signature"
  | "skipped_missing_ref";

type LaborTicketSyncDecisionRow = {
  ref: string | null;
  updatedAt: string | null;
  barrierTime: string;
  decision: LaborTicketSyncDecision;
  pass: number;
  operation?: "create" | "update";
  priorOutcome?: "success" | "failure";
};

type LaborTicketSyncDecisionState = {
  rows: LaborTicketSyncDecisionRow[];
  pendingRowBySignature: Map<string, LaborTicketSyncDecisionRow>;
};

class LaborTicketSyncDecisionTracker {
  static init(): LaborTicketSyncDecisionState {
    return { rows: [], pendingRowBySignature: new Map() };
  }

  private static countDecisions(
    rows: LaborTicketSyncDecisionRow[]
  ): Record<string, number> {
    const counts: Record<string, number> = {};
    for (const r of rows) {
      counts[r.decision] = (counts[r.decision] ?? 0) + 1;
    }
    return counts;
  }

  private static formatDecisionRow(
    row: LaborTicketSyncDecisionRow & { ref: string }
  ): {
    ref: string;
    updatedAt: string | null;
    barrierTime: string;
    pass: number;
    operation?: "create" | "update";
    priorOutcome?: "success" | "failure";
    decision: LaborTicketSyncDecision;
  } {
    // NOTE: We intentionally omit `comparableTs` and `deltaMs` from INFO logs to reduce redundancy.
    return {
      ref: row.ref,
      updatedAt: row.updatedAt,
      barrierTime: row.barrierTime,
      pass: row.pass,
      operation: row.operation,
      priorOutcome: row.priorOutcome,
      decision: row.decision,
    };
  }

  static logPassSummary(
    log: typeof logger,
    state: LaborTicketSyncDecisionState,
    options: {
      pass: number;
      barrierTime: string;
      fetchedCount: number;
      pendingCount: number;
    }
  ): void {
    const passRows = state.rows.filter((r) => r.pass === options.pass);

    // Keep full fidelity for the pass (do not roll up by ref), so pass 1 and pass 2 can be compared.
    const ticketDecisions = passRows
      .filter((r): r is LaborTicketSyncDecisionRow & { ref: string } =>
        Boolean(r.ref)
      )
      .map((r) => this.formatDecisionRow(r));

    log.info("syncLaborTicketsToERP: Pass decision summary", {
      pass: options.pass,
      barrierTime: options.barrierTime,
      fetchedCount: options.fetchedCount,
      pendingCount: options.pendingCount,
      decisionCounts: this.countDecisions(passRows),
      ticketDecisions,
    });
  }

  private static addRow(
    state: LaborTicketSyncDecisionState,
    row: Omit<LaborTicketSyncDecisionRow, "operation" | "priorOutcome"> & {
      operation?: "create" | "update";
      priorOutcome?: "success" | "failure";
      signature?: string;
    }
  ): void {
    const { signature, ...base } = row;
    const full: LaborTicketSyncDecisionRow = {
      ref: base.ref,
      updatedAt: base.updatedAt,
      barrierTime: base.barrierTime,
      decision: base.decision,
      pass: base.pass,
      operation: base.operation,
      priorOutcome: base.priorOutcome,
    };
    if (signature) {
      state.pendingRowBySignature.set(signature, full);
    }
    state.rows.push(full);
  }

  static recordMissingRef(
    state: LaborTicketSyncDecisionState,
    options: {
      updatedAt: string | null;
      barrierTime: string;
      pass: number;
    }
  ): void {
    this.addRow(state, {
      ref: null,
      updatedAt: options.updatedAt,
      barrierTime: options.barrierTime,
      decision: "skipped_missing_ref",
      pass: options.pass,
    });
  }

  static recordPersistedSameSignatureSkip(
    state: LaborTicketSyncDecisionState,
    options: {
      laborTicketRef: string;
      updatedAt: string | null;
      barrierTime: string;
      pass: number;
      priorOutcome: "success" | "failure";
    }
  ): void {
    this.addRow(state, {
      ref: options.laborTicketRef,
      updatedAt: options.updatedAt,
      barrierTime: options.barrierTime,
      decision: "skipped_persisted_same_signature",
      pass: options.pass,
      priorOutcome: options.priorOutcome,
    });
  }

  static recordAttemptedSignatureSkip(
    state: LaborTicketSyncDecisionState,
    options: {
      laborTicketRef: string;
      updatedAt: string | null;
      barrierTime: string;
      pass: number;
    }
  ): void {
    this.addRow(state, {
      ref: options.laborTicketRef,
      updatedAt: options.updatedAt,
      barrierTime: options.barrierTime,
      decision: "skipped_dedupe_attempted_signature",
      pass: options.pass,
    });
  }

  static recordDuplicateInFetchSkip(
    state: LaborTicketSyncDecisionState,
    options: {
      laborTicketRef: string;
      updatedAt: string | null;
      barrierTime: string;
      pass: number;
    }
  ): void {
    this.addRow(state, {
      ref: options.laborTicketRef,
      updatedAt: options.updatedAt,
      barrierTime: options.barrierTime,
      decision: "skipped_duplicate_in_fetch",
      pass: options.pass,
    });
  }

  static recordPending(
    state: LaborTicketSyncDecisionState,
    options: {
      signature: string;
      laborTicketRef: string;
      updatedAt: string | null;
      barrierTime: string;
      pass: number;
      operation: "create" | "update";
    }
  ): void {
    this.addRow(state, {
      signature: options.signature,
      ref: options.laborTicketRef,
      updatedAt: options.updatedAt,
      barrierTime: options.barrierTime,
      decision: "pending",
      pass: options.pass,
      operation: options.operation,
    });
  }

  static markProcessed(
    state: LaborTicketSyncDecisionState,
    signature: string
  ): void {
    const row = state.pendingRowBySignature.get(signature);
    if (row) row.decision = "processed";
  }

  static markFailed(
    state: LaborTicketSyncDecisionState,
    signature: string
  ): void {
    const row = state.pendingRowBySignature.get(signature);
    if (row) row.decision = "failed";
  }

  static logSummary(
    log: typeof logger,
    state: LaborTicketSyncDecisionState,
    options: { checkpointTimestamp: string }
  ): void {
    // Per-ticket decision table (low-volume friendly). Clarifies why a ticket was (or wasn't) processed.
    const latestDecisionByRef = new Map<string, LaborTicketSyncDecisionRow>();
    for (const row of state.rows) {
      if (!row.ref) continue;
      latestDecisionByRef.set(row.ref, row);
    }
    const ticketDecisions = Array.from(latestDecisionByRef.values()).sort(
      (a, b) => String(a.ref).localeCompare(String(b.ref))
    );

    log.info("syncLaborTicketsToERP: Ticket decisions", {
      checkpointTimestamp: options.checkpointTimestamp,
      ticketDecisions: ticketDecisions.map((r) =>
        this.formatDecisionRow(
          r as LaborTicketSyncDecisionRow & { ref: string }
        )
      ),
    });
  }
}