/**
 * Work-order gap filling from labor tickets.
 *
 * A connector's from-ERP work-order sync is windowed (newest N jobs, a lookback floor,
 * a plant-scoped query, …), so a long-running job can age out of the batch while operators
 * are still clocked into it. Labor tickets for that job then fail the SDK's labor import
 * filter (`notInSyncedWorkOrderOperation`, see {@link filterLaborTicketsForMmImport}) or,
 * worse, `erp.labor_ticket.work_order_operation_fk` on insert.
 *
 * This module holds the MM-side policy for closing that gap, independent of any ERP:
 *
 *  - which work orders to FETCH from the ERP (bounded by OPEN labor, plus a persisted
 *    watch list so a filled job is re-fetched until its terminal state is imported),
 *  - which entities of the fetched batch to WRITE without fighting the bulk writer,
 *  - which operation rows to hand back so the same cycle's labor filter accepts the tickets.
 *
 * The ERP query and the ERP → {@link MMSendWorkOrderBatch} mapping stay in the connector.
 * {@link StandardProcessDrivers.fillWorkOrderGapsFromLaborTickets} orchestrates these
 * functions around a connector-supplied fetch callback.
 *
 * Every key here comes from the labor-import helpers ({@link normalizeLaborTicketWorkOrderOpKey},
 * {@link buildSyncedWorkOrderOperationKeySet}) so "missing" can never drift from what the
 * labor filter rejects.
 *
 * This path is SUPPLEMENTARY to the connector's regular work-order sync, never a replacement:
 * labor tickets do not describe the shop schedule.
 */

import { MMWorkOrderStatus, normalizeMMStatus } from "../types/erp-types.js";
import type {
  MMSendWorkOrder,
  MMSendWorkOrderBatch,
  MMSendWorkOrderOperation,
} from "../services/mm-api-service/types/send-types.js";
import {
  buildSyncedWorkOrderOperationKeySet,
  isLotSplitAwareRowSet,
  normalizeLaborTicketWorkOrderOpKey,
  type SyncedWorkOrderOperationRow,
} from "./labor-import-from-erp.js";
import {
  getRuntimeStateText,
  setRuntimeStateJson,
} from "./local-data-store/sdk-runtime-state-db.js";
import { sdkLogger as logger } from "../services/reporting-service/logger.js";

/**
 * The fields of a labor ticket that identify its work order operation and its state.
 * `lot` / `split` take part in the key only when the bulk rows opt in
 * ({@link isLotSplitAwareRowSet}), exactly as in the labor filter.
 */
export interface LaborTicketWorkOrderOpRef {
  workOrderId?: unknown;
  lot?: unknown;
  split?: unknown;
  sub?: unknown;
  sequenceNumber?: unknown;
  /** `"OPEN"` while the operator is clocked in (see {@link OPEN_LABOR_TICKET_STATE}). */
  state?: unknown;
}

/** Labor ticket state meaning an operator is currently clocked in. Compared case-insensitively. */
export const OPEN_LABOR_TICKET_STATE = "OPEN";

/** Days a tracked work order is kept after the last cycle in which open labor referenced it. */
export const DEFAULT_TRACKED_WORK_ORDER_TTL_DAYS = 30;

/**
 * Hard cap on the watch list, purely to prevent unbounded growth. Oldest entries are evicted
 * first. Pending close-outs (fetched terminal, write not yet landed) are exempt: they are
 * bounded by one cycle's fetch and leave the list once their write lands.
 */
export const DEFAULT_MAX_TRACKED_WORK_ORDERS = 1000;

/** Fetch-list size above which the driver logs a warning (no cap; every planned id is fetched). */
export const DEFAULT_GAP_FILL_FETCH_WARN_THRESHOLD = 200;

/** A work order on the gap-fill watch list. */
export interface TrackedLaborReferencedWorkOrder {
  workOrderId: string;
  /** ISO timestamp of the last cycle in which a labor ticket drove a fetch of this work order. */
  lastLaborSeenAt: string;
}

const upper = (value: unknown): string => String(value ?? "").trim().toUpperCase();

/**
 * The MM work-order id as written: trimmed, case preserved. Coverage decisions use this
 * rather than {@link upper} because they are asymmetric — a work order wrongly judged
 * "covered" has its parent rows omitted from the write, while one wrongly judged uncovered
 * costs one redundant upsert. Case folding belongs to the labor filter, which compares
 * against a key set, not to decisions about which writer owns a row.
 */
const exactKey = (value: unknown): string => String(value ?? "").trim();

/** A ticket's key in the form the bulk row set uses (see {@link isLotSplitAwareRowSet}). */
const ticketOpKey = (ticket: LaborTicketWorkOrderOpRef, lotSplitAware: boolean): string =>
  normalizeLaborTicketWorkOrderOpKey(
    ticket.workOrderId,
    ticket.sub,
    ticket.sequenceNumber,
    lotSplitAware ? { lot: ticket.lot, split: ticket.split } : undefined
  );

