import { Prisma } from '@prisma/client';
import { Logger } from '../logger';
import {
  BatchUploadCandidate,
  CONFLICT_DETECTION_ENTITY_BATCH_SIZE,
  ConflictResult,
  DuplicateOperationCandidate,
  LatestBatchEntityOperationRow,
  LatestEntityOperationRow,
  Operation,
  VectorClock,
  compareVectorClocks,
  isFullStateOpType,
  limitVectorClockSize,
} from './sync.types';

/**
 * Check if an incoming operation conflicts with existing operations.
 * Returns conflict info if a concurrent modification is detected.
 */
export const detectConflict = async (
  userId: number,
  op: Operation,
  tx: Prisma.TransactionClient,
): Promise<ConflictResult> => {
  // Skip conflict detection for full-state operations
  if (
    op.opType === 'SYNC_IMPORT' ||
    op.opType === 'BACKUP_IMPORT' ||
    op.opType === 'REPAIR'
  ) {
    return { hasConflict: false };
  }

  // Build list of entity IDs to check for conflicts.
  // Operations may have either entityId (singular) or entityIds (batch operations).
  const rawEntityIdsToCheck = op.entityIds?.length
    ? op.entityIds
    : op.entityId
      ? [op.entityId]
      : [];

  // Skip if no entity IDs (can't have entity-level conflicts)
  if (rawEntityIdsToCheck.length === 0) {
    return { hasConflict: false };
  }

  if (rawEntityIdsToCheck.length === 1) {
    return detectConflictForEntity(userId, op, rawEntityIdsToCheck[0], tx);
  }

  const entityIdsToCheck = Array.from(new Set(rawEntityIdsToCheck));
  return detectConflictForEntities(userId, op, entityIdsToCheck, tx);
};

export const detectConflictForEntities = async (
  userId: number,
  op: Operation,
  entityIdsToCheck: string[],
  tx: Prisma.TransactionClient,
): Promise<ConflictResult> => {
  for (
    let start = 0;
    start < entityIdsToCheck.length;
    start += CONFLICT_DETECTION_ENTITY_BATCH_SIZE
  ) {
    const batchEntityIds = entityIdsToCheck.slice(
      start,
      start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE,
    );
    // Match each requested id against the op's full entity set: the entity_ids
    // array UNION the scalar entity_id. The scalar is always folded in (not just
    // when entity_ids is empty) so an op whose entity_id is NOT a member of its
    // own entity_ids — possible when a multi-entity op dedups to a different
    // primary, see getStoredEntityIds — still exposes that scalar entity here.
    // DISTINCT ON dedupes the harmless overlap when entity_id is already in the
    // array (the common entity_id = entityIds[0] case). The `&&`/`= ANY`
    // prefilter keeps the GIN(entity_ids) + entity_id indexes usable; `eid = ANY`
    // keeps only the requested entities, then DISTINCT ON picks the latest op per
    // entity. Kept inline (not a shared fragment) so the positional params stay
    // stable for the conflict-detection.spec mock; the same shape is duplicated in
    // prefetchLatestEntityOpsForBatch — keep them in sync. (#8334)
    const idArray = Prisma.sql`ARRAY[${Prisma.join(batchEntityIds)}]::text[]`;
    const latestOps = await tx.$queryRaw<LatestEntityOperationRow[]>`
      SELECT DISTINCT ON (eid)
        eid AS "entityId",
        o.client_id AS "clientId",
        o.vector_clock AS "vectorClock"
      FROM operations o
      CROSS JOIN LATERAL unnest(
        o.entity_ids || CASE WHEN o.entity_id IS NULL THEN '{}'::text[] ELSE ARRAY[o.entity_id] END
      ) AS eid
      WHERE o.user_id = ${userId}
        AND o.entity_type = ${op.entityType}
        AND (o.entity_ids && ${idArray} OR o.entity_id = ANY(${idArray}))
        AND eid = ANY(${idArray})
      ORDER BY eid, o.server_seq DESC
    `;

    const latestOpByEntityId = new Map<string, LatestEntityOperationRow>();
    for (const latestOp of latestOps) {
      latestOpByEntityId.set(latestOp.entityId, latestOp);
    }

    for (const entityId of batchEntityIds) {
      const existingOp = latestOpByEntityId.get(entityId);
      if (!existingOp) continue;

      const conflictResult = resolveConflictForExistingOp(op, entityId, existingOp);
      if (conflictResult.hasConflict) {
        return conflictResult;
      }
    }
  }

  return { hasConflict: false };
};

