import {
  newFlock,
  get_ffi,
  get_entry_ffi,
  put_json_ffi,
  put_with_meta_ffi,
  delete_ffi,
  merge,
  set_peer_id,
  export_json_ffi,
  import_json_ffi,
  import_json_str_ffi,
  version_ffi,
  inclusiveVersion_ffi,
  get_max_physical_time_ffi,
  peer_id_ffi,
  kv_to_json_ffi,
  digest_hex_ffi,
  put_mvr_ffi,
  get_mvr_ffi,
  scan_ffi,
  subscribe_ffi,
  check_consistency_ffi,
  check_invariants_ffi,
  from_json_ffi,
  txn_begin_ffi,
  txn_commit_ffi,
  txn_rollback_ffi,
  is_in_txn_ffi,
} from "./_moon_flock";

import {
  EventBatcher,
  type EventBatcherRuntime,
} from "../../packages/flock-sqlite/src/event-batcher";

type RawVersionVector = Record<string, [number, number]>;
type RawScanRow = { key: KeyPart[]; raw: ExportRecord; value?: Value };
type RawEventPayload = { data?: Value; metadata?: MetadataMap };
type RawEventEntry = {
  key?: KeyPart[];
  value?: Value;
  metadata?: MetadataMap;
  payload?: RawEventPayload;
};
type RawEventBatch = { source?: string; events?: RawEventEntry[] };
type RawEntryClock = {
  physicalTime?: number;
  logicalCounter?: number;
  peerId?: string;
};
type RawEntryInfo = {
  data?: Value;
  metadata?: MetadataMap;
  clock?: RawEntryClock;
};

type MaybePromise<T> = T | Promise<T>;

type ExportOptions = {
  from?: VersionVector;
  hooks?: ExportHooks;
  pruneTombstonesBefore?: number;
  peerId?: string;
};

type ImportOptions = {
  bundle: ExportBundle;
  hooks?: ImportHooks;
};

type RawImportReport = {
  accepted?: number;
  skipped?: Array<{ key?: KeyPart[]; reason?: string }>;
};

export type VersionVectorEntry = {
  physicalTime: number;
  logicalCounter: number;
};

export interface VersionVector {
  [peer: string]: VersionVectorEntry | undefined;
}

export function encodeVersionVector(vector: VersionVector): Uint8Array {
  return encodeVersionVectorBinary(vector);
}

export function decodeVersionVector(bytes: Uint8Array): VersionVector {
  return decodeVersionVectorBinary(bytes);
}

export type Value =
  | string
  | number
  | boolean
  | null
  | Array<Value>
  | { [key: string]: Value };

export type KeyPart = Value;

export type MetadataMap = Record<string, unknown>;

export type ExportRecord = {
  c: string;
  d?: Value;
  m?: MetadataMap;
};

export type ExportBundle = {
  version: number;
  entries: Record<string, ExportRecord>;
};

export type EntryClock = {
  physicalTime: number;
  logicalCounter: number;
  peerId: string;
};

export type EntryInfo = {
  data?: Value;
  metadata: MetadataMap;
  clock: EntryClock;
};

export type ExportPayload = {
  data?: Value;
  metadata?: MetadataMap;
};

export type ExportHookContext = {
  key: KeyPart[];
  clock: EntryClock;
  raw: ExportRecord;
};

export type ExportHooks = {
  transform?: (
    context: ExportHookContext,
    payload: ExportPayload,
  ) => MaybePromise<ExportPayload | void>;
};

export type ImportPayload = ExportPayload;

export type ImportHookContext = ExportHookContext;

export type ImportAccept = {
  accept: true;
};

export type ImportSkip = {
  accept: false;
  reason: string;
};

export type ImportDecision = ImportAccept | ImportSkip | ImportPayload | void;

export type ImportHooks = {
  preprocess?: (
    context: ImportHookContext,
    payload: ImportPayload,
  ) => MaybePromise<ImportDecision>;
};

export type ImportReport = {
  accepted: number;
  skipped: Array<{ key: KeyPart[]; reason: string }>;
};

export type PutPayload = ExportPayload;

export type PutHookContext = {
  key: KeyPart[];
  now?: number;
};

export type PutHooks = {
  transform?: (
    context: PutHookContext,
    payload: PutPayload,
  ) => MaybePromise<PutPayload | void>;
};

export type PutWithMetaOptions = {
  metadata?: MetadataMap;
  now?: number;
  hooks?: PutHooks;
};

export type ScanBound =
  | { kind: "inclusive"; key: KeyPart[] }
  | { kind: "exclusive"; key: KeyPart[] }
  | { kind: "unbounded" };

export type ScanOptions = {
  start?: ScanBound;
  end?: ScanBound;
  prefix?: KeyPart[];
};

export type ScanRow = {
  key: KeyPart[];
  raw: ExportRecord;
  value?: Value;
};

export type EventPayload = ExportPayload;

export type Event = {
  key: KeyPart[];
  value?: Value;
  metadata?: MetadataMap;
  payload: EventPayload;
};

export type EventBatch = {
  source: string;
  events: Event[];
};

const textEncoder = new TextEncoder();
const textDecoder = new TextDecoder();

function utf8ByteLength(value: string): number {
  return textEncoder.encode(value).length;
}

function isValidPeerId(peerId: unknown): peerId is string {
  return typeof peerId === "string" && utf8ByteLength(peerId) < 128;
}

function createRandomPeerId(): string {
  const id = new Uint8Array(32);
  if (
    typeof crypto !== "undefined" &&
    typeof crypto.getRandomValues === "function"
  ) {
    crypto.getRandomValues(id);
  } else {
    for (let i = 0; i < 32; i += 1) {
      id[i] = Math.floor(Math.random() * 256);
    }
  }
  return Array.from(id, (byte) => byte.toString(16).padStart(2, "0")).join("");
}

function normalizePeerId(peerId?: string): string {
  if (peerId === undefined) {
    return createRandomPeerId();
  }
  if (!isValidPeerId(peerId)) {
    throw new TypeError("peerId must be a UTF-8 string under 128 bytes.");
  }
  return peerId;
}