/** A send operation's key in the same form. */
const sendOpKey = (
  op: Pick<MMSendWorkOrderOperation, "workOrderId" | "lot" | "split" | "sub" | "sequenceNumber">,
  lotSplitAware: boolean
): string =>
  normalizeLaborTicketWorkOrderOpKey(
    op.workOrderId,
    op.sub,
    op.sequenceNumber,
    lotSplitAware ? { lot: op.lot, split: op.split } : undefined
  );

const isOpen = (ticket: LaborTicketWorkOrderOpRef): boolean =>
  upper(ticket?.state) === OPEN_LABOR_TICKET_STATE;

/**
 * A ticket the labor filter would reject as `missingFields` before the FK check (blank work
 * order id or sequence number) can never be helped by a fetch, so it never drives one.
 */
const referencesAnOperation = (ticket: LaborTicketWorkOrderOpRef): boolean =>
  !!ticket && exactKey(ticket.workOrderId) !== "" && exactKey(ticket.sequenceNumber) !== "";

/**
 * Distinct work order ids referenced by OPEN labor tickets whose
 * (workOrderId, sub, sequenceNumber) has no operation among the bulk-synced ops.
 *
 * OPEN-only, unlike {@link findMissingWorkOrderOpKeys}: this picks which work orders to
 * FETCH from the ERP, so restricting it to active labor bounds the fetch by how many
 * operators are clocked in rather than by the ticket volume in the labor window. What to
 * WRITE among the work orders it selects is a separate question.
 *
 * Ids are returned trimmed with their case preserved.
 */
export function findOpenLaborWorkOrderIdsMissingOps(
  laborTickets: ReadonlyArray<LaborTicketWorkOrderOpRef>,
  bulkSyncedWorkOrderOps: ReadonlyArray<SyncedWorkOrderOperationRow>
): string[] {
  return workOrderIdsMissingOps(laborTickets, bulkSyncedWorkOrderOps, isOpen);
}

/**
 * Like {@link findOpenLaborWorkOrderIdsMissingOps} but for tickets of EVERY state, so a
 * work order observed only through CLOSED tickets (an operation that opened and closed
 * between two labor polls, or the watch list lost to a container recreate while the job
 * was already closed) is fetched too, and its tickets are not rejected for good.
 *
 * The price is the bound: the fetch is then limited by the closed tickets in the
 * connector's labor window rather than by concurrent clock-ins, so a wide lookback or an
 * initial load can name many aged-out work orders at once. Opt in via
 * `includeClosedTickets` on the driver when the labor window is narrow enough.
 */
export function findLaborWorkOrderIdsMissingOps(
  laborTickets: ReadonlyArray<LaborTicketWorkOrderOpRef>,
  bulkSyncedWorkOrderOps: ReadonlyArray<SyncedWorkOrderOperationRow>
): string[] {
  return workOrderIdsMissingOps(laborTickets, bulkSyncedWorkOrderOps, () => true);
}

function workOrderIdsMissingOps(
  laborTickets: ReadonlyArray<LaborTicketWorkOrderOpRef>,
  bulkSyncedWorkOrderOps: ReadonlyArray<SyncedWorkOrderOperationRow>,
  include: (ticket: LaborTicketWorkOrderOpRef) => boolean
): string[] {
  const bulkOps = bulkSyncedWorkOrderOps ?? [];
  const keySet = buildSyncedWorkOrderOperationKeySet(bulkOps);
  const aware = isLotSplitAwareRowSet(bulkOps);
  const missing = new Set<string>();

  for (const ticket of laborTickets ?? []) {
    if (!referencesAnOperation(ticket) || !include(ticket)) continue;
    if (!keySet.has(ticketOpKey(ticket, aware))) {
      missing.add(exactKey(ticket.workOrderId));
    }
  }

  return [...missing];
}

/**
 * The normalized operation keys that labor tickets reference but the bulk-synced ops do
 * not contain — the same computation as {@link findOpenLaborWorkOrderIdsMissingOps},
 * keeping the key instead of collapsing it to a work order id. Keys are in the form the
 * bulk row set uses (with lot / split when it opts in), and
 * {@link computeWorkOrderGapFillWriteScope} matches fetched operations the same way.
 *
 * These are the gap fill's primary write set, though not its only one: a terminal work
 * order also writes its full operation set for close-out, and a work order the bulk sync
 * is not covering writes its backing work order, part and part operation rows too. What
 * all three have in common is no competing writer — a missing operation has none by
 * definition, and neither case above is being written by the bulk sync this cycle.
 *
 * Every ticket state counts, not just OPEN. That is what makes the returned op set safe
 * to hand to the labor filter: an op a ticket references is then either in the bulk batch
 * (written this cycle by the bulk sync) or in this set (written this cycle by the gap
 * fill), so "returned" implies "in MM" without depending on an earlier cycle's write
 * having landed. Restricted to OPEN, a write that failed while a ticket was open would
 * never be retried once the operator clocked out — the key would be gone but the op still
 * returned, and the CLOSED ticket would violate `work_order_operation_fk` every cycle
 * until the work order's watch-list TTL expired.
 */