export const resolveConflictForExistingOp = (
  op: Operation,
  entityId: string,
  existingOp: { clientId: string; vectorClock: unknown },
): ConflictResult => {
  // Stored JSON/vector_clock values arrive as unknown from both Prisma model
  // reads and raw SQL rows; cast only at the vector-clock comparison boundary.
  const existingClock = existingOp.vectorClock as unknown as VectorClock;

  // Compare vector clocks
  const comparison = compareVectorClocks(op.vectorClock, existingClock);

  // If the incoming op's clock is GREATER_THAN existing, it's a valid successor
  if (comparison === 'GREATER_THAN') {
    return { hasConflict: false };
  }

  // If clocks are EQUAL, this might be a retry of the same operation - check if from same client
  if (comparison === 'EQUAL' && op.clientId === existingOp.clientId) {
    return { hasConflict: false };
  }

  // EQUAL clocks from different clients is suspicious - treat as conflict
  // This could happen if client IDs rotate or clocks are somehow reused
  if (comparison === 'EQUAL') {
    return {
      hasConflict: true,
      conflictType: 'equal_different_client',
      reason: `Equal vector clocks from different clients for ${op.entityType}:${entityId} (client ${op.clientId} vs ${existingOp.clientId})`,
      existingClock,
    };
  }

  // CONCURRENT means both clocks have entries the other doesn't
  if (comparison === 'CONCURRENT') {
    return {
      hasConflict: true,
      conflictType: 'concurrent',
      reason: `Concurrent modification detected for ${op.entityType}:${entityId}`,
      existingClock,
    };
  }

  // LESS_THAN means the incoming op is older than what we have
  if (comparison === 'LESS_THAN') {
    return {
      hasConflict: true,
      conflictType: 'superseded',
      reason: `Superseded operation: server has newer version of ${op.entityType}:${entityId}`,
      existingClock,
    };
  }

  // Should never reach here - all comparison cases handled above
  // But if we do, default to conflict for safety
  return {
    hasConflict: true,
    conflictType: 'unknown',
    reason: `Unknown vector clock comparison result for ${op.entityType}:${entityId}`,
    existingClock,
  };
};

/**
 * Checks conflicts for the common single-entity upload path using Prisma's
 * typed model API. Multi-entity operations use the batched raw-SQL path above
 * to avoid one round trip per entity.
 */
export const detectConflictForEntity = async (
  userId: number,
  op: Operation,
  entityId: string,
  tx: Prisma.TransactionClient,
): Promise<ConflictResult> => {
  // Get the latest op that touched this entity. A multi-entity (batch) op stores
  // its full set in entity_ids, so match the entity as the scalar entity_id OR a
  // member of entity_ids. Single-entity ops store an empty array and are matched
  // via the scalar; pre-migration rows likewise fall back to the scalar (#8334).
  //
  // PERF: the OR spans the entity_id btree + the entity_ids GIN, so the planner
  // uses a BitmapOr + sort rather than an ordered LIMIT-1 walk. Bounded by one
  // entity's stored version depth (op-log pruning keeps that small, so the sort is
  // sub-ms in practice). If a real-Postgres EXPLAIN on a deep-history entity ever
  // shows this hot path (run per single-entity upload) is a problem, split into two
  // ordered LIMIT-1 lookups (scalar btree + entity_ids GIN) and take the higher
  // server_seq — the array branch stays small because entity_ids is multi-entity-only.
  // The batch unnest paths (detectConflictForEntities / prefetchLatestEntityOpsForBatch)
  // carry the larger sort, so EXPLAIN those first under heavy-user latency.
  const existingOp = await tx.operation.findFirst({
    where: {
      userId,
      entityType: op.entityType,
      OR: [{ entityId }, { entityIds: { has: entityId } }],
    },
    select: { clientId: true, vectorClock: true },
    orderBy: { serverSeq: 'desc' },
  });

  // No existing operation = no conflict
  if (!existingOp) {
    return { hasConflict: false };
  }

  return resolveConflictForExistingOp(op, entityId, existingOp);
};