const SYNC_TXN_CALLBACK_ERROR = "Flock.txn callback must be synchronous";

function isAsyncCallback(callback: Function): boolean {
  return callback.constructor?.name === "AsyncFunction";
}

function isThenable(value: unknown): value is PromiseLike<unknown> {
  return (
    (typeof value === "object" || typeof value === "function") &&
    value !== null &&
    typeof (value as PromiseLike<unknown>).then === "function"
  );
}

type EncodableVersionVectorEntry = {
  peer: string;
  peerBytes: Uint8Array;
  timestamp: number;
  counter: number;
};

function comparePeerBytes(a: Uint8Array, b: Uint8Array): number {
  if (a === b) {
    return 0;
  }
  const limit = Math.min(a.length, b.length);
  for (let i = 0; i < limit; i += 1) {
    const diff = a[i] - b[i];
    if (diff !== 0) {
      return diff;
    }
  }
  return a.length - b.length;
}

function collectEncodableVersionVectorEntries(
  vv?: VersionVector,
): EncodableVersionVectorEntry[] {
  if (!vv || typeof vv !== "object") {
    return [];
  }
  const entries: EncodableVersionVectorEntry[] = [];
  for (const [peer, entry] of Object.entries(vv)) {
    if (!entry || !isValidPeerId(peer)) {
      continue;
    }
    const { physicalTime, logicalCounter } = entry;
    if (
      typeof physicalTime !== "number" ||
      !Number.isFinite(physicalTime) ||
      typeof logicalCounter !== "number" ||
      !Number.isFinite(logicalCounter)
    ) {
      continue;
    }
    const peerBytes = textEncoder.encode(peer);
    entries.push({
      peer,
      peerBytes,
      timestamp: Math.trunc(physicalTime),
      counter: Math.max(0, Math.trunc(logicalCounter)),
    });
  }
  entries.sort((a, b) => {
    if (a.timestamp !== b.timestamp) {
      return a.timestamp - b.timestamp;
    }
    const peerCmp = comparePeerBytes(a.peerBytes, b.peerBytes);
    if (peerCmp !== 0) {
      return peerCmp;
    }
    return a.counter - b.counter;
  });
  return entries;
}

function writeUnsignedLeb128(value: number, out: number[]): void {
  if (!Number.isFinite(value) || value < 0) {
    throw new TypeError("leb128 values must be finite and non-negative");
  }
  let remaining = Math.trunc(value);
  if (remaining === 0) {
    out.push(0);
    return;
  }
  while (remaining > 0) {
    const byte = remaining % 0x80;
    remaining = Math.floor(remaining / 0x80);
    out.push(remaining > 0 ? byte | 0x80 : byte);
  }
}

function writeVarStringBytes(bytes: Uint8Array, out: number[]): void {
  writeUnsignedLeb128(bytes.length, out);
  for (let i = 0; i < bytes.length; i += 1) {
    out.push(bytes[i]);
  }
}

const VERSION_VECTOR_MAGIC = new Uint8Array([86, 69, 86, 69]); // "VEVE"

function encodeVersionVectorBinary(vv?: VersionVector): Uint8Array {
  const entries = collectEncodableVersionVectorEntries(vv);
  const buffer: number[] = Array.from(VERSION_VECTOR_MAGIC);
  if (entries.length === 0) {
    return Uint8Array.from(buffer);
  }

  let lastTimestamp = 0;
  for (let i = 0; i < entries.length; i += 1) {
    const entry = entries[i];
    if (entry.timestamp < 0) {
      throw new TypeError("timestamp must be non-negative");
    }
    if (i === 0) {
      writeUnsignedLeb128(entry.timestamp, buffer);
      lastTimestamp = entry.timestamp;
    } else {
      const delta = entry.timestamp - lastTimestamp;
      if (delta < 0) {
        throw new TypeError("version vector timestamps must be non-decreasing");
      }
      writeUnsignedLeb128(delta, buffer);
      lastTimestamp = entry.timestamp;
    }

    writeUnsignedLeb128(entry.counter, buffer);
    writeVarStringBytes(entry.peerBytes, buffer);
  }

  return Uint8Array.from(buffer);
}

function decodeUnsignedLeb128(
  bytes: Uint8Array,
  offset: number,
): [number, number] {
  let result = 0;
  let multiplier = 1;
  let consumed = 0;
  while (offset + consumed < bytes.length) {
    const byte = bytes[offset + consumed];
    consumed += 1;
    // Use arithmetic instead of bitwise operations to avoid 32-bit overflow.
    // JavaScript bitwise operators convert to 32-bit signed integers,
    // which breaks for values >= 2^31.
    result += (byte & 0x7f) * multiplier;
    if ((byte & 0x80) === 0) {
      break;
    }
    multiplier *= 128;
  }
  return [result, consumed];
}

function decodeVarString(bytes: Uint8Array, offset: number): [string, number] {
  const [length, used] = decodeUnsignedLeb128(bytes, offset);
  const start = offset + used;
  const end = start + length;
  if (end > bytes.length) {
    throw new TypeError("varString length exceeds buffer");
  }
  const slice = bytes.subarray(start, end);
  return [textDecoder.decode(slice), used + length];
}

function hasMagic(bytes: Uint8Array): boolean {
  return (
    bytes.length >= 4 &&
    bytes[0] === VERSION_VECTOR_MAGIC[0] &&
    bytes[1] === VERSION_VECTOR_MAGIC[1] &&
    bytes[2] === VERSION_VECTOR_MAGIC[2] &&
    bytes[3] === VERSION_VECTOR_MAGIC[3]
  );
}