export function findMissingWorkOrderOpKeys(
  laborTickets: ReadonlyArray<LaborTicketWorkOrderOpRef>,
  bulkSyncedWorkOrderOps: ReadonlyArray<SyncedWorkOrderOperationRow>
): Set<string> {
  const bulkOps = bulkSyncedWorkOrderOps ?? [];
  const keySet = buildSyncedWorkOrderOperationKeySet(bulkOps);
  const aware = isLotSplitAwareRowSet(bulkOps);
  const missing = new Set<string>();

  for (const ticket of laborTickets ?? []) {
    if (!referencesAnOperation(ticket)) continue;
    const key = ticketOpKey(ticket, aware);
    if (!keySet.has(key)) {
      missing.add(key);
    }
  }

  return missing;
}

/**
 * Work order ids the bulk sync touched this cycle, from its work order operations. Id-level
 * only: {@link computeWorkOrderGapFillWriteScope} judges coverage per MM work order row
 * (`workOrderId + lot + split + sub`), since the bulk window may carry one lot or
 * sub-assembly of a job but not its siblings.
 */
export function buildBulkSyncWorkOrderIdSet(
  bulkWorkOrderIds: Iterable<unknown>
): Set<string> {
  const set = new Set<string>();
  for (const id of bulkWorkOrderIds ?? []) {
    const key = exactKey(id);
    if (key) set.add(key);
  }
  return set;
}

/** Whether `workOrderId` is in a set built by {@link buildBulkSyncWorkOrderIdSet}. */
export function isWorkOrderCoveredByBulkSync(
  bulkWorkOrderIds: ReadonlySet<string>,
  workOrderId: unknown
): boolean {
  return bulkWorkOrderIds.has(exactKey(workOrderId));
}

/** Terminal work-order statuses: the ERP will not post more labor to a work order in one of these. */
const TERMINAL_WORK_ORDER_STATUSES: ReadonlySet<string> = new Set([
  MMWorkOrderStatus.COMPLETED,
  MMWorkOrderStatus.CANCELLED,
  MMWorkOrderStatus.DELETED,
]);

/**
 * Default terminal predicate for a fetched work order: a terminal status (per
 * {@link normalizeMMStatus}) or a non-blank `closedDate`. A terminal work order's final
 * state is imported once more, after which it leaves the watch list. Connectors whose
 * mapping encodes completion differently pass their own predicate to the driver.
 */
export function isTerminalWorkOrderForGapFill(
  workOrder: Pick<MMSendWorkOrder, "status" | "closedDate">
): boolean {
  if (TERMINAL_WORK_ORDER_STATUSES.has(normalizeMMStatus(workOrder?.status))) {
    return true;
  }
  const closedDate = workOrder?.closedDate;
  return typeof closedDate === "string"
    ? closedDate.trim().length > 0
    : Boolean(closedDate);
}

export interface PlanWorkOrderGapFillFetchInput {
  tracked: ReadonlyArray<TrackedLaborReferencedWorkOrder>;
  /**
   * Work order ids labor tickets referenced this cycle with no bulk-synced operation: from
   * {@link findOpenLaborWorkOrderIdsMissingOps}, or {@link findLaborWorkOrderIdsMissingOps}
   * when closed tickets may drive a fetch too.
   */
  laborWorkOrderIds: ReadonlyArray<string>;
  now: Date;
  /** Default {@link DEFAULT_TRACKED_WORK_ORDER_TTL_DAYS}. */
  trackedTtlDays?: number;
}

export interface PlanWorkOrderGapFillFetchResult {
  /** Work order ids to fetch from the ERP this cycle: labor-referenced ∪ carried watch list, deduped. */
  fetchWorkOrderIds: string[];
  /** Watch-list entries still valid after pruning; the next watch list is built from these. */
  carriedTracked: TrackedLaborReferencedWorkOrder[];
  dropped: { expired: number };
}

/**
 * Decide which work orders to fetch this cycle and which tracked entries to carry forward.
 * Prunes entries that expired (TTL). Entries the bulk sync covers again are NOT pruned here:
 * the bulk rows name a job id, but a job may have lots / sub-assemblies the window did not
 * carry, and only the fetched rows tell. {@link updateTrackedAfterWorkOrderGapFillFetch}
 * drops them once the fetch shows every row covered.
 */