export const isSameDuplicateOperation = (
  existingOp: DuplicateOperationCandidate,
  userId: number,
  op: Operation,
  maxClockDriftMs: number,
  originalTimestamp: number = op.timestamp,
): boolean => {
  const storedVectorClock = limitVectorClockSize(op.vectorClock, [op.clientId]);
  const incomingEncrypted = op.isPayloadEncrypted ?? false;

  // encrypt() uses a fresh random IV per call, so retried ciphertext differs
  // from the stored bytes. Skip byte-equality when both sides are encrypted;
  // the structural fields below still guard against id collisions.
  const payloadsMatch =
    (existingOp.isPayloadEncrypted && incomingEncrypted) ||
    areJsonValuesEqual(existingOp.payload, op.payload);

  return (
    existingOp.userId === userId &&
    existingOp.clientId === op.clientId &&
    existingOp.actionType === op.actionType &&
    existingOp.opType === op.opType &&
    existingOp.entityType === op.entityType &&
    existingOp.entityId === (op.entityId ?? null) &&
    payloadsMatch &&
    areJsonValuesEqual(existingOp.vectorClock, storedVectorClock) &&
    existingOp.schemaVersion === op.schemaVersion &&
    isSameDuplicateTimestamp(
      existingOp.clientTimestamp,
      existingOp.receivedAt,
      op.timestamp,
      originalTimestamp,
      maxClockDriftMs,
    ) &&
    existingOp.isPayloadEncrypted === incomingEncrypted &&
    existingOp.syncImportReason === (op.syncImportReason ?? null)
  );
};

export const isSameDuplicateTimestamp = (
  existingTimestamp: bigint | number | string,
  existingReceivedAt: bigint | number | string,
  incomingStoredTimestamp: number,
  incomingOriginalTimestamp: number,
  maxClockDriftMs: number,
): boolean => {
  if (existingTimestamp.toString() === incomingStoredTimestamp.toString()) {
    return true;
  }

  // A retry can arrive after server time advances enough that the original
  // timestamp no longer needs clamping, so original===stored does not rule
  // out a duplicate.
  // Future timestamps are clamped at receive time. A retry of the same op may
  // be clamped to a later value, or stop clamping once server time reaches the
  // original timestamp's allowed drift window. Allow exact-content duplicates
  // whose stored timestamp came from an earlier clamp of that same original
  // client timestamp.
  const existingTimestampValue = BigInt(existingTimestamp);
  const existingReceivedAtValue = BigInt(existingReceivedAt);
  return (
    existingTimestampValue === existingReceivedAtValue + BigInt(maxClockDriftMs) &&
    existingTimestampValue <= BigInt(incomingOriginalTimestamp)
  );
};

export const areJsonValuesEqual = (a: unknown, b: unknown): boolean => {
  return stableJsonStringify(a) === stableJsonStringify(b);
};

export const stableJsonStringify = (value: unknown): string => {
  return JSON.stringify(toStableJsonValue(value)) ?? 'undefined';
};

export const toStableJsonValue = (value: unknown): unknown => {
  if (Array.isArray(value)) {
    return value.map((item) => toStableJsonValue(item));
  }

  if (value !== null && typeof value === 'object') {
    return Object.fromEntries(
      Object.keys(value as Record<string, unknown>)
        .sort()
        .map((key) => [key, toStableJsonValue((value as Record<string, unknown>)[key])]),
    );
  }

  return value;
};

/**
 * All entities an op touches (full set, including single-entity ops). Use for the
 * incoming-op conflict check and the in-memory in-batch map — NOT for the persisted
 * `entity_ids` column, which uses {@link getStoredEntityIds} (multi-entity only).
 */
export const getConflictEntityIds = (op: Operation): string[] => {
  const rawEntityIds = op.entityIds?.length
    ? op.entityIds
    : op.entityId
      ? [op.entityId]
      : [];
  return Array.from(new Set(rawEntityIds));
};