function decodeLegacyVersionVector(bytes: Uint8Array): VersionVector {
  let offset = 0;
  const [count, usedCount] = decodeUnsignedLeb128(bytes, offset);
  offset += usedCount;
  const [baseTimestamp, usedBase] = decodeUnsignedLeb128(bytes, offset);
  offset += usedBase;
  const vv: VersionVector = {};
  for (let i = 0; i < count; i += 1) {
    const [peer, usedPeer] = decodeVarString(bytes, offset);
    offset += usedPeer;
    if (!isValidPeerId(peer)) {
      throw new TypeError("invalid peer id in encoded version vector");
    }
    const [delta, usedDelta] = decodeUnsignedLeb128(bytes, offset);
    offset += usedDelta;
    const [counter, usedCounter] = decodeUnsignedLeb128(bytes, offset);
    offset += usedCounter;
    vv[peer] = {
      physicalTime: baseTimestamp + delta,
      logicalCounter: counter,
    };
  }
  return vv;
}

function decodeNewVersionVector(bytes: Uint8Array): VersionVector {
  let offset = 4;
  const vv: VersionVector = {};
  if (offset === bytes.length) {
    return vv;
  }

  const [firstTimestamp, usedTs] = decodeUnsignedLeb128(bytes, offset);
  offset += usedTs;
  const [firstCounter, usedCounter] = decodeUnsignedLeb128(bytes, offset);
  offset += usedCounter;
  const [firstPeer, usedPeer] = decodeVarString(bytes, offset);
  offset += usedPeer;
  if (!isValidPeerId(firstPeer)) {
    throw new TypeError("invalid peer id in encoded version vector");
  }
  vv[firstPeer] = {
    physicalTime: firstTimestamp,
    logicalCounter: firstCounter,
  };

  let lastTimestamp = firstTimestamp;
  while (offset < bytes.length) {
    const [delta, usedDelta] = decodeUnsignedLeb128(bytes, offset);
    offset += usedDelta;
    const [counter, usedCtr] = decodeUnsignedLeb128(bytes, offset);
    offset += usedCtr;
    const [peer, usedPeerLen] = decodeVarString(bytes, offset);
    offset += usedPeerLen;
    if (!isValidPeerId(peer)) {
      throw new TypeError("invalid peer id in encoded version vector");
    }
    const timestamp = lastTimestamp + delta;
    if (timestamp < lastTimestamp) {
      throw new TypeError("version vector timestamps must be non-decreasing");
    }
    vv[peer] = { physicalTime: timestamp, logicalCounter: counter };
    lastTimestamp = timestamp;
  }

  return vv;
}

function decodeVersionVectorBinary(bytes: Uint8Array): VersionVector {
  if (hasMagic(bytes)) {
    return decodeNewVersionVector(bytes);
  }
  return decodeLegacyVersionVector(bytes);
}

function encodeVersionVectorForFfi(
  vv?: VersionVector,
): RawVersionVector | undefined {
  if (!vv) {
    return undefined;
  }
  const raw: RawVersionVector = {};
  for (const entry of collectEncodableVersionVectorEntries(vv)) {
    raw[entry.peer] = [entry.timestamp, entry.counter];
  }
  return raw;
}

function normalizePruneBefore(
  pruneTombstonesBefore?: number,
): number | undefined {
  if (pruneTombstonesBefore === undefined) {
    return undefined;
  }
  if (
    typeof pruneTombstonesBefore !== "number" ||
    !Number.isFinite(pruneTombstonesBefore)
  ) {
    return undefined;
  }
  return pruneTombstonesBefore;
}

function normalizeMutationNow(now?: number | null): number | undefined {
  if (now === undefined || now === null) {
    return undefined;
  }
  if (typeof now !== "number" || !Number.isFinite(now)) {
    throw new TypeError("now must be a finite number");
  }
  return now;
}

function decodeVersionVectorFromRaw(raw: unknown): VersionVector {
  if (raw === null || typeof raw !== "object") {
    return {};
  }
  const result: VersionVector = {};
  for (const [peer, value] of Object.entries(raw as Record<string, unknown>)) {
    if (!Array.isArray(value) || value.length < 2) {
      continue;
    }
    if (!isValidPeerId(peer)) {
      continue;
    }
    const [physicalTime, logicalCounter] = value;
    if (typeof physicalTime !== "number" || !Number.isFinite(physicalTime)) {
      continue;
    }
    if (
      typeof logicalCounter !== "number" ||
      !Number.isFinite(logicalCounter)
    ) {
      continue;
    }
    result[peer] = {
      physicalTime,
      logicalCounter: Math.trunc(logicalCounter),
    };
  }
  return result;
}

function encodeBound(bound?: ScanBound): Record<string, unknown> | undefined {
  if (bound === undefined) {
    return undefined;
  }
  if (!bound || typeof bound !== "object") {
    throw new TypeError("scan bound must be an object");
  }
  if (bound.kind === "unbounded") {
    return { kind: "unbounded" };
  }
  if (bound.kind !== "inclusive" && bound.kind !== "exclusive") {
    throw new TypeError(
      "scan bound kind must be inclusive, exclusive, or unbounded",
    );
  }
  if (!Array.isArray(bound.key)) {
    throw new TypeError("scan bound key must be a key array");
  }
  return { kind: bound.kind, key: cloneKey(bound.key) };
}

function normalizeScanPrefix(prefix: unknown): KeyPart[] | undefined {
  if (prefix === undefined) {
    return undefined;
  }
  if (!Array.isArray(prefix)) {
    throw new TypeError("scan prefix must be a key array");
  }
  return cloneKey(prefix as KeyPart[]);
}

function decodeEntryInfo(raw: unknown): EntryInfo | undefined {
  if (!raw || typeof raw !== "object") {
    return undefined;
  }
  const info = raw as RawEntryInfo;
  const clock = normalizeEntryClock(info.clock);
  if (!clock) {
    return undefined;
  }
  const metadata = normalizeMetadataMap(info.metadata);
  const result: EntryInfo = { metadata, clock };
  if ("data" in info) {
    result.data = cloneJson(info.data as Value);
  }
  return result;
}

function decodeEventBatch(raw: unknown): EventBatch {
  if (!raw || typeof raw !== "object") {
    return { source: "local", events: [] };
  }
  const batch = raw as RawEventBatch;
  const source = typeof batch.source === "string" ? batch.source : "local";
  const eventsRaw = Array.isArray(batch.events) ? batch.events : [];
  const events = eventsRaw
    .filter((entry): entry is RawEventEntry => Boolean(entry))
    .map((entry) => buildEvent(entry));
  return { source, events };
}