export function planWorkOrderGapFillFetch(
  input: PlanWorkOrderGapFillFetchInput
): PlanWorkOrderGapFillFetchResult {
  const { tracked, laborWorkOrderIds, now } = input;
  const ttlDays = input.trackedTtlDays ?? DEFAULT_TRACKED_WORK_ORDER_TTL_DAYS;
  const ttlFloorMs = now.getTime() - ttlDays * 24 * 60 * 60 * 1000;

  const dropped = { expired: 0 };
  const carriedTracked: TrackedLaborReferencedWorkOrder[] = [];
  for (const entry of tracked ?? []) {
    const seenAtMs = Date.parse(entry.lastLaborSeenAt);
    if (Number.isNaN(seenAtMs) || seenAtMs < ttlFloorMs) {
      dropped.expired += 1;
    } else {
      carriedTracked.push(entry);
    }
  }

  const fetchWorkOrderIds: string[] = [];
  const seen = new Set<string>();
  for (const id of [
    ...(laborWorkOrderIds ?? []),
    ...carriedTracked.map((t) => t.workOrderId),
  ]) {
    const key = upper(id);
    if (!key || seen.has(key)) continue;
    seen.add(key);
    fetchWorkOrderIds.push(exactKey(id));
  }

  return { fetchWorkOrderIds, carriedTracked, dropped };
}

export interface UpdateTrackedAfterWorkOrderGapFillFetchInput {
  carriedTracked: ReadonlyArray<TrackedLaborReferencedWorkOrder>;
  /** The same ids given to {@link planWorkOrderGapFillFetch}. */
  laborWorkOrderIds: ReadonlyArray<string>;
  /** Ids of fetched work orders the terminal predicate accepted. */
  terminalWorkOrderIds: Iterable<unknown>;
  /**
   * Fetched work order ids the bulk sync covered row for row this cycle and the gap fill is
   * writing nothing for (`bulkCoveredWorkOrderIds` from
   * {@link computeWorkOrderGapFillWriteScope}). The bulk sync owns them now, so they leave
   * the watch list; an id that later ages out again is re-added by labor like any other.
   */
  bulkCoveredWorkOrderIds?: Iterable<unknown>;
  now: Date;
  /** Default {@link DEFAULT_MAX_TRACKED_WORK_ORDERS}. */
  maxTrackedWorkOrders?: number;
}

export interface UpdateTrackedAfterWorkOrderGapFillFetchResult {
  /** The next watch list. */
  tracked: TrackedLaborReferencedWorkOrder[];
  /**
   * Entries removed because their fetched work order is terminal. Returned so the caller
   * can keep persisting them until the close-out write actually lands — these work orders
   * are outside the bulk window, so one dropped before a failed write is never fetched
   * again and its MM work order keeps a stale status for good.
   */
  terminal: TrackedLaborReferencedWorkOrder[];
  dropped: { coveredByBulkSync: number; terminal: number; evicted: number };
}

/**
 * Produce the next watch list after a gap-fill fetch: refresh/add work orders that labor
 * referenced this cycle, remove work orders the bulk sync fully covers again (it owns them)
 * and work orders whose fetched state is terminal (their final state was just imported),
 * and evict the oldest entries beyond the cap.
 */
export function updateTrackedAfterWorkOrderGapFillFetch(
  input: UpdateTrackedAfterWorkOrderGapFillFetchInput
): UpdateTrackedAfterWorkOrderGapFillFetchResult {
  const { carriedTracked, laborWorkOrderIds, terminalWorkOrderIds, now } = input;
  const cap = input.maxTrackedWorkOrders ?? DEFAULT_MAX_TRACKED_WORK_ORDERS;

  const byKey = new Map<string, TrackedLaborReferencedWorkOrder>();
  for (const entry of carriedTracked ?? []) {
    byKey.set(upper(entry.workOrderId), entry);
  }

  const nowIso = now.toISOString();
  for (const id of laborWorkOrderIds ?? []) {
    const key = upper(id);
    if (!key) continue;
    byKey.set(key, { workOrderId: exactKey(id), lastLaborSeenAt: nowIso });
  }

  const dropped = { coveredByBulkSync: 0, terminal: 0, evicted: 0 };
  // Before the terminal pass: a fully covered terminal id has nothing left for the gap
  // fill to close out, so it must not linger as a pending close-out.
  for (const id of input.bulkCoveredWorkOrderIds ?? []) {
    const key = upper(id);
    if (key && byKey.delete(key)) dropped.coveredByBulkSync += 1;
  }

  const terminal: TrackedLaborReferencedWorkOrder[] = [];
  for (const id of terminalWorkOrderIds ?? []) {
    const key = upper(id);
    const entry = key ? byKey.get(key) : undefined;
    if (entry) {
      terminal.push(entry);
      byKey.delete(key);
      dropped.terminal += 1;
    }
  }

  let tracked = [...byKey.values()];
  if (tracked.length > cap) {
    tracked = tracked
      .sort(
        (a, b) => Date.parse(b.lastLaborSeenAt) - Date.parse(a.lastLaborSeenAt)
      )
      .slice(0, cap);
    dropped.evicted = byKey.size - cap;
  }

  return { tracked, terminal, dropped };
}