/**
 * The entity_ids array to persist with an op. Ops whose touched-entity set is
 * already covered by the scalar entity_id store an empty array, so single-entity
 * ops (the vast majority) stay out of the entity_ids GIN index — keeping it small
 * and off their insert write path (Postgres GIN indexes no keys for an empty
 * array). Any other set is stored in full.
 *
 * The gate is "is the set exactly [entity_id]?", NOT "length > 1": a batch op
 * whose ids dedup to a single value that differs from entity_id (the server does
 * not enforce entity_id === entityIds[0]) must still be stored, or that entity
 * would be invisible to conflict lookups (#8334).
 */
export const getStoredEntityIds = (op: Operation): string[] => {
  const ids = getConflictEntityIds(op);
  if (ids.length <= 1 && ids[0] === op.entityId) {
    return [];
  }
  return ids;
};

export const getEntityConflictKey = (entityType: string, entityId: string): string => {
  return `${entityType}\u0000${entityId}`;
};

export const getBatchConflictEntityPairs = (
  candidates: BatchUploadCandidate[],
): { entityType: string; entityId: string }[] => {
  const entityPairs = new Map<string, { entityType: string; entityId: string }>();

  for (const candidate of candidates) {
    if (isFullStateOpType(candidate.op.opType)) continue;

    for (const entityId of getConflictEntityIds(candidate.op)) {
      entityPairs.set(getEntityConflictKey(candidate.op.entityType, entityId), {
        entityType: candidate.op.entityType,
        entityId,
      });
    }
  }

  return Array.from(entityPairs.values());
};

export const prefetchLatestEntityOpsForBatch = async (
  userId: number,
  entityPairs: { entityType: string; entityId: string }[],
  tx: Prisma.TransactionClient,
): Promise<Map<string, LatestBatchEntityOperationRow>> => {
  const latestByEntity = new Map<string, LatestBatchEntityOperationRow>();
  if (entityPairs.length === 0) return latestByEntity;

  for (
    let start = 0;
    start < entityPairs.length;
    start += CONFLICT_DETECTION_ENTITY_BATCH_SIZE
  ) {
    const batchPairs = entityPairs.slice(
      start,
      start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE,
    );
    const touchedRows = batchPairs.map(
      ({ entityType, entityId }) => Prisma.sql`(${entityType}, ${entityId})`,
    );
    // Prefilter array (all requested ids) so the JOIN below can match a requested
    // id inside a stored op's entity_ids set, not just its scalar entity_id, while
    // keeping the GIN(entity_ids) + entity_id indexes usable. The unnest folds the
    // scalar entity_id into the entity_ids set (UNION, deduped by DISTINCT ON) so a
    // divergent scalar is never missed — see the matching note in
    // detectConflictForEntities; keep the two shapes in sync. (#8334)
    const idArray = Prisma.sql`ARRAY[${Prisma.join(
      batchPairs.map((pair) => pair.entityId),
    )}]::text[]`;

    const latestOps = await tx.$queryRaw<LatestBatchEntityOperationRow[]>`
      SELECT DISTINCT ON (o.entity_type, eid)
        o.entity_type AS "entityType",
        eid AS "entityId",
        o.client_id AS "clientId",
        o.vector_clock AS "vectorClock"
      FROM operations o
      CROSS JOIN LATERAL unnest(
        o.entity_ids || CASE WHEN o.entity_id IS NULL THEN '{}'::text[] ELSE ARRAY[o.entity_id] END
      ) AS eid
      JOIN (VALUES ${Prisma.join(touchedRows)}) AS touched(entity_type, entity_id)
        ON touched.entity_type = o.entity_type
       AND touched.entity_id = eid
      WHERE o.user_id = ${userId}
        AND (o.entity_ids && ${idArray} OR o.entity_id = ANY(${idArray}))
      ORDER BY o.entity_type, eid, o.server_seq DESC
    `;

    for (const latestOp of latestOps) {
      latestByEntity.set(
        getEntityConflictKey(latestOp.entityType, latestOp.entityId),
        latestOp,
      );
    }
  }

  return latestByEntity;
};

export const pruneVectorClockForStorage = (op: Operation): void => {
  const beforeSize = Object.keys(op.vectorClock).length;
  op.vectorClock = limitVectorClockSize(op.vectorClock, [op.clientId]);
  const afterSize = Object.keys(op.vectorClock).length;
  if (afterSize < beforeSize) {
    Logger.debug(
      `[client:${op.clientId}] Vector clock pruned from ${beforeSize} to ${afterSize} before storage`,
    );
  }
};
