import {
  compileCriteria,
  criteriaCompileRefusal,
  criteriaFields,
  type CriteriaOptions,
  filterByCriteria,
} from '../db/criteria.ts';
import { strictCriteriaRefusal } from '../db/criteriaDialect.ts';
import { byRecordId, rowComparator } from '../db/ordering.ts';
import { isDeclared, queryCriteriaTypes, queryFieldNames, querySchema } from '../db/schema.ts';
import type { Store } from '../db/store.ts';
import { bigDecimalFields, encodeRow } from '../db/values.ts';
import type { Connection, Subscription, ViewRow } from '../protocol/connection.ts';
import { isLegacyFidelity } from '../protocol/fidelity.ts';
import { MESSAGE_TYPE, ERROR_CODE } from '../protocol/messageTypes.ts';
import {
  nonExistentSubscriptionNack,
  resourceNotMappedNack,
  subscriptionAlreadyExistsNack,
} from '../protocol/msgNack.ts';
import { dataserverRow, rowRefFor, stampRow } from '../protocol/rowUpdate.ts';

import type { GenesisMessage, IndexDef, MockServerConfig, QueryDef, Row } from '../types.ts';
import { permittedQueryRow, requestingUser } from './resourceAuth.ts';

// The real dataserver's defaults (genesis-pal-dataserver ClientOptions,
// GSF 8.15.x) — also what foundation-comms' DatasourceDefaults send.
const DEFAULT_MAX_ROWS = 250;
const DEFAULT_MAX_VIEW = 1000;

function readCount(value: unknown, fallback: number): number {
  const count = typeof value === 'string' && value.trim() !== '' ? Number(value) : value;
  return typeof count === 'number' && Number.isInteger(count) && count >= 0 ? count : fallback;
}

// MAX_ROWS is the page size, MAX_VIEW the most rows the client's view may
// hold across all pages. 0 means "no limit" for either, and MAX_ROWS is
// clamped to MAX_VIEW. MOVING_VIEW defaults to true. A logon with a
// VIEW_NUMBER other than 0 asks for numbered pages: its view is one page,
// with no moving view. All of it as ClientOptions.parse does (v8.15.29).
export function pagingOptions(details: Row): {
  maxRows: number;
  maxView: number;
  movingView: boolean;
} {
  const pageNumbered = readCount(details.VIEW_NUMBER, 0) !== 0;
  const requestedRows = readCount(details.MAX_ROWS, DEFAULT_MAX_ROWS);
  const maxView =
    (pageNumbered ? requestedRows : readCount(details.MAX_VIEW, DEFAULT_MAX_VIEW)) || Infinity;
  const maxRows = requestedRows || Infinity;
  const movingView = !pageNumbered && readFlag(details.MOVING_VIEW, true);
  return { maxRows: Math.min(maxRows, maxView), maxView, movingView };
}

// The key a row is tracked under in Subscription.pending / held — undefined
// for a row without a ROW_REF, which no live change could ever be matched
// against (and which the client can't identify either).
export function pendingKey(rowRef: unknown): string | undefined {
  return rowRef === undefined || rowRef === null ? undefined : String(rowRef);
}

// A row of the view, built from its stored (or view) row. `row` is its wire
// form: stamped as an INSERT while it waits, plain once held.
export function viewRow(row: Row, rowRef: unknown, source: Row): ViewRow {
  return { row, rowRef, source, timestamp: source.TIMESTAMP ?? source.RECORD_ID };
}

// What the client holds once a waiting row is sent: its values, no DETAILS.
function heldRow(waiting: ViewRow): ViewRow {
  const { DETAILS: _details, ...row } = waiting.row;
  return { ...waiting, row };
}

