import { Clock, Effect, Latch } from 'effect';
import { uTime, type DecodedEntity } from '../../../core/index.js';
import type { CollectionItem } from '../../domain/collection-item/index.js';
import type { WriteError } from '../../domain/sync-error/index.js';
import type { CadenceConfig } from '../strategy/index.js';
import type { PartitionValue } from '../../domain/identity/index.js';

export type { CadenceConfig } from '../strategy/index.js';

// The repair loop reads the collection's in-memory rows synchronously; TanStack
// DB exposes no synchronous query method, only iteration over the current state.
export type SyncCollection<T> = {
  subscriberCount: number;
  on(
    event: 'subscribers:change',
    cb: (args: {
      previousSubscriberCount: number;
      subscriberCount: number;
    }) => void,
  ): () => void;
  values(): IterableIterator<CollectionItem<T>>;
};

type Meta<T> = DecodedEntity<T>['meta'];

const metaOf = <T>(row: CollectionItem<T>): Meta<T> | undefined =>
  (row as { _meta?: Meta<T> })._meta;

// Reconstruct the `{ value, meta }` entity shape from a flat collection row so
// the anchor handed to `fetchFrom` carries `meta._u`. Only `meta` is read
// downstream, but value is preserved (minus `_meta` and runtime virtual props).
const toEntity = <T>(row: CollectionItem<T>): DecodedEntity<T> => {
  const {
    _meta,
    $synced: _synced,
    $origin: _origin,
    ...value
  } = row as Record<string, unknown> &
    Partial<{ _meta: Meta<T>; $synced: unknown; $origin: unknown }>;
  return { value: value as T, meta: _meta! };
};

export type CadenceRepairDeps<T, E, R = never> = {
  collection: SyncCollection<T>;
  fetchFrom: (
    anchor: DecodedEntity<T> | null,
  ) => Effect.Effect<DecodedEntity<T>[], E, R>;
  applyToSyncReplica: (
    entities: DecodedEntity<T>[],
  ) => Effect.Effect<void, WriteError>;
  // Restricts the repair scope to a single partition: only rows whose flat
  // `field` equals `value` are considered. Omitted for unpartitioned collections.
  partition?: { field: string; value: PartitionValue } | undefined;
  config: CadenceConfig;
};

// `_u` is format-agnostic (ULID or ISO timestamp); `uTime` resolves either.
// A row whose `_u` fits neither format is a protocol violation — surfaced via
// `scanSuspects` so the repair loop fails instead of guessing.
const isSuspect = <T>(
  meta: DecodedEntity<T>['meta'],
  window: number,
): boolean | 'invalid-u' => {
  if (meta._s == null) return false;
  const uMs = uTime(meta._u);
  if (uMs == null) return 'invalid-u';
  return meta._s - uMs < window;
};

const msUntilReady = <T>(
  meta: DecodedEntity<T>['meta'],
  nowMs: number,
  readiness: number,
): number => {
  const skew = meta._c != null && meta._s != null ? meta._c - meta._s : 0;
  const elapsed = nowMs - skew - (uTime(meta._u) ?? 0);
  return Math.max(0, readiness - elapsed);
};

type Partition = { field: string; value: PartitionValue };

const inPartition = (
  row: Record<string, unknown>,
  partition: Partition | undefined,
): boolean =>
  partition == null || Object.is(row[partition.field], partition.value);

type SuspectScan<T> = {
  oldest: DecodedEntity<T> | undefined;
  suspectCount: number;
  scanned: number;
  invalidU: string | undefined;
};

// Single pass over the partition's rows: count suspects and keep the one with
// the oldest `_u` (the most likely to be ready for repair).
const scanSuspects = <T>(
  collection: SyncCollection<T>,
  window: number,
  partition: Partition | undefined,
  repaired: ReadonlySet<string>,
): SuspectScan<T> => {
  let scanned = 0;
  let suspectCount = 0;
  let oldestRow: CollectionItem<T> | undefined;
  let oldestU = '';
  for (const row of collection.values()) {
    const meta = metaOf(row);
    if (meta == null || !inPartition(row as Record<string, unknown>, partition))
      continue;
    scanned += 1;
    if (repaired.has(meta._u)) continue;
    const suspect = isSuspect<T>(meta, window);
    if (suspect === 'invalid-u') {
      return {
        oldest: undefined,
        suspectCount,
        scanned,
        invalidU: meta._u,
      };
    }
    if (!suspect) continue;
    suspectCount += 1;
    if (oldestRow === undefined || meta._u < oldestU) {
      oldestRow = row;
      oldestU = meta._u;
    }
  }
  return {
    oldest: oldestRow ? toEntity(oldestRow) : undefined,
    suspectCount,
    scanned,
    invalidU: undefined,
  };
};