function buildEvent(entry: RawEventEntry): Event {
  const key = Array.isArray(entry.key) ? entry.key : [];
  const payload = buildEventPayload(entry);
  return {
    key,
    value: payload.data,
    metadata: cloneMetadata(payload.metadata),
    payload,
  };
}

function buildEventPayload(entry: RawEventEntry): EventPayload {
  const base: ExportPayload = {};
  if ("value" in entry) {
    base.data = cloneJson(entry.value as Value);
  }
  const entryMetadata = cloneMetadata(entry.metadata);
  if (entryMetadata !== undefined) {
    base.metadata = entryMetadata;
  }
  const update = normalizeRawEventPayload(entry.payload);
  return mergePayload(base, update);
}

function normalizeRawEventPayload(
  payload: RawEventPayload | undefined,
): ExportPayload | undefined {
  if (!payload || typeof payload !== "object") {
    return undefined;
  }
  const result: ExportPayload = {};
  if ("data" in payload) {
    result.data = cloneJson(payload.data as Value);
  }
  const metadata = cloneMetadata(payload.metadata);
  if (metadata !== undefined) {
    result.metadata = metadata;
  }
  return result;
}

const JSON_SERIALIZATION_ERROR = "Value is not JSON serializable";
const JSON_SERIALIZATION_WARNING_PREFIX = "[flock] JSON value normalized:";

function warnJsonNormalization(message: string): void {
  try {
    console.warn(`${JSON_SERIALIZATION_WARNING_PREFIX} ${message}`);
  } catch {
    // Ignore console failures so value normalization never becomes a write error.
  }
}

function normalizeInvalidJsonValue(message: string): null {
  warnJsonNormalization(message);
  return null;
}

function normalizeJsonValue(
  value: unknown,
  seen: WeakSet<object>,
  inObjectEntry = false,
): unknown {
  if (typeof value === "number" && !Number.isFinite(value)) {
    return normalizeInvalidJsonValue("non-finite number stored as null");
  }
  if (value === undefined) {
    return inObjectEntry ? undefined : null;
  }
  if (typeof value === "function" || typeof value === "symbol") {
    warnJsonNormalization(`${typeof value} value omitted`);
    return inObjectEntry ? undefined : null;
  }
  if (typeof value === "bigint") {
    if (
      value <= BigInt(Number.MAX_SAFE_INTEGER) &&
      value >= BigInt(Number.MIN_SAFE_INTEGER)
    ) {
      return Number(value);
    }
    return normalizeInvalidJsonValue(
      "bigint exceeds JavaScript safe integer range and was stored as null",
    );
  }
  if (!value || typeof value !== "object") {
    return value;
  }

  const toJson = (value as { toJSON?: unknown }).toJSON;
  if (typeof toJson === "function") {
    try {
      return normalizeJsonValue(toJson.call(value), seen, inObjectEntry);
    } catch {
      return normalizeInvalidJsonValue("toJSON threw and value was stored as null");
    }
  }

  if (seen.has(value)) {
    return normalizeInvalidJsonValue("circular reference stored as null");
  }
  seen.add(value);
  if (Array.isArray(value)) {
    const result = value.map((item) => {
      const normalized = normalizeJsonValue(item, seen);
      return normalized === undefined ? null : normalized;
    });
    seen.delete(value);
    return result;
  }

  let keys: string[];
  try {
    keys = Object.keys(value);
  } catch {
    seen.delete(value);
    return normalizeInvalidJsonValue("object keys could not be read and value was stored as null");
  }

  const result: Record<string, unknown> = {};
  for (const key of keys) {
    let rawValue: unknown;
    try {
      rawValue = (value as Record<string, unknown>)[key];
    } catch {
      result[key] = normalizeInvalidJsonValue(
        "object property could not be read and was stored as null",
      );
      continue;
    }
    const normalized = normalizeJsonValue(
      rawValue,
      seen,
      true,
    );
    if (normalized !== undefined) {
      result[key] = normalized;
    }
  }
  seen.delete(value);
  return result;
}

function validateJsonPart(value: unknown): unknown {
  if (typeof value === "number" && !Number.isFinite(value)) {
    throw new TypeError(JSON_SERIALIZATION_ERROR);
  }
  if (
    value === undefined ||
    typeof value === "function" ||
    typeof value === "symbol" ||
    typeof value === "bigint"
  ) {
    throw new TypeError(JSON_SERIALIZATION_ERROR);
  }
  return value;
}

function stringifyJson(value: unknown): string {
  const normalized = normalizeJsonValue(value, new WeakSet<object>());
  return JSON.stringify(normalized) ?? "null";
}

function stringifyStrictJson(value: unknown): string {
  const encoded = JSON.stringify(value, (_key, part) =>
    validateJsonPart(part),
  );
  if (encoded === undefined) {
    throw new TypeError(JSON_SERIALIZATION_ERROR);
  }
  return encoded;
}

function stringifyStoredJson(value: Value | undefined): string {
  return value === undefined ? "null" : stringifyJson(value);
}

function cloneJson<T>(value: T): T {
  if (value === undefined) {
    return value;
  }
  return JSON.parse(stringifyJson(value)) as T;
}

function cloneStrictJson<T>(value: T): T {
  if (value === undefined) {
    return value;
  }
  return JSON.parse(stringifyStrictJson(value)) as T;
}

function cloneStoredJson(value: Value | undefined): Value {
  return value === undefined ? null : cloneJson(value);
}

function cloneKey(key: KeyPart[]): KeyPart[] {
  if (!Array.isArray(key)) {
    throw new TypeError("key must be an array");
  }
  return cloneStrictJson(key) as KeyPart[];
}

function parseKeyString(key: string): KeyPart[] {
  try {
    const parsed = JSON.parse(key);
    return Array.isArray(parsed) ? (parsed as KeyPart[]) : [];
  } catch {
    return [];
  }
}