function buildPending(rows: ViewRow[]): Map<string, ViewRow> {
  const pending = new Map<string, ViewRow>();
  rows.forEach((entry, index) => {
    const key = pendingKey(entry.rowRef);
    // A row with no ROW_REF (or a duplicate one) still has to be delivered;
    // park it under a key no ROW_REF can produce rather than dropping it.
    // Live changes can't reach it there, which is as much as the client
    // could make of them: to it, rows sharing a ROW_REF are one row.
    pending.set(key === undefined || pending.has(key) ? `\u0000${index}` : key, entry);
  });
  return pending;
}

// A matched row the client doesn't hold and isn't sent now. It waits for a
// page when it sorts after the last row a page delivered, since the real
// NextRowsStrategy reads on from there; one that sorts before it is never
// read again. Pending stays in delivery order.
export function waitForPage(subscription: Subscription, key: string, entry: ViewRow): void {
  const { pending, order, cursor } = subscription;
  if (!pending) return;
  pending.delete(key);
  if (!order) {
    pending.set(key, entry);
    return;
  }
  if (cursor && order(entry.source, cursor) <= 0) return;
  const waiting = [...pending];
  const at = waiting.findIndex(([, other]) => order(entry.source, other.source) < 0);
  waiting.splice(at === -1 ? waiting.length : at, 0, [key, entry]);
  pending.clear();
  for (const [waitingKey, other] of waiting) pending.set(waitingKey, other);
}

// The next page: up to MAX_ROWS waiting rows, but never more than the view
// has room for. The real NextRowsStrategy reads while the client index is
// below MAX_VIEW, so a full view gets an empty page, and a slot freed by a
// DELETE is filled by the next row waiting. Delivered rows move to `held`,
// and the last one is the new read position.
function takePage(subscription: Subscription): Row[] {
  const { pending, held } = subscription;
  if (!pending) return [];
  const room = Math.max(
    0,
    Math.min(
      subscription.maxRows ?? Infinity,
      (subscription.maxView ?? Infinity) - (held?.size ?? 0),
    ),
  );
  const page: Row[] = [];
  for (const [key, entry] of pending) {
    if (page.length >= room) break;
    page.push(entry.row);
    pending.delete(key);
    held?.set(key, heldRow(entry));
    subscription.cursor = entry.source;
  }
  return page;
}

// The dataserver's own refusal of a logon (genesis-pal-dataserver): a
// LOGON_NACK with a StandardError, sent instead of LOGON_ACK. No subscription
// is opened, so a later DATA_LOGOFF on it gets the router's 404.
function logonNack(
  code: 'INVALID_CRITERIA' | 'INVALID_INDEX' | 'INVALID_PARAMETER',
  text: string,
): Row {
  return {
    MESSAGE_TYPE: MESSAGE_TYPE.LOGON_NACK,
    ERROR: [{ '@type': 'StandardError', CODE: code, TEXT: text, STATUS_CODE: '400 Bad Request' }],
    WARNING: [],
  };
}

// DETAILS.FIELDS: a space-separated projection of the query's columns, in the
// query's order. Names it doesn't expose are dropped, as the real dataserver
// drops them.
function requestedFieldNames(
  details: Row,
  query: QueryDef,
  exposed: string[] | undefined,
  config: MockServerConfig,
  store: Store,
): string[] | undefined {
  const requested =
    typeof details.FIELDS === 'string' && details.FIELDS.trim() !== ''
      ? details.FIELDS.trim().split(/\s+/)
      : undefined;
  if (!requested) return exposed;
  const available = exposed ?? querySchema(query, config, store).fields.map((field) => field.name);
  return available.filter((name) => requested.includes(name));
}

// Ascending by the index's fields, nulls first. Ties are broken by RECORD_ID,
// as the real index breaks them, so REVERSE, which flips the whole order,
// puts the newest of equal rows first.
function indexComparator(index: IndexDef): (a: Row, b: Row) => number {
  return rowComparator(index.fields.map((field) => ({ field, descending: false })));
}