// Newest row whose `_u` precedes the suspect: the repair re-fetches inclusively
// from this anchor so siblings that landed at the suspect's `_u` are picked up.
const queryPredecessor = <T>(
  collection: SyncCollection<T>,
  suspect: DecodedEntity<T>,
  partition: Partition | undefined,
): DecodedEntity<T> | undefined => {
  const suspectU = suspect.meta._u;
  let bestRow: CollectionItem<T> | undefined;
  let bestU = '';
  for (const row of collection.values()) {
    const meta = metaOf(row);
    if (meta == null || !inPartition(row as Record<string, unknown>, partition))
      continue;
    if (!(meta._u < suspectU)) continue;
    if (bestRow === undefined || meta._u > bestU) {
      bestRow = row;
      bestU = meta._u;
    }
  }
  return bestRow ? toEntity(bestRow) : undefined;
};

export const runCadenceRepair = <T, E, R = never>(
  deps: CadenceRepairDeps<T, E, R>,
): Effect.Effect<void, WriteError, R> =>
  Effect.scoped(
    Effect.gen(function* () {
      const { collection, fetchFrom, applyToSyncReplica, partition, config } =
        deps;
      const repaired = new Set<string>();

      const dbg = (event: string, data: Record<string, unknown>) =>
        config.debug
          ? Effect.logDebug({
              scope: 'sync.cadence-repair',
              event,
              partition: partition?.value,
              ...data,
            })
          : Effect.void;

      const latch = yield* Latch.make(collection.subscriberCount > 0);

      yield* dbg('start', {
        subscriberCount: collection.subscriberCount,
        window: config.window,
        readiness: config.readiness,
      });

      const offSubscribersChange = collection.on(
        'subscribers:change',
        ({ subscriberCount }) => {
          if (subscriberCount > 0) {
            latch.openUnsafe();
          } else {
            latch.closeUnsafe();
          }
        },
      );

      yield* Effect.addFinalizer(() => Effect.sync(offSubscribersChange));

      const loop: Effect.Effect<void, WriteError, R> = Effect.gen(function* () {
        yield* latch.await;

        const {
          oldest: suspect,
          suspectCount,
          scanned,
          invalidU,
        } = scanSuspects(collection, config.window, partition, repaired);

        if (invalidU !== undefined) {
          return yield* Effect.fail({
            _tag: 'Invalid',
            reason: `cadence repair stopped: _u is neither a ULID nor an ISO timestamp: ${JSON.stringify(invalidU)}`,
          } as WriteError);
        }

        if (!suspect) {
          yield* dbg('idle', { scanned, suspectCount: 0 });
          yield* Effect.sleep(config.pollDelay);
          return yield* loop;
        }

        const nowMs = yield* Clock.currentTimeMillis;

        const waitMs = msUntilReady(suspect.meta, nowMs, config.readiness);
        if (waitMs > 0) {
          yield* dbg('wait', {
            scanned,
            suspectCount,
            u: suspect.meta._u,
            waitMs,
          });
          yield* Effect.sleep(waitMs);
          return yield* loop;
        }

        const predecessor = queryPredecessor(collection, suspect, partition);
        const anchor = predecessor ?? null;
        yield* dbg('repair:fetch', {
          scanned,
          suspectCount,
          suspectU: suspect.meta._u,
          anchorU: anchor?.meta._u ?? null,
        });
        const results = yield* fetchFrom(anchor).pipe(
          Effect.mapError(
            (e) => ({ _tag: 'Invalid', reason: String(e) }) as WriteError,
          ),
        );
        yield* applyToSyncReplica(results);
        repaired.add(suspect.meta._u);
        yield* dbg('repair:wrote', { fetched: results.length });
        yield* Effect.sleep(config.pollDelay);

        return yield* loop;
      });

      yield* loop;
    }),
  );