/**
 * Work order operation rows in the shape the labor filter consumes
 * (`FilterLaborTicketsForMmImportOptions.syncedWorkOrderOperations`), so gap-filled ops can
 * be concatenated with the bulk sync's rows.
 */
export function toSyncedWorkOrderOperationRows(
  workOrderOperations: ReadonlyArray<
    Pick<MMSendWorkOrderOperation, "workOrderId" | "lot" | "split" | "sub" | "sequenceNumber">
  >
): SyncedWorkOrderOperationRow[] {
  return (workOrderOperations ?? []).map((op) => ({
    work_order_id: op.workOrderId,
    lot: op.lot,
    split: op.split,
    sub: op.sub,
    sequence_number: op.sequenceNumber,
  }));
}

export interface ComputeWorkOrderGapFillWriteScopeInput {
  /** Everything the connector mapped for the fetched work orders. */
  fullBatch: MMSendWorkOrderBatch;
  /**
   * Work order operation rows the bulk sync wrote this cycle. `work_order_id` + `sub` (and
   * `lot` / `split` when present) decide which MM work order rows are covered; their
   * `sequence_number`s decide which part operations of a shared part the bulk sync owns.
   */
  bulkSyncedWorkOrderOperations: ReadonlyArray<SyncedWorkOrderOperationRow>;
  /** From {@link findMissingWorkOrderOpKeys}. */
  missingOpKeys: ReadonlySet<string>;
  /** Fetched work orders judged terminal by the terminal predicate (any case / whitespace). */
  terminalWorkOrderIds: Iterable<unknown>;
}

export interface ComputeWorkOrderGapFillWriteScopeResult {
  /** The subset of `fullBatch` to write, FK-complete for every operation it contains. */
  batchToWrite: MMSendWorkOrderBatch;
  /** Every work order operation of `fullBatch`, for widening this cycle's labor filter key set. */
  keySetWorkOrderOps: SyncedWorkOrderOperationRow[];
  /**
   * Terminal work orders with at least one fetched row the bulk sync did not cover: those
   * rows and their full operation sets are written.
   */
  closingWorkOrderIds: string[];
  /**
   * Fetched work order ids with at least one fetched row, every one of which the bulk sync
   * covered, and none of whose operations is written this cycle. Compared
   * case-insensitively, as the watch list keys ids: an id is listed (in every fetched
   * spelling) only when the rows of all its spellings are covered and no spelling has an
   * operation in `batchToWrite`. Feed to {@link updateTrackedAfterWorkOrderGapFillFetch}:
   * the bulk sync owns them, and the gap fill has nothing left to retry for them.
   */
  bulkCoveredWorkOrderIds: string[];
  /** Work order ids with at least one row (`workOrderId + lot + split + sub`) whose parents are written. */
  uncoveredWorkOrderIds: string[];
  /**
   * Part identities ({@link mmPartIdentityKey}) used by both an uncovered work order being
   * written and a covered fetched work order. Their part row is left to the bulk sync and
   * only the part operations the written operations need, and the bulk sync did not write,
   * are included.
   */
  sharedPartIdentities: string[];
}

/**
 * The one definition of "the same MM part" for SDK-side grouping: raw part number and
 * revision (`null` / `undefined` → ""), `method` defaulting to "Standard". No trimming or
 * case folding — the MM constructors send these values as given. Used by
 * `buildWorkOrderBatchFromFlattened` to dedupe rows into parts and by
 * {@link computeWorkOrderGapFillWriteScope} to select a work order's part rows.
 */
export function mmPartIdentityKey(row: {
  partNumber?: string | number | null;
  partRevision?: string | number | null;
  method?: string | null;
}): string {
  return [
    row.partNumber === undefined || row.partNumber === null ? "" : String(row.partNumber),
    row.partRevision === undefined || row.partRevision === null ? "" : String(row.partRevision),
    row.method || "Standard",
  ].join("|");
}
const partIdentity = mmPartIdentityKey;