// The order a snapshot and its MORE_ROWS pages come in. With ORDER_BY, the
// index ascending, and REVERSE flips it. Without one the dataserver reads its
// primary index, RECORD_ID, newest first, and REVERSE alone gives oldest
// first — the order the rows were written in (PrimaryKeyDispenser.kt
// buildPrimaryRange: `if (reverse) Order.ASC else Order.DESC`, GSF v8.15.29).
function snapshotOrder(index: IndexDef | undefined, reverse: boolean): (a: Row, b: Row) => number {
  const ascending = index ? indexComparator(index) : byRecordId;
  const flipped = index ? reverse : !reverse;
  return flipped ? (a, b) => ascending(b, a) : ascending;
}

// Without ORDER_BY, the index a criteria makes the dataserver read through:
// the query index whose leading fields the criteria uses most (the first such
// one on a tie), or none (CriteriaFilterBuilder.findBestMatchingIndex with no
// default index; QueryClientFactory.build adopts it as the client's index,
// GSF v8.15.29). The rows then come in that index's order, as under ORDER_BY.
function criteriaIndex(query: QueryDef, fields: string[]): IndexDef | undefined {
  let best: IndexDef | undefined;
  let bestCount = 0;
  for (const index of query.indexes ?? []) {
    const unmatched = index.fields.findIndex((field) => !fields.includes(field));
    const count = unmatched < 0 ? index.fields.length : unmatched;
    if (count > bestCount) [best, bestCount] = [index, count];
  }
  return best;
}

// A flag as GenesisSet.getBoolean reads it: true, or anything whose text is
// "true" in any case (obj.toString().toBoolean()).
function readBoolean(value: unknown): boolean {
  return value === true || (value !== undefined && value !== null && /^true$/i.test(String(value)));
}

// A flag with a default, as GenesisSet.getBoolean(key, default) reads one:
// the default when it is absent, otherwise as readBoolean.
function readFlag(value: unknown, fallback: boolean): boolean {
  return value === undefined || value === null ? fallback : readBoolean(value);
}

// How a query's rows go on the wire: each BIGDECIMAL column as a string with
// its scale, as GSF's serializer writes a BigDecimal (db/values.ts).
// undefined when the query has none — known without resolving the schema
// for an undeclared source with no BIGDECIMAL fieldTypes or derived types.
export function rowEncoder(
  query: QueryDef,
  config: MockServerConfig,
  store: Store,
): ((row: Row) => Row) | undefined {
  const view = store.views.get(query.source) ?? config.views?.[query.source];
  const overridden = [
    ...Object.values(query.fieldTypes ?? {}),
    ...Object.values(view?.derivedTypes ?? {}).map((entry) =>
      typeof entry === 'string' ? entry : entry.type,
    ),
  ];
  if (!isDeclared(query.source, config, store) && !overridden.includes('BIGDECIMAL')) {
    return undefined;
  }
  const fields = bigDecimalFields(querySchema(query, config, store).fields);
  return fields && ((row: Row) => encodeRow(row, fields));
}