function cloneMetadata(metadata: unknown): MetadataMap | undefined {
  if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) {
    return undefined;
  }
  return cloneJson(metadata as MetadataMap);
}

function normalizeMetadataMap(metadata: unknown): MetadataMap {
  const cloned = cloneMetadata(metadata);
  return cloned ?? {};
}

function decodeClock(record: ExportRecord): EntryClock {
  const rawClock = typeof record.c === "string" ? record.c : "";
  const firstComma = rawClock.indexOf(",");
  const secondComma =
    firstComma === -1 ? -1 : rawClock.indexOf(",", firstComma + 1);
  if (firstComma === -1 || secondComma === -1) {
    return { physicalTime: 0, logicalCounter: 0, peerId: "" };
  }
  const physicalTime = Number(rawClock.slice(0, firstComma));
  const logicalCounter = Number(rawClock.slice(firstComma + 1, secondComma));
  const peerIdRaw = rawClock.slice(secondComma + 1);
  const peerId = isValidPeerId(peerIdRaw) ? peerIdRaw : "";
  return {
    physicalTime: Number.isFinite(physicalTime) ? physicalTime : 0,
    logicalCounter: Number.isFinite(logicalCounter)
      ? Math.trunc(logicalCounter)
      : 0,
    peerId,
  };
}

function normalizeEntryClock(
  clock: RawEntryClock | undefined,
): EntryClock | undefined {
  if (!clock || typeof clock !== "object") {
    return undefined;
  }
  const { physicalTime, logicalCounter, peerId } = clock;
  if (typeof physicalTime !== "number" || !Number.isFinite(physicalTime)) {
    return undefined;
  }
  if (typeof logicalCounter !== "number" || !Number.isFinite(logicalCounter)) {
    return undefined;
  }
  if (!isValidPeerId(peerId)) {
    return undefined;
  }
  return {
    physicalTime,
    logicalCounter: Math.trunc(logicalCounter),
    peerId,
  };
}

function createExportPayload(record: ExportRecord): ExportPayload {
  const payload: ExportPayload = {};
  if (record.d !== undefined) {
    payload.data = cloneJson(record.d);
  }
  const metadata = cloneMetadata(record.m);
  if (metadata !== undefined) {
    payload.metadata = metadata;
  }
  return payload;
}

function createPutPayload(
  value: Value | undefined,
  metadata?: MetadataMap,
): ExportPayload {
  const payload: ExportPayload = { data: cloneStoredJson(value) };
  const cleanMetadata = cloneMetadata(metadata);
  if (cleanMetadata !== undefined) {
    payload.metadata = cleanMetadata;
  }
  return payload;
}

function assignPayload(
  target: ExportPayload,
  source?: ExportPayload | void,
): void {
  if (!source || typeof source !== "object") {
    return;
  }
  if ("data" in source) {
    const value = source.data;
    target.data = value === undefined ? undefined : cloneJson(value);
  }
  if ("metadata" in source) {
    target.metadata = cloneMetadata(source.metadata);
  }
}

function clonePayload(payload: ExportPayload | undefined): ExportPayload {
  const result: ExportPayload = {};
  assignPayload(result, payload);
  return result;
}

function mergePayload(
  base: ExportPayload,
  update?: ExportPayload | void,
): ExportPayload {
  const result = clonePayload(base);
  assignPayload(result, update);
  return result;
}

function buildRecord(clock: string, payload: ExportPayload): ExportRecord {
  const record: ExportRecord = { c: clock };
  if (payload.data !== undefined) {
    record.d = cloneJson(payload.data);
  }
  const metadata = cloneMetadata(payload.metadata);
  if (metadata !== undefined) {
    record.m = metadata;
  }
  return record;
}

function cloneRecord(record: ExportRecord): ExportRecord {
  return buildRecord(record.c, createExportPayload(record));
}

function buildContext(key: string, record: ExportRecord): ExportHookContext {
  return {
    key: parseKeyString(key),
    clock: decodeClock(record),
    raw: cloneRecord(record),
  };
}

function normalizeImportDecision(
  decision: ImportDecision,
): ImportAccept | ImportSkip {
  if (!decision || typeof decision !== "object") {
    return { accept: true };
  }
  if ("accept" in decision) {
    if (!decision.accept) {
      return { accept: false, reason: decision.reason ?? "rejected" };
    }
    return { accept: true };
  }
  return { accept: true };
}

function payloadFromImportDecision(
  decision: ImportDecision,
): ImportPayload | undefined {
  if (!decision || typeof decision !== "object") {
    return undefined;
  }
  if ("accept" in decision) {
    return undefined;
  }
  if ("data" in decision || "metadata" in decision) {
    return decision as ImportPayload;
  }
  return undefined;
}

function decodeImportReport(raw: unknown): ImportReport {
  if (!raw || typeof raw !== "object") {
    return { accepted: 0, skipped: [] };
  }
  const report = raw as RawImportReport;
  const accepted = typeof report.accepted === "number" ? report.accepted : 0;
  const skippedRaw = Array.isArray(report.skipped) ? report.skipped : [];
  const skipped = skippedRaw.map((entry) => {
    const key = entry && Array.isArray(entry.key) ? entry.key : [];
    const reason =
      entry && typeof entry.reason === "string" ? entry.reason : "unknown";
    return { key, reason };
  });
  return { accepted, skipped };
}

function cloneBundle(bundle: ExportBundle): ExportBundle {
  const next: ExportBundle = { version: bundle.version, entries: {} };
  for (const [key, record] of Object.entries(bundle.entries)) {
    next.entries[key] = cloneRecord(record);
  }
  return next;
}

function isVersionVectorEntry(value: unknown): value is VersionVectorEntry {
  return (
    typeof value === "object" &&
    value !== null &&
    typeof (value as VersionVectorEntry).physicalTime === "number" &&
    Number.isFinite((value as VersionVectorEntry).physicalTime) &&
    typeof (value as VersionVectorEntry).logicalCounter === "number" &&
    Number.isFinite((value as VersionVectorEntry).logicalCounter)
  );
}