/**
 * Decide what to write by whether the bulk sync covered each row this cycle.
 *
 * Coverage is judged at the finest identity the bulk rows can express. The fetch and the
 * watch list work per work order id (the ERP job), but MM identifies a work order by
 * `workOrderId + lot + split + sub`, and the bulk window may hold one lot or sub-assembly of
 * a job but not its siblings. A fetched work order row or operation is COVERED when a bulk
 * row matches its id and sub (empty sub → "0", as in the labor key) and, where the bulk rows
 * carry `lot` / `split`, those too. Bulk rows without lot / split cover every lot / split of
 * that id + sub.
 *
 * - A COVERED row has a live competing writer. The gap fill's mapping may derive fields
 *   differently (Epicor's JobEntry path vs its BAQ/SQL paths differ on finish quantity,
 *   quantity per part and cycle time), so rewriting its siblings would make the two paths
 *   overwrite each other every cycle. Only its individually missing operations are written;
 *   the work order row and its parts are left to the bulk sync, which wrote them this cycle.
 * - An UNCOVERED row has no other writer, so a missing operation on it brings its work
 *   order row and part / part operations along — which it needs: a row that has never been
 *   inside the window may be entirely new to MM, and an operation whose parents are absent
 *   violates its FK.
 * - A CLOSING work order id (every fetched row terminal, and at least one row uncovered)
 *   writes each UNCOVERED row and its full operation set, so terminal status and closed
 *   date reach MM before it leaves the watch list. Its covered rows are left to the bulk
 *   sync, which owns them; a terminal id the bulk sync covers entirely is not closing.
 * - Parts are SHARED across work orders, so the per-row gate is not enough for them: an
 *   uncovered row being written and a bulk-written work order for the same part would have
 *   the gap fill rewrite the part operations the bulk sync wrote. For such a part the part
 *   row is left to the bulk sync, and only the part operations the written operations
 *   need, minus the sequences the bulk sync wrote for that part, are included. Bulk
 *   ownership is seen through the covered fetched rows and, when the bulk rows carry
 *   `part_number` / `part_revision` / `method`, through every bulk row — without part
 *   identity on the bulk rows, a bulk work order for the same part that was not fetched
 *   cannot be seen.
 * - A missing operation on a COVERED row is written alone, plus its part operation when
 *   no bulk row is known to own it: the bulk mapping may have dropped the operation and
 *   its part operation together, and the operation's FK needs the part operation.
 *
 * `keySetWorkOrderOps` is every operation of `fullBatch`, not only the written ones: a
 * ticket that CLOSED since last cycle no longer produces a missing key, and narrowing the
 * key set to the written ops would reject its clock-out.
 */