// DATA_LOGON opens a persistent subscription keyed by SOURCE_REF: reply with
// LOGON_ACK and the first page of the snapshot, then keep the SOURCE_REF
// registered so MORE_ROWS can page the rest and later inserts/amends/deletes
// on this resource are pushed on the same channel (see server.ts broadcast())
// until the client sends DATA_LOGOFF.
export function handleDataLogon(
  message: GenesisMessage,
  connection: Connection,
  config: MockServerConfig,
  store: Store,
): void {
  const legacy = isLegacyFidelity(config);
  const details = message.DETAILS ?? {};
  const resourceName = details.DATASOURCE_NAME;
  const query = config.queries?.[resourceName];

  if (!query) {
    connection.reply(
      message,
      legacy
        ? {
            MESSAGE_TYPE: MESSAGE_TYPE.QUERY_UPDATE,
            ERROR: [
              { CODE: ERROR_CODE.UNKNOWN_RESOURCE, TEXT: `Unknown datasource: ${resourceName}` },
            ],
          }
        : resourceNotMappedNack(resourceName),
    );
    return;
  }

  const sourceRef = message.SOURCE_REF!;
  // The router refuses a second logon on a live SOURCE_REF and leaves the
  // first subscription running. (Legacy mode silently replaced it.)
  if (!legacy && connection.subscriptions.has(sourceRef)) {
    connection.reply(message, subscriptionAlreadyExistsNack());
    return;
  }

  const criteria = details.CRITERIA_MATCH;
  const user = requestingUser(connection);
  // The query's fields block (or the source's declared columns) is all a
  // criteria may name and all a row carries; DETAILS.FIELDS narrows the rows.
  // Legacy subscriptions ignore all of it, and ORDER_BY, as 15.47 did.
  const exposed = queryFieldNames(query, config, store);
  // Legacy subscriptions keep the store's insertion order.
  let compare: ((a: Row, b: Row) => number) | undefined;
  // How the criteria reads the rows: by the fields' declared types, as GSF
  // compiles it against its entity; legacy as up to 15.47.
  const criteriaOptions: CriteriaOptions = legacy
    ? { lenient: true }
    : { types: queryCriteriaTypes(query, config, store) };
  const paging = pagingOptions(details);
  if (!legacy) {
    // A moving view needs a MAX_VIEW to move in, and MOVING_VIEW defaults to
    // true, so MAX_VIEW: 0 alone is refused. Checked before ORDER_BY
    // (ConnectionManager.verifyClientOptions, v8.15.29).
    if (paging.movingView && paging.maxView === Infinity) {
      connection.reply(
        message,
        logonNack('INVALID_PARAMETER', "MOVING_VIEW can't be true if maxView is set to 0"),
      );
      return;
    }
    // ORDER_BY names one of the query's own indexes (QueryDef.indexes). GSF
    // checks it before it looks at the criteria
    // (ConnectionManager.verifyClientOptions, v8.15.29).
    let index: IndexDef | undefined;
    if (details.ORDER_BY) {
      index = query.indexes?.find((candidate) => candidate.name === details.ORDER_BY);
      if (!index) {
        connection.reply(
          message,
          logonNack(
            'INVALID_INDEX',
            `No such index ${details.ORDER_BY} for dataserver query ${resourceName}`,
          ),
        );
        return;
      }
    }
    // GSF parses the criteria as Groovy first, validates it and compiles it,
    // refusing one that fails with the message (CriteriaFilterBuilder
    // getCriteriaFilter, v8.15.29): no subscription opens. Refused here: the
    // syntax errors Groovy would report too, and this engine's own dialect —
    // AND / OR / NOT, `<>`, a single `=`, dotted names, `$` in double quotes,
    // calls off GSF's whitelists (db/criteriaDialect.ts), operands side by
    // side (db/criteriaJuxtaposition.ts). A criteria outside
    // this engine's grammar that
    // may well be valid Groovy still fails open below.
    const refusal =
      typeof criteria === 'string'
        ? strictCriteriaRefusal(criteria, exposed && new Set(exposed), criteriaOptions.types)
        : undefined;
    if (refusal) {
      connection.reply(message, logonNack('INVALID_CRITERIA', refusal));
      return;
    }
    const parsedFields = criteriaFields(criteria);
    // The real dataserver lets a criteria read only the query's own fields
    // (QueryClientFactory.kt). Unknown columns are only checked when the
    // query's fields are known; an inferred query fails open as before.
    const outside = exposed && parsedFields?.find((field) => !exposed.includes(field));
    if (outside) {
      connection.reply(
        message,
        logonNack(
          'INVALID_CRITERIA',
          `GenericCriteriaValidationException : Criteria validation failed: Variable ${outside} is not allowed in criteria expression`,
        ),
      );
      return;
    }
    // Then GSF compiles it, refusing a call no overload of the function
    // takes (db/criteria.ts criteriaCompileRefusal).
    const uncompiled =
      typeof criteria === 'string'
        ? criteriaCompileRefusal(criteria, criteriaOptions.types)
        : undefined;
    if (uncompiled) {
      connection.reply(message, logonNack('INVALID_CRITERIA', uncompiled));
      return;
    }
    compare = snapshotOrder(
      index ?? criteriaIndex(query, parsedFields ?? []),
      readBoolean(details.REVERSE),
    );
  }
  // Order matters: the query's own filter (the where-clause equivalent) and
  // the user's row permissioning narrow the set BEFORE the client's
  // CRITERIA_MATCH; hidden fields are stripped last, off the final rows.
  // Paging applies after all of them.
  let rows = store.getResourceRows(query.source);
  if (query.filter) rows = rows.filter(query.filter);
  // ROWS_COUNT is the real dataserver's index count: it ignores
  // CRITERIA_MATCH and the user's rowFilter (recorded: ROWS_COUNT 10 with 6
  // rows matching), which grid-pro and the AI assistant already allow for.
  // The query's own filter is the where clause the index is built over, so
  // the count is taken after it — an assumption, as no recording covers it.
  const rowsCount = rows.length;
  rows = filterByCriteria(rows, criteria, criteriaOptions);
  if (compare) rows = [...rows].sort(compare);
  // Permissioning is applied per row so the ROW_REF can still be computed from
  // the UNSTRIPPED row — pk fields may themselves be hidden from this user,
  // and a snapshot row with no ROW_REF can never be matched to the later
  // INSERT/MODIFY/DELETE pushes (which stamp the real ref, see server.ts).
  // RECORD_ID (or, in legacy mode, the pk fields) is never sent unless the
  // query's fields block names it.
  const getRowRef = (row: Row) => rowRefFor(config, store, query.source, row);

  const fieldNames = legacy
    ? undefined
    : requestedFieldNames(details, query, exposed, config, store);
  // A BigDecimal goes out as a string with its column's scale (db/values.ts).
  const encode = legacy ? undefined : rowEncoder(query, config, store);
  const snapshotRows: ViewRow[] = [];
  for (const row of rows) {
    const permitted = permittedQueryRow(row, query.auth, user, fieldNames);
    if (!permitted) continue;
    const rowRef = getRowRef(row);
    const sent = dataserverRow(permitted, row, query.fields);
    snapshotRows.push(
      viewRow(stampRow(encode ? encode(sent) : sent, rowRef, 'INSERT', legacy), rowRef, row),
    );
  }

  const subscription: Subscription = {
    resourceName,
    getRowRef,
    predicate: compileCriteria(criteria, criteriaOptions),
    sequenceId: 1,
    ...(encode ? { encode } : {}),
  };

  if (legacy) {
    connection.subscribe(sourceRef, subscription);
    connection.reply(message, {
      MESSAGE_TYPE: MESSAGE_TYPE.QUERY_UPDATE,
      ROW: snapshotRows.map((entry) => entry.row),
      ROWS_COUNT: snapshotRows.length,
      SEQUENCE_ID: 1,
      MORE_ROWS: false,
    });
    return;
  }

  // Every matched row starts out waiting; the first page moves up to
  // MAX_ROWS of them into `held`. Rows the view has no room for (past
  // MAX_VIEW) keep waiting, so MORE_ROWS stays true — as on the real server,
  // which then answers every MORE_ROWS with an empty {MORE_ROWS: true} page.
  // A live INSERT is pushed straight away whatever the order, even when it
  // sorts among the rows still waiting, and a later page doesn't repeat it:
  // the real ClientIndex.processInsertUpdate sends it as soon as it arrives
  // (a full view takes it only when it moves), and NextRowsStrategy skips
  // rows the client index already holds (GSF v8.15.29). See server.ts
  // viewUpdate.
  const tracked: Subscription = {
    ...subscription,
    pending: buildPending(snapshotRows),
    held: new Map<string, ViewRow>(),
    ...paging,
    ...(compare ? { order: compare } : {}),
    fieldNames,
  };
  const firstPage = takePage(tracked);
  const pending = tracked.pending!;
  connection.subscribe(sourceRef, tracked);

  connection.reply(message, { MESSAGE_TYPE: MESSAGE_TYPE.LOGON_ACK, SEQUENCE_ID: 0 });
  // No matching rows: the real dataserver sends neither ROW nor ROWS_COUNT.
  connection.reply(
    message,
    firstPage.length === 0
      ? { MESSAGE_TYPE: MESSAGE_TYPE.QUERY_UPDATE, MORE_ROWS: false, SEQUENCE_ID: 1 }
      : {
          MESSAGE_TYPE: MESSAGE_TYPE.QUERY_UPDATE,
          ROW: firstPage,
          MORE_ROWS: pending.size > 0,
          // Only on this first page (see rowsCount above).
          ROWS_COUNT: rowsCount,
          SEQUENCE_ID: 1,
        },
  );
}