function isVersionVectorLike(value: unknown): value is VersionVector {
  return (
    typeof value === "object" &&
    value !== null &&
    Object.keys(value).length > 0 &&
    Object.values(value).every(isVersionVectorEntry)
  );
}

function isExportOptions(value: unknown): value is ExportOptions {
  return (
    typeof value === "object" &&
    value !== null &&
    !isVersionVectorLike(value) &&
    (Object.prototype.hasOwnProperty.call(value, "hooks") ||
      Object.prototype.hasOwnProperty.call(value, "from") ||
      Object.prototype.hasOwnProperty.call(value, "pruneTombstonesBefore") ||
      Object.prototype.hasOwnProperty.call(value, "peerId"))
  );
}

function isImportOptions(value: unknown): value is ImportOptions {
  return (
    typeof value === "object" &&
    value !== null &&
    Object.prototype.hasOwnProperty.call(value, "bundle")
  );
}

const defaultEventBatcherRuntime: EventBatcherRuntime = {
  now: () => Date.now(),
  setTimeout: (fn, ms) => setTimeout(fn, ms) as unknown,
  clearTimeout: (handle) => clearTimeout(handle as any),
};

export class Flock {
  private inner: ReturnType<typeof newFlock>;
  private listeners: Set<(batch: EventBatch) => void> = new Set();
  private nativeUnsubscribe: (() => void) | undefined;
  private readonly eventBatcher: EventBatcher<Event>;

  constructor(peerId?: string) {
    this.inner = newFlock(normalizePeerId(peerId));
    this.eventBatcher = new EventBatcher<Event>({
      runtime: defaultEventBatcherRuntime,
      emit: (source, events) => {
        this.deliverBatch({ source, events });
      },
    });
  }

  private static fromInner(inner: ReturnType<typeof newFlock>): Flock {
    const flock = new Flock();
    flock.inner = inner;
    return flock;
  }

  static fromJson(bundle: ExportBundle, peerId: string): Flock {
    const inner = from_json_ffi(bundle, normalizePeerId(peerId));
    return Flock.fromInner(inner as ReturnType<typeof newFlock>);
  }

  static checkConsistency(a: Flock, b: Flock): boolean {
    return Boolean(check_consistency_ffi(a.inner, b.inner));
  }

  checkInvariants(): void {
    check_invariants_ffi(this.inner);
  }

  setPeerId(peerId: string): void {
    set_peer_id(this.inner, normalizePeerId(peerId));
  }

  private putWithMetaInternal(
    key: KeyPart[],
    value: Value | undefined,
    metadata?: MetadataMap,
    now?: number,
  ): void {
    this.putWithMetaPrepared(
      cloneKey(key),
      value,
      metadata,
      normalizeMutationNow(now),
    );
  }

  private putWithMetaPrepared(
    cleanKey: KeyPart[],
    value: Value | undefined,
    metadata: MetadataMap | undefined,
    cleanNow: number | undefined,
  ): void {
    const metadataClone = cloneMetadata(metadata);
    put_with_meta_ffi(
      this.inner,
      cleanKey,
      stringifyStoredJson(value),
      metadataClone,
      cleanNow,
    );
  }

  private async putWithMetaWithHooks(
    key: KeyPart[],
    value: Value | undefined,
    options: PutWithMetaOptions,
  ): Promise<void> {
    const cleanKey = cloneKey(key);
    const cleanNow = normalizeMutationNow(options.now);
    const basePayload = createPutPayload(value, options.metadata);
    const transform = options.hooks?.transform;
    if (!transform) {
      this.putWithMetaPrepared(cleanKey, value, options.metadata, cleanNow);
      return;
    }
    const workingPayload = clonePayload(basePayload);
    const transformed = await transform(
      { key: cleanKey.slice(), now: cleanNow },
      workingPayload,
    );
    const finalPayload = mergePayload(
      basePayload,
      transformed ?? workingPayload,
    );
    const finalValue = finalPayload.data;
    if (finalValue === undefined) {
      throw new TypeError("putWithMeta requires a data value");
    }
    this.putWithMetaPrepared(
      cleanKey,
      finalValue,
      finalPayload.metadata,
      cleanNow,
    );
  }

  /**
   * Put a value into the flock. If the given entry already exists, this insert will be skipped.
   * @param key
   * @param value
   * @param now
   */
  put(key: KeyPart[], value: Value | undefined, now?: number): void {
    put_json_ffi(
      this.inner,
      cloneKey(key),
      stringifyStoredJson(value),
      normalizeMutationNow(now),
    );
  }

  putWithMeta(
    key: KeyPart[],
    value: Value | undefined,
    options?: PutWithMetaOptions,
  ): void | Promise<void> {
    const opts = options ?? {};
    if (opts.hooks?.transform) {
      return this.putWithMetaWithHooks(key, value, opts);
    }
    this.putWithMetaInternal(key, value, opts.metadata, opts.now);
  }

  set(key: KeyPart[], value: Value | undefined, now?: number): void {
    this.put(key, value, now);
  }

  /**
   * Delete a value from the flock. If the given entry does not exist, this delete will be skipped.
   * @param key
   * @param now
   */
  delete(key: KeyPart[], now?: number): void {
    delete_ffi(this.inner, cloneKey(key), normalizeMutationNow(now));
  }

  get(key: KeyPart[]): Value | undefined {
    return get_ffi(this.inner, key) as Value | undefined;
  }

  /**
   * Returns the full entry payload (data, metadata, and clock) for a key.
   *
   * Unlike `get`, this distinguishes between a missing key (`undefined`) and a
   * tombstone (returns the clock and metadata with `data` omitted). Metadata is
   * cloned and defaults to `{}` when absent.
   */
  getEntry(key: KeyPart[]): EntryInfo | undefined {
    const raw = get_entry_ffi(this.inner, key) as RawEntryInfo | undefined;
    return decodeEntryInfo(raw);
  }

  merge(other: Flock): void {
    this.eventBatcher.beforeImport();
    merge(this.inner, other.inner);
  }