export function computeWorkOrderGapFillWriteScope(
  input: ComputeWorkOrderGapFillWriteScopeInput
): ComputeWorkOrderGapFillWriteScopeResult {
  const { fullBatch, missingOpKeys } = input;
  const parts = fullBatch?.parts ?? [];
  const partOperations = fullBatch?.partOperations ?? [];
  const workOrders = fullBatch?.workOrders ?? [];
  const workOrderOperations = fullBatch?.workOrderOperations ?? [];

  const bulkRows = input.bulkSyncedWorkOrderOperations ?? [];
  const lotSplitAware = isLotSplitAwareRowSet(bulkRows);

  // Bulk coverage per id + sub, each entry carrying the lot / split it names (undefined =
  // any) and the sequence it wrote. Sub normalizes as the labor key does (empty → "0").
  const subKey = (sub: unknown): string => exactKey(sub) || "0";
  const idSubKey = (id: unknown, sub: unknown): string => `${exactKey(id)}\u0000${subKey(sub)}`;
  type BulkEntry = { lot?: string; split?: string; sequenceNumber: string; partKey?: string };
  const bulkByIdSub = new Map<string, BulkEntry[]>();
  const partOpKey = (partKey: string, sequenceNumber: unknown): string =>
    `${partKey}\u0000${exactKey(sequenceNumber)}`;
  // Parts (and the part operations) the bulk sync wrote, as far as its rows say so. Rows
  // carrying part identity make bulk work orders the gap fill did not fetch visible here.
  const bulkPartKeys = new Set<string>();
  const bulkOwnedPartOpKeys = new Set<string>();
  for (const row of bulkRows) {
    const id = exactKey(row?.work_order_id);
    if (!id) continue;
    const key = idSubKey(id, row?.sub);
    const entries = bulkByIdSub.get(key) ?? [];
    const hasPart = row?.part_number !== undefined && row?.part_number !== null;
    const partKey = hasPart
      ? mmPartIdentityKey({
          partNumber: String(row.part_number),
          partRevision:
            row.part_revision === undefined || row.part_revision === null
              ? ""
              : String(row.part_revision),
          method: row.method === undefined || row.method === null ? "" : String(row.method),
        })
      : undefined;
    entries.push({
      lot: row?.lot === undefined ? undefined : exactKey(row.lot),
      split: row?.split === undefined ? undefined : exactKey(row.split),
      sequenceNumber: exactKey(row?.sequence_number),
      partKey,
    });
    bulkByIdSub.set(key, entries);
    if (partKey !== undefined) {
      bulkPartKeys.add(partKey);
      bulkOwnedPartOpKeys.add(partOpKey(partKey, row?.sequence_number));
    }
  }
  type RowIdentity = Pick<MMSendWorkOrder, "workOrderId" | "lot" | "split" | "sub">;
  const bulkEntriesFor = (row: RowIdentity): BulkEntry[] => {
    const lot = exactKey(row.lot);
    const split = exactKey(row.split);
    return (bulkByIdSub.get(idSubKey(row.workOrderId, row.sub)) ?? []).filter(
      (e) => (e.lot === undefined || e.lot === lot) && (e.split === undefined || e.split === split)
    );
  };
  const isRowCovered = (row: RowIdentity): boolean => bulkEntriesFor(row).length > 0;
  const rowKey = (row: RowIdentity): string =>
    [exactKey(row.workOrderId), exactKey(row.lot), exactKey(row.split), subKey(row.sub)].join("\u0000");

  // Per fetched id: does it have a row the bulk sync did not cover? Exact spelling, as
  // every write decision here is.
  const hasUncoveredRow = new Map<string, boolean>();
  for (const row of [...workOrders, ...workOrderOperations]) {
    const id = exactKey(row.workOrderId);
    if (!id) continue;
    hasUncoveredRow.set(id, (hasUncoveredRow.get(id) ?? false) || !isRowCovered(row));
  }
  const closing = new Set<string>();
  for (const id of input.terminalWorkOrderIds ?? []) {
    const key = exactKey(id);
    if (key && hasUncoveredRow.get(key)) closing.add(key);
  }
  const isClosingRow = (row: RowIdentity): boolean =>
    closing.has(exactKey(row.workOrderId)) && !isRowCovered(row);

  const opsToWrite = workOrderOperations.filter(
    (op) => isClosingRow(op) || missingOpKeys.has(sendOpKey(op, lotSplitAware))
  );

  // Bulk-covered ids feed the watch list, which keys ids case-insensitively, so they are
  // folded the same way: a job spelled two ways is covered only when every spelling is,
  // and then every spelling is reported (as the driver does for terminal ids). Reporting a
  // fully covered spelling alone would drop the shared entry while a sibling spelling
  // still has an uncovered row, and with it any pending close-out of that row.
  //
  // An id with an operation written this cycle (a missing operation on a covered row) is
  // not reported either: its watch-list entry is what retries a failed write once the
  // ticket has closed, since by default only OPEN tickets drive a fetch. A later fetch
  // that finds nothing to write releases it.
  const foldedIdsWithWrites = new Set(opsToWrite.map((op) => upper(op.workOrderId)));
  const spellingsByFoldedId = new Map<string, string[]>();
  for (const id of hasUncoveredRow.keys()) {
    const spellings = spellingsByFoldedId.get(upper(id)) ?? [];
    spellings.push(id);
    spellingsByFoldedId.set(upper(id), spellings);
  }
  const bulkCoveredWorkOrderIds = [...spellingsByFoldedId.entries()]
    .filter(
      ([folded, spellings]) =>
        !foldedIdsWithWrites.has(folded) && spellings.every((id) => !hasUncoveredRow.get(id))
    )
    .flatMap(([, spellings]) => spellings);

  // Work order rows whose parents are written, by composite identity. Seeded with every
  // uncovered row of a closing id: those rows must be written even when the fetched batch
  // carries no operations for them — otherwise the id leaves the watch list with their
  // terminal state never reaching MM.
  const uncoveredRowKeys = new Set<string>();
  for (const wo of workOrders) {
    if (isClosingRow(wo)) uncoveredRowKeys.add(rowKey(wo));
  }
  for (const op of opsToWrite) {
    if (!isRowCovered(op)) uncoveredRowKeys.add(rowKey(op));
  }

  const workOrdersToWrite = workOrders.filter((wo) => uncoveredRowKeys.has(rowKey(wo)));
  const partKeysToWrite = new Set(workOrdersToWrite.map(partIdentity));

  // Parts the bulk sync also wrote this cycle, with the part-operation sequences it wrote
  // for them: seen through the covered fetched rows (the bulk sequences of that row belong
  // to that row's part) and, when the bulk rows carry part identity, through every bulk
  // row. A part in `sharedPartKeys` keeps its part row with the bulk sync.
  const sharedPartKeys = new Set<string>();
  const partKeysByRow = new Map<string, Set<string>>();
  for (const wo of workOrders) {
    const key = rowKey(wo);
    const partKey = partIdentity(wo);
    const keys = partKeysByRow.get(key) ?? new Set<string>();
    keys.add(partKey);
    partKeysByRow.set(key, keys);
    const bulkEntries = bulkEntriesFor(wo);
    if (bulkEntries.length === 0) continue;
    for (const entry of bulkEntries) {
      bulkOwnedPartOpKeys.add(partOpKey(partKey, entry.sequenceNumber));
    }
    if (partKeysToWrite.has(partKey)) sharedPartKeys.add(partKey);
  }
  for (const partKey of partKeysToWrite) {
    if (bulkPartKeys.has(partKey)) sharedPartKeys.add(partKey);
  }

  // Part operations the written operations depend on. For an uncovered row they come with
  // the row's full part set unless the part is shared; for a covered row the operation is
  // written alone, so its part operation must be added when nothing else owns it — the
  // bulk mapping may have dropped the operation and its part operation together.
  const neededPartOpKeys = new Set<string>();
  const coveredNeededPartOpKeys = new Set<string>();
  for (const op of opsToWrite) {
    const key = rowKey(op);
    const target = uncoveredRowKeys.has(key) ? neededPartOpKeys : coveredNeededPartOpKeys;
    for (const partKey of partKeysByRow.get(key) ?? []) {
      target.add(partOpKey(partKey, op.sequenceNumber));
    }
  }

  const partsToWrite = parts.filter((p) => {
    const partKey = partIdentity(p);
    return partKeysToWrite.has(partKey) && !sharedPartKeys.has(partKey);
  });
  const partOpsToWrite = partOperations.filter((p) => {
    const partKey = partIdentity(p);
    const key = partOpKey(partKey, p.sequenceNumber);
    if (bulkOwnedPartOpKeys.has(key)) return false;
    if (coveredNeededPartOpKeys.has(key)) return true;
    if (!partKeysToWrite.has(partKey)) return false;
    if (!sharedPartKeys.has(partKey)) return true;
    return neededPartOpKeys.has(key);
  });

  const uncoveredWorkOrderIds = new Set<string>();
  for (const wo of workOrdersToWrite) uncoveredWorkOrderIds.add(exactKey(wo.workOrderId));
  for (const op of opsToWrite) {
    if (uncoveredRowKeys.has(rowKey(op))) uncoveredWorkOrderIds.add(exactKey(op.workOrderId));
  }

  return {
    batchToWrite: {
      parts: partsToWrite,
      partOperations: partOpsToWrite,
      workOrders: workOrdersToWrite,
      workOrderOperations: opsToWrite,
    },
    keySetWorkOrderOps: toSyncedWorkOrderOperationRows(workOrderOperations),
    closingWorkOrderIds: [...closing],
    bulkCoveredWorkOrderIds,
    uncoveredWorkOrderIds: [...uncoveredWorkOrderIds],
    sharedPartIdentities: [...sharedPartKeys],
  };
}