// MORE_ROWS names its subscription in DETAILS.SOURCE_REF (foundation-comms'
// createMoreRowsMessage gives the request a SOURCE_REF of its own) or, when
// that is absent, in its own SOURCE_REF. The page goes out as a QUERY_UPDATE
// on the subscription's SOURCE_REF without ROWS_COUNT; MORE_ROWS_ACK answers
// the request's own SOURCE_REF, and only when DETAILS.SOURCE_REF names another
// one. The real router rewrites DETAILS.SOURCE_REF to its own channel-prefixed
// form only then, so that is the only ack ConnectionManager.processMoreRows
// sends that it can route back to the socket (MsgType.MoreRowsOrColumns
// @v8.15.29); on the subscription's own SOURCE_REF the ack is lost. The ack goes first
// because the real server writes it while its dataserver is still reading the
// page. Subscriptions are per connection, so another socket can never page
// this one: its lookup misses and it gets the 404.
export function handleMoreRows(
  message: GenesisMessage,
  connection: Connection,
  config: MockServerConfig,
): void {
  if (isLegacyFidelity(config)) {
    // Up to 15.47 MORE_ROWS was unrouted: logged, never answered.
    console.warn(`[mock-server] Unhandled MESSAGE_TYPE: ${message.MESSAGE_TYPE}`);
    return;
  }
  const detailsSourceRef: string | undefined = message.DETAILS?.SOURCE_REF;
  const streamSourceRef = detailsSourceRef ?? message.SOURCE_REF;
  const subscription =
    streamSourceRef === undefined ? undefined : connection.subscriptions.get(streamSourceRef);
  if (!subscription) {
    connection.reply(message, nonExistentSubscriptionNack());
    return;
  }
  if (detailsSourceRef !== undefined && detailsSourceRef !== message.SOURCE_REF) {
    connection.reply(message, { MESSAGE_TYPE: MESSAGE_TYPE.MORE_ROWS_ACK });
  }

  const page = takePage(subscription);
  subscription.sequenceId += 1;
  // No rows to send still gets an answer — an empty QUERY_UPDATE, as the real
  // QueryClient sends when a read finds nothing: MORE_ROWS: false at the end
  // of the data, true while rows are waiting for room in a full view.
  connection.send({
    MESSAGE_TYPE: MESSAGE_TYPE.QUERY_UPDATE,
    SOURCE_REF: streamSourceRef,
    ...(page.length > 0 ? { ROW: page } : {}),
    MORE_ROWS: (subscription.pending?.size ?? 0) > 0,
    SEQUENCE_ID: subscription.sequenceId,
  });
}

export function handleDataLogoff(
  message: GenesisMessage,
  connection: Connection,
  config: MockServerConfig,
): void {
  const sourceRef = message.SOURCE_REF;
  const subscribed = sourceRef !== undefined && connection.subscriptions.has(sourceRef);
  if (subscribed) connection.unsubscribe(sourceRef);
  // Legacy mode never answered a logoff.
  if (isLegacyFidelity(config)) return;
  connection.reply(
    message,
    subscribed ? { MESSAGE_TYPE: MESSAGE_TYPE.LOGOFF_ACK } : nonExistentSubscriptionNack(),
  );
}