  /**
   * Returns the exclusive/visible version vector.
   *
   * Only peers that currently own at least one visible entry are included.
   * This vector is consistent with current visible state and is the correct
   * baseline for incremental export/replication.
   *
   * Complexity: O(M + V log M + R (log L + log S)).
   * - M = memtablePeerCount
   * - V = vvPeerCount
   * - L = memtableLen
   * - R = scanned candidate rows in KV_BY_PEER_CLOCK
   * - S = storage key count in KV_BY_KEY
   * No full O(memtableSize) pre-scan is performed.
   *
   * Use this version when sending to other peers for incremental sync.
   */
  version(): VersionVector {
    return decodeVersionVectorFromRaw(version_ffi(this.inner));
  }

  /**
   * Returns the inclusive/max-seen version vector.
   *
   * Tracks max seen clocks per peer for this process lifetime
   * (open/import/local writes), including peers that may no longer own visible
   * entries.
   *
   * Use this version for completeness checks, not incremental export baselines.
   */
  inclusiveVersion(): VersionVector {
    return decodeVersionVectorFromRaw(inclusiveVersion_ffi(this.inner));
  }

  private exportJsonInternal(
    from?: VersionVector,
    pruneTombstonesBefore?: number,
    peerId?: string,
  ): ExportBundle {
    const pruneBefore = normalizePruneBefore(pruneTombstonesBefore);
    const normalizedPeerId =
      peerId !== undefined && isValidPeerId(peerId) ? peerId : undefined;
    return export_json_ffi(
      this.inner,
      encodeVersionVectorForFfi(from),
      pruneBefore,
      normalizedPeerId,
    ) as ExportBundle;
  }

  private async exportJsonWithHooks(
    options: ExportOptions,
  ): Promise<ExportBundle> {
    const base = this.exportJsonInternal(
      options.from,
      options.pruneTombstonesBefore,
      options.peerId,
    );
    const transform = options.hooks?.transform;
    if (!transform) {
      return base;
    }
    const result: ExportBundle = { version: base.version, entries: {} };
    for (const [key, record] of Object.entries(base.entries)) {
      const context = buildContext(key, record);
      const basePayload = createExportPayload(record);
      const workingPayload = clonePayload(basePayload);
      const transformed = await transform(context, workingPayload);
      const finalPayload = mergePayload(
        basePayload,
        transformed ?? workingPayload,
      );
      result.entries[key] = buildRecord(record.c, finalPayload);
    }
    return result;
  }

  exportJson(): ExportBundle;
  exportJson(from: VersionVector): ExportBundle;
  exportJson(from: VersionVector, pruneTombstonesBefore: number): ExportBundle;
  exportJson(options: ExportOptions): Promise<ExportBundle>;
  exportJson(
    arg?: VersionVector | ExportOptions,
    pruneTombstonesBefore?: number,
  ): ExportBundle | Promise<ExportBundle> {
    if (arg === undefined) {
      return this.exportJsonInternal(undefined, pruneTombstonesBefore);
    }
    if (isExportOptions(arg)) {
      return this.exportJsonWithHooks(arg);
    }
    return this.exportJsonInternal(arg, pruneTombstonesBefore);
  }

  private importJsonInternal(bundle: ExportBundle): ImportReport {
    this.eventBatcher.beforeImport();
    const report = import_json_ffi(this.inner, bundle) as
      | RawImportReport
      | undefined;
    return decodeImportReport(report);
  }

  private async importJsonWithHooks(
    options: ImportOptions,
  ): Promise<ImportReport> {
    const preprocess = options.hooks?.preprocess;
    const working = preprocess ? cloneBundle(options.bundle) : options.bundle;
    const skippedByHooks: Array<{ key: KeyPart[]; reason: string }> = [];
    if (preprocess) {
      for (const key of Object.keys(working.entries)) {
        const record = working.entries[key];
        if (!record) {
          continue;
        }
        const context = buildContext(key, record);
        const basePayload = createExportPayload(record);
        const decision = await preprocess(context, clonePayload(basePayload));
        const normalized = normalizeImportDecision(decision);
        if (!normalized.accept) {
          skippedByHooks.push({ key: context.key, reason: normalized.reason });
          delete working.entries[key];
          continue;
        }
        working.entries[key] = buildRecord(
          record.c,
          mergePayload(basePayload, payloadFromImportDecision(decision)),
        );
      }
    }
    const coreReport = this.importJsonInternal(working);
    return {
      accepted: coreReport.accepted,
      skipped: skippedByHooks.concat(coreReport.skipped),
    };
  }

  importJson(bundle: ExportBundle): ImportReport;
  importJson(options: ImportOptions): Promise<ImportReport>;
  importJson(
    arg: ExportBundle | ImportOptions,
  ): ImportReport | Promise<ImportReport> {
    if (isImportOptions(arg)) {
      return this.importJsonWithHooks(arg);
    }
    return this.importJsonInternal(arg);
  }

  importJsonStr(bundle: string): ImportReport {
    this.eventBatcher.beforeImport();
    const report = import_json_str_ffi(this.inner, bundle) as
      | RawImportReport
      | undefined;
    return decodeImportReport(report);
  }

  getMaxPhysicalTime(): number {
    return Number(get_max_physical_time_ffi(this.inner));
  }

  peerId(): string {
    const id = peer_id_ffi(this.inner);
    if (typeof id !== "string") {
      throw new TypeError("peerId ffi returned unexpected value");
    }
    if (!isValidPeerId(id)) {
      throw new TypeError("peerId ffi returned an invalid string");
    }
    return id;
  }

  digest(): string {
    const hex = digest_hex_ffi(this.inner);
    if (typeof hex !== "string") {
      throw new TypeError("digest ffi returned unexpected value");
    }
    return hex;
  }

  kvToJson(): ExportBundle {
    return kv_to_json_ffi(this.inner) as ExportBundle;
  }

  putMvr(key: KeyPart[], value: Value, now?: number): void {
    put_mvr_ffi(
      this.inner,
      cloneKey(key),
      stringifyJson(value),
      normalizeMutationNow(now),
    );
  }