const TRACKER_KEY_PREFIX = "laborWorkOrderGapFill:v1:";

/**
 * Persists the gap-fill watch list in the SDK's cross-process runtime state store
 * (`sdk_runtime_state` in the local SQLite file), under its own key per connector type, so
 * it never shares a row with record tracking or any other feature.
 *
 * On edge devices the SQLite file lives in the container's writable layer and is wiped by
 * any container recreate. The watch list rebuilds from open labor on later cycles, so that
 * loss is tolerable; the only effect is that a work order which closed during the gap is
 * not re-fetched for close-out.
 *
 * `load()` / `save()` are a plain read-modify-write with no locking. That is safe for the
 * SDK's own use — only the `from-erp` job calls the driver, and Bree never overlaps a
 * running job — but a connector must not run the gap fill concurrently for the same
 * `connectorType`, or one call's save can overwrite the other's.
 */
export class LaborWorkOrderGapFillTracker {
  private readonly key: string;

  constructor(connectorType: string) {
    const suffix = exactKey(connectorType);
    if (!suffix) {
      throw new Error("LaborWorkOrderGapFillTracker requires a non-blank connectorType");
    }
    this.key = `${TRACKER_KEY_PREFIX}${suffix}`;
  }

  /**
   * The watch list, `[]` when none was saved. Throws when the store cannot be read: an
   * empty list in its place would be saved over the persisted one later in the cycle,
   * losing entries no current ticket references, and their pending close-outs with them.
   *
   * Content that is not a JSON array is the one exception. It cannot be recovered, and
   * failing on it would block every cycle until someone cleared it by hand, so it yields
   * `[]` with a warning and the next save replaces it. Malformed entries are dropped.
   */
  load(): TrackedLaborReferencedWorkOrder[] {
    const text = getRuntimeStateText(this.key);
    if (text === null) return [];

    let raw: unknown;
    let parseError: unknown;
    try {
      raw = JSON.parse(text);
    } catch (error) {
      parseError = error;
    }
    if (!Array.isArray(raw)) {
      logger.warn(
        "LaborWorkOrderGapFillTracker: the stored watch list is not a JSON array and cannot be recovered; starting empty, and the next save replaces it",
        { key: this.key, error: parseError }
      );
      return [];
    }
    return raw.filter(
      (entry): entry is TrackedLaborReferencedWorkOrder =>
        !!entry &&
        typeof entry === "object" &&
        typeof (entry as TrackedLaborReferencedWorkOrder).workOrderId === "string" &&
        typeof (entry as TrackedLaborReferencedWorkOrder).lastLaborSeenAt === "string"
    );
  }

  /** Replace the watch list. Throws on a store failure: the caller decides whether that is fatal. */
  save(tracked: ReadonlyArray<TrackedLaborReferencedWorkOrder>): void {
    setRuntimeStateJson(this.key, tracked);
  }
}