  getMvr(key: KeyPart[]): Value[] {
    const raw = get_mvr_ffi(this.inner, cloneKey(key));
    return Array.isArray(raw) ? (raw as Value[]) : [];
  }

  scan(options: ScanOptions = {}): ScanRow[] {
    if (!options || typeof options !== "object") {
      throw new TypeError("scan options must be an object");
    }
    const start = encodeBound(options.start);
    const end = encodeBound(options.end);
    const prefix = normalizeScanPrefix(options.prefix);
    const rows = scan_ffi(this.inner, start, end, prefix) as
      | RawScanRow[]
      | undefined;
    if (!Array.isArray(rows)) {
      return [];
    }
    return rows
      .filter((row): row is RawScanRow => Boolean(row))
      .map((row) => ({
        key: Array.isArray(row.key) ? row.key : [],
        raw: row.raw,
        value: row.value,
      }));
  }

  private ensureNativeSubscription(): void {
    if (this.nativeUnsubscribe !== undefined) {
      return;
    }
    this.nativeUnsubscribe = subscribe_ffi(this.inner, (payload: unknown) => {
      const batch = decodeEventBatch(payload);
      this.handleBatch(batch);
    }) as () => void;
  }

  private handleBatch(batch: EventBatch): void {
    const bufferable = batch.source === "local";
    this.eventBatcher.handleCommitEvents(batch.source, batch.events, bufferable);
  }

  private deliverBatch(batch: EventBatch): void {
    if (this.listeners.size === 0) {
      return;
    }
    const listeners = Array.from(this.listeners);
    for (const listener of listeners) {
      try {
        listener(batch);
      } catch (error) {
        void error;
      }
    }
  }

  subscribe(listener: (batch: EventBatch) => void): () => void {
    this.listeners.add(listener);
    this.ensureNativeSubscription();

    return () => {
      this.listeners.delete(listener);
      // Optionally clean up native subscription when no listeners remain
      if (this.listeners.size === 0 && this.nativeUnsubscribe !== undefined) {
        this.nativeUnsubscribe();
        this.nativeUnsubscribe = undefined;
      }
    };
  }

  /**
   * Enable auto-debounce mode. Events will be accumulated and emitted after
   * the specified timeout of inactivity. Each new operation resets the timer.
   *
   * Use `commit()` to force immediate emission of pending events.
   * Use `disableAutoDebounceCommit()` to disable and emit pending events.
   *
   * @param timeout - Debounce timeout in milliseconds
   * @param options - Optional configuration object with maxDebounceTime (default: 10000ms)
   * @throws Error if called while a transaction is active
   * @throws Error if autoDebounceCommit is already active
   *
   * @example
   * ```ts
   * flock.autoDebounceCommit(100);
   * flock.put(["a"], 1);
   * flock.put(["b"], 2);
   * // No events emitted yet...
   * // After 100ms of inactivity, subscribers receive single EventBatch
   * // If operations keep coming, commit happens after maxDebounceTime (10s default)
   * ```
   */
  autoDebounceCommit(
    timeout: number,
    options?: { maxDebounceTime?: number },
  ): void {
    if (this.isInTxn()) {
      throw new Error(
        "Cannot enable autoDebounceCommit while transaction is active",
      );
    }
    this.eventBatcher.autoDebounceCommit(timeout, options);
  }

  /**
   * Disable auto-debounce mode and emit any pending events immediately.
   * No-op if autoDebounceCommit is not active.
   */
  disableAutoDebounceCommit(): void {
    this.eventBatcher.disableAutoDebounceCommit();
  }

  /**
   * Force immediate emission of any pending debounced events.
   * Does not disable auto-debounce mode - new operations will continue to be debounced.
   * No-op if autoDebounceCommit is not active or no events are pending.
   */
  commit(): void {
    this.eventBatcher.commit();
  }

  /**
   * Check if auto-debounce mode is currently active.
   */
  isAutoDebounceActive(): boolean {
    return this.eventBatcher.isAutoDebounceActive();
  }

  /**
   * Execute operations within a transaction. All put/delete operations inside
   * the callback will be batched and emitted as a single EventBatch when the
   * transaction commits successfully.
   *
   * If the callback throws an error, the transaction is rolled back and no
   * events are emitted. Note: Data changes are NOT rolled back - only event
   * emission is affected.
   *
   * The callback must be synchronous. For async operations, use FlockSQLite.
   *
   * @param callback - Synchronous function containing put/delete operations
   * @returns The return value of the callback
   * @throws Error if nested transaction attempted
   * @throws Error if import is called during the transaction (auto-commits first)
   * @throws Error if called while autoDebounceCommit is active
   *
   * @example
   * ```ts
   * flock.txn(() => {
   *   flock.put(["a"], 1);
   *   flock.put(["b"], 2);
   *   flock.put(["c"], 3);
   * });
   * // Subscribers receive a single EventBatch with 3 events
   * ```
   */
  txn<T>(callback: () => T): T {
    if (this.eventBatcher.isAutoDebounceActive()) {
      throw new Error(
        "Cannot start transaction while autoDebounceCommit is active",
      );
    }
    if (isAsyncCallback(callback)) {
      throw new TypeError(SYNC_TXN_CALLBACK_ERROR);
    }
    txn_begin_ffi(this.inner);
    try {
      const result = callback();
      if (isThenable(result)) {
        void Promise.resolve(result).catch(() => {});
        if (is_in_txn_ffi(this.inner)) {
          txn_rollback_ffi(this.inner);
        }
        throw new TypeError(SYNC_TXN_CALLBACK_ERROR);
      }
      txn_commit_ffi(this.inner);
      return result;
    } catch (e) {
      // Only rollback if transaction is still active.
      // import_json auto-commits the transaction before throwing,
      // so we must check before attempting rollback.
      if (is_in_txn_ffi(this.inner)) {
        txn_rollback_ffi(this.inner);
      }
      throw e;
    }
  }

  /**
   * Check if a transaction is currently active.
   */
  isInTxn(): boolean {
    return Boolean(is_in_txn_ffi(this.inner));
  }
}