// REQ_<NAME> -> REP_<NAME>, as GSF 8.15.29's request server answers it
// (genesis-pal-requestserver). A request server is one of two kinds, and they
// read a request differently:
//
// - criteria-only (`criteriaOnly`, criteriaOnlyRequest = true:
//   EntityDbRequestReply.kt): REQUEST must be present but its fields are
//   ignored; CRITERIA_MATCH, ORDER_BY, OFFSET / VIEW_NUMBER and MAX_ROWS run
//   as one database read. Every reply carries MORE_ROWS, and NEXT_OFFSET while
//   there are more (the SQL flavour, as the recordings show).
// - plain (RequestReply.kt): REQUEST's fields select rows — exact values,
//   `*` wildcards (WildcardFilter.kt), FIELD_FROM/FIELD_TO ranges — and an
//   array REQUEST runs each object and concatenates the results. MAX_ROWS
//   caps each one; OFFSET and ORDER_BY are ignored, and nothing says whether
//   there are more rows.
//
// See the README's "Request servers" for every rule and its evidence.

import {
  criteriaCompileRefusal,
  CriteriaSyntaxError,
  filterByCriteria,
  parseCriteria,
  parseGenesisDateTime,
  UnsupportedCriteriaError,
  type ParsedCriteria,
  type Predicate,
} from '../db/criteria.ts';
import { strictCriteriaRefusal } from '../db/criteriaDialect.ts';
import { checkGroovySyntax } from '../db/groovySyntax.ts';
import { withoutRecordFields } from '../db/identity.ts';
import { compareValues, parseOrderBy, rowComparator, type OrderedField } from '../db/ordering.ts';
import {
  pkFieldsOf,
  primaryKeyIndexName,
  projectRow,
  requestReplyFieldNames,
  requestReplySourceSchema,
  typedFields,
  type ResolvedField,
} from '../db/schema.ts';
import type { Store } from '../db/store.ts';
import { bigDecimalFields, decimalText, encodeRow } from '../db/values.ts';
import type { Connection } from '../protocol/connection.ts';
import { ErrorCode, NackError, nackPayload } from '../protocol/errors.ts';
import { isLegacyFidelity } from '../protocol/fidelity.ts';
import { daoClassOf, viewEntityClassOf } from '../protocol/fieldTypes.ts';
import { ERROR_CODE } from '../protocol/messageTypes.ts';
import { resourceNotMappedNack, serviceMsgNack } from '../protocol/msgNack.ts';
import type {
  GenesisMessage,
  MockServerConfig,
  RequestingUser,
  RequestReplyDef,
  Row,
} from '../types.ts';
import { applyResourceAuth, permittedQueryRow, requestingUser } from './resourceAuth.ts';

// RequestReply.DEFAULT_MAXIMUM_ROWS: the rows a REP_ carries when neither the
// request (DETAILS.MAX_ROWS) nor the server (rowLimit) says otherwise.
const DEFAULT_ROW_LIMIT = 10000;
// Request.DEFAULT_VIEW_NUMBER: 0 is "not paging by view".
const DEFAULT_VIEW_NUMBER = 0;
const WILDCARD = '*';

// GSF answers an IllegalArgumentException from the request server (a failed
// require(), a bad ORDER_BY) with StandardError REQUEST_FAILED and its message.
const requestFailed = (text: string) => NackError.standard(ErrorCode.REQUEST_FAILED, text);
// Anything else a criteria-only server hits — a REQUEST that doesn't
// deserialize, a criteria Groovy can't compile — is INTERNAL_ERROR with the
// code's own text ("Unexpected server error"), the message only in the log.
const internalError = () => NackError.standard(ErrorCode.INTERNAL_ERROR);

const isPlainObject = (value: unknown): value is Row =>
  typeof value === 'object' && value !== null && !Array.isArray(value);

// --- request details ---------------------------------------------------------

type Strictness = 'strict' | 'lenient';

// An integer DETAILS value. Jackson coerces numeric strings and truncates
// fractions; anything else fails to deserialize the request (strict: a
// criteria-only server's Request<Unit>) or reads as absent (lenient: a plain
// server's GenesisSet.getInteger).
function readInteger(value: unknown, strictness: Strictness): number | undefined {
  if (value === undefined || value === null) return undefined;
  const number = typeof value === 'string' && value.trim() !== '' ? Number(value) : value;
  if (typeof number === 'number' && Number.isFinite(number)) return Math.trunc(number);
  if (strictness === 'strict') throw internalError();
  return undefined;
}

const readText = (value: unknown): string => (typeof value === 'string' ? value : '');

// --- the source --------------------------------------------------------------

interface Source {
  name: string;
  // Every field the entity has (its columns, RECORD_ID and TIMESTAMP), by name.
  fields: Map<string, ResolvedField>;
  pkFields: string[];
  // Each field's Genesis type when the source declares its fields: what GSF
  // compiles a criteria against (db/criteria.ts).
  types?: Map<string, string>;
}

function sourceOf(requestReply: RequestReplyDef, config: MockServerConfig, store: Store): Source {
  const name = requestReply.source!;
  const schema = requestReplySourceSchema(requestReply, config, store);
  return {
    name,
    fields: new Map(schema.fields.map((field) => [field.name, field])),
    pkFields: store.getPkFields(name),
    ...(schema.declared
      ? { types: new Map(schema.fields.map((field) => [field.name, field.type])) }
      : {}),
  };
}

function isView(name: string, config: MockServerConfig, store: Store): boolean {
  return store.views.has(name) || !!config.views?.[name];
}

// The generated entity class the rows are (definition.source.entityType):
// the DAO for a table, the view entity for a view.
function parametricTypeOf(name: string, config: MockServerConfig, store: Store): string {
  return isView(name, config, store) ? viewEntityClassOf(name) : daoClassOf(name);
}

const byPrimaryKey = (source: Source): OrderedField[] =>
  source.pkFields.map((field) => ({ field, descending: false }));

// The REP_ rows as they go on the wire: each BigDecimal a string with its
// column's scale, as GSF's serializer writes it (db/values.ts).
function wireRows(rows: Row[], source: Source): Row[] {
  const fields = bigDecimalFields([...source.fields.values()]);
  return fields ? rows.map((row) => encodeRow(row, fields)) : rows;
}

// --- criteria ----------------------------------------------------------------

type CriteriaFilter = { kind: 'all' } | { kind: 'none' } | { kind: 'match'; predicate: Predicate };

// CRITERIA_MATCH against the entity. A malformed one fails as `onSyntaxError`
// says (the two server kinds differ): one Groovy refuses (db/groovySyntax.ts),
// or one this grammar can't parse that the Groovy check can't vouch for
// either. A criteria naming a field the entity doesn't have matches nothing:
// GSF evaluates it outside the database, where the unknown variable never
// compares true (recorded: REPLY: []). Valid Groovy this grammar doesn't model
// (a bare boolean field, `in [...]`, a function it doesn't know) fails open
// (logged), as everywhere else in the engine.
function criteriaFilter(
  criteria: string,
  source: Source,
  onSyntaxError: (error: CriteriaSyntaxError) => Error,
): CriteriaFilter {
  if (criteria.trim() === '') return { kind: 'all' };
  // What GSF can't compile: a syntax error Groovy reports, this engine's
  // dialect (AND, OR, NOT, `<>`, `=`, a call off GSF's whitelist, ...), or a
  // call no overload takes, with GSF's text.
  const refusal =
    strictCriteriaRefusal(criteria, new Set(source.fields.keys()), source.types) ??
    criteriaCompileRefusal(criteria, source.types);
  if (refusal) throw onSyntaxError(new CriteriaSyntaxError(refusal));
  const groovy = checkGroovySyntax(criteria);
  let parsed: ParsedCriteria;
  try {
    parsed = parseCriteria(criteria, { types: source.types });
  } catch (error) {
    if (error instanceof CriteriaSyntaxError && groovy.kind !== 'valid') {
      throw onSyntaxError(error);
    }
    if (error instanceof CriteriaSyntaxError || error instanceof UnsupportedCriteriaError) {
      console.warn(
        `[mock-server] Unsupported CRITERIA_MATCH syntax, ignoring filter: ${criteria}`,
        error,
      );
      return { kind: 'all' };
    }
    throw error;
  }
  if (parsed.fields.some((field) => !source.fields.has(field))) return { kind: 'none' };
  return { kind: 'match', predicate: parsed.predicate };
}

function applyCriteria(rows: Row[], filter: CriteriaFilter): Row[] {
  if (filter.kind === 'all') return rows;
  if (filter.kind === 'none') return [];
  return rows.filter((row) => {
    try {
      return filter.predicate(row);
    } catch {
      return false;
    }
  });
}

// --- ORDER_BY ----------------------------------------------------------------

// The index names ORDER_BY may use instead of fields (OrderSpec.kt
// replaceIndexWithFields): 'PK', the primary key's index and the table's
// declared indexes — a view's, its base table's.
function orderIndexes(
  source: Source,
  config: MockServerConfig,
  store: Store,
): Map<string, string[]> {
  const indexes = new Map<string, string[]>([['PK', source.pkFields]]);
  const tableName = isView(source.name, config, store)
    ? (store.views.get(source.name) ?? config.views?.[source.name])?.base
    : source.name;
  const table = tableName ? config.tables?.[tableName] : undefined;
  if (table && tableName) {
    indexes.set(primaryKeyIndexName(tableName, table), pkFieldsOf(table));
    for (const index of table.indexes ?? []) indexes.set(index.name, index.fields);
  }
  return indexes;
}

// DETAILS.ORDER_BY on a criteria-only server: fields, each optionally ASC or
// DESC ('INT_FIELD DESC', 'A, B DESC'). An index name (or 'PK') that isn't
// also a field stands for its fields. A malformed string, or a field the
// entity doesn't have, is REQUEST_FAILED with GSF's text
// (GetByCriteriaOperation.kt prints the Kotlin list of OrderedFields).
function orderingOf(
  orderBy: string,
  source: Source,
  config: MockServerConfig,
  store: Store,
): OrderedField[] | undefined {
  if (orderBy.trim() === '') return undefined;
  let fields: OrderedField[];
  try {
    fields = parseOrderBy(orderBy);
  } catch (error) {
    throw requestFailed((error as Error).message);
  }
  const indexes = orderIndexes(source, config, store);
  const namesIndex = ({ field }: OrderedField) => !source.fields.has(field) && indexes.has(field);
  if (fields.some(namesIndex)) {
    fields = fields.flatMap((ordered) =>
      namesIndex(ordered)
        ? indexes.get(ordered.field)!.map((field) => ({ field, descending: ordered.descending }))
        : [ordered],
    );
  }
  const missing = fields.filter(({ field }) => !source.fields.has(field));
  if (missing.length > 0) {
    const listed = missing
      .map(({ field, descending }) => `${descending ? 'Desc' : 'Asc'}(field=${field})`)
      .join(', ');
    throw requestFailed(`Order fields [${listed}] not found on table ${source.name}`);
  }
  return fields;
}

// --- criteria-only servers (EntityDbRequestReply.kt) -------------------------

interface Page {
  rows: Row[];
  nextOffset?: number;
  moreRows?: boolean;
}

// EntityDbRequestReply.data(): one row past MAX_ROWS says there are more.
// NEXT_OFFSET is the request's OFFSET plus how far the read got (offsetShift),
// or that alone when no OFFSET was sent. A view-numbered page can't tell when
// it was read through a filter (isNextTokenKnowable false): no MORE_ROWS then.
function pageOf(
  rows: Row[],
  maxRows: number,
  offset: number | undefined,
  offsetShift: number,
  nextKnowable: boolean,
): Page {
  if (rows.length === 0) return { rows: [], moreRows: false };
  if (!nextKnowable) return { rows };
  if (rows.length > maxRows) {
    return {
      rows: rows.slice(0, maxRows),
      nextOffset: offset === undefined ? offsetShift : offset + offsetShift,
      moreRows: true,
    };
  }
  return { rows, moreRows: false };
}

function criteriaOnlyReply(
  resourceName: string,
  requestReply: RequestReplyDef,
  message: GenesisMessage,
  config: MockServerConfig,
  store: Store,
  user: RequestingUser,
): Row {
  // The request deserializes as Request<Unit>: without a REQUEST object
  // (missing, null, nested in DETAILS) Jackson throws and the reply is
  // INTERNAL_ERROR (recorded: KotlinInvalidNullException in the server log).
  if (!isPlainObject(message.REQUEST)) throw internalError();
  const details: Row = isPlainObject(message.DETAILS) ? message.DETAILS : {};
  const offset = readInteger(details.OFFSET, 'strict');
  const viewNumber = readInteger(details.VIEW_NUMBER, 'strict') ?? DEFAULT_VIEW_NUMBER;
  const maxRows =
    readInteger(details.MAX_ROWS, 'strict') ?? requestReply.rowLimit ?? DEFAULT_ROW_LIMIT;

  // EntityDbRequestReply.process's require()s, in order.
  if (offset !== undefined && viewNumber !== DEFAULT_VIEW_NUMBER) {
    throw requestFailed(
      `You must not supply both offset and viewNumber != ${DEFAULT_VIEW_NUMBER}, ` +
        'unable to combine offset and page request',
    );
  }
  if (offset !== undefined && offset < 0) {
    throw requestFailed(
      'When providing an offset, you must provide an offset greater than or equal to 0',
    );
  }
  if (viewNumber < 0) {
    throw requestFailed(
      'When providing a view number, You must provide a view number greater than or equal to 0',
    );
  }
  if (maxRows <= 0) throw requestFailed('maxRows should be greater than 0');

  const source = sourceOf(requestReply, config, store);
  const criteria = criteriaFilter(readText(details.CRITERIA_MATCH), source, internalError);
  const ordering = orderingOf(readText(details.ORDER_BY), source, config, store);
  const byView = viewNumber !== DEFAULT_VIEW_NUMBER;
  // A read that skips rows (an OFFSET past 0, or a view number) is a paged
  // read: GetByCriteriaOperation orders it by the primary key when ORDER_BY
  // doesn't say. An unpaged read comes in the table's own order, oldest
  // first (recorded).
  const paged = byView || (offset ?? 0) > 0;
  if (criteria.kind === 'none' && paged) {
    // Criteria evaluated outside the database can't be paged
    // (PagingSpec.pagingNotSupported).
    throw requestFailed('Paging not supported on lazy criteria evaluation');
  }

  let rows = applyCriteria(store.getResourceRows(source.name), criteria);
  const order = ordering ?? (paged ? byPrimaryKey(source) : undefined);
  if (order) rows = [...rows].sort(rowComparator(order));

  const fieldNames = requestReplyFieldNames(requestReply, config, store);
  let page: Page;
  if (!requestReply.filter && !requestReply.auth) {
    // Eager: the database does the paging, reading one row past the page.
    const start = byView ? maxRows * (viewNumber - 1) : (offset ?? 0);
    const read = rows.slice(start, start + maxRows + 1);
    page = pageOf(read, maxRows, offset, maxRows, true);
    page.rows = page.rows.map((row) => projectRow(row, fieldNames));
  } else {
    // Lazy: the filter and permissions run on the rows the database returns,
    // so the read goes on until one row past the page has passed them, and
    // NEXT_OFFSET says how many rows that took. A view-numbered read gets one
    // database page and keeps what passes.
    const window = byView
      ? rows.slice(maxRows * (viewNumber - 1), maxRows * viewNumber)
      : rows.slice(offset ?? 0);
    const kept: Row[] = [];
    let readCount = 0;
    for (const row of window) {
      readCount += 1;
      if (requestReply.filter && !requestReply.filter(row)) continue;
      const permitted = permittedQueryRow(row, requestReply.auth, user, fieldNames);
      if (!permitted) continue;
      kept.push(permitted);
      if (kept.length > maxRows) break;
    }
    page = pageOf(kept, maxRows, offset, readCount - 1, !byView);
  }

  return {
    REPLY: wireRows(page.rows, source),
    PARAMETRIC_TYPE: parametricTypeOf(source.name, config, store),
    MESSAGE_TYPE: `REP_${resourceName}`,
    ...(byView && page.rows.length > 0 ? { NEXT_VIEW: viewNumber + 1 } : {}),
    ...(page.nextOffset !== undefined ? { NEXT_OFFSET: page.nextOffset } : {}),
    ...(page.moreRows !== undefined ? { MORE_ROWS: page.moreRows } : {}),
  };
}

// --- plain servers (RequestReply.kt) ----------------------------------------

// A REQUEST value as the field's type (DataTypeConverter.convert). Fails for a
// value the type can't take.
function convertValue(value: unknown, field: ResolvedField): unknown {
  const fail = () => {
    throw requestFailed(`Unable to convert ${String(value)} to ${field.type} for ${field.name}`);
  };
  switch (field.type) {
    case 'INT':
    case 'SHORT':
    case 'LONG':
    case 'NANO_TIMESTAMP':
    case 'DOUBLE':
    case 'BIGDECIMAL': {
      const number = typeof value === 'string' && value.trim() !== '' ? Number(value) : value;
      return typeof number === 'number' && Number.isFinite(number) ? number : fail();
    }
    case 'BOOLEAN':
      if (typeof value === 'boolean') return value;
      if (typeof value === 'string' && /^(true|false)$/i.test(value)) {
        return value.toLowerCase() === 'true';
      }
      return fail();
    case 'DATE':
    case 'DATETIME': {
      // yyyyMMdd as well as the criteria formats (yyyyMMdd-HH:mm[:ss[.SSS]],
      // ISO-8601, epoch millis).
      const day = typeof value === 'string' ? /^(\d{4})(\d{2})(\d{2})$/.exec(value) : null;
      if (day) return Date.UTC(Number(day[1]), Number(day[2]) - 1, Number(day[3]));
      return parseGenesisDateTime(value) ?? fail();
    }
    case 'STRING':
    case 'ENUM':
    case 'RAW':
      return typeof value === 'object' ? fail() : String(value);
    default:
      return value;
  }
}

// What Kotlin's toString() makes of a row value, for wildcard matching: a
// Double keeps its fraction ("2.0"), a BigDecimal its scale ("15.00000"), a
// date prints as ISO-8601, null is "null".
function kotlinString(value: unknown, field: ResolvedField | undefined): string {
  if (value === undefined || value === null) return 'null';
  if (field?.type === 'BIGDECIMAL') return decimalText(value, field.scale) ?? String(value);
  if (typeof value === 'number') {
    if ((field?.type === 'DATE' || field?.type === 'DATETIME') && Number.isFinite(value)) {
      return new Date(value).toISOString();
    }
    if (field?.type === 'DOUBLE' && Number.isInteger(value)) return value.toFixed(1);
  }
  return String(value);
}

// WildcardFilter.matchesWildCard: '*' at both ends is "contains", at the start
// "ends with", at the end "starts with" — all case-sensitive, after trimming
// every leading and trailing '*' ('*' alone matches anything, null included).
// A '*' only in the middle matches nothing.
function matchesWildcard(text: string, wildcard: string): boolean {
  const starts = wildcard.startsWith(WILDCARD);
  const ends = wildcard.endsWith(WILDCARD);
  const trimmed = wildcard.replace(/^\*+|\*+$/g, '');
  if (starts && ends) return text.includes(trimmed);
  return (starts && text.endsWith(trimmed)) || (ends && text.startsWith(trimmed));
}

const RANGE_SUFFIXES = ['_FROM', '_TO'];

// tryRemoveRangeSuffix: a key that isn't a field may name one's range end.
function fieldOfKey(key: string, source: Source): string {
  if (source.fields.has(key)) return key;
  let field = key;
  for (const suffix of RANGE_SUFFIXES) {
    if (field.endsWith(suffix)) field = field.slice(0, -suffix.length);
  }
  return field;
}

// One REQUEST object as a row predicate (RequestReplyBuilderImpl
// buildFieldPredicates + FieldPredicate.buildPredicate, then the
// WildcardFilter): undefined when it can't match anything (NoData).
// - A string containing '*' is a wildcard on that field.
// - Any other value must equal the field's (converted to its type; null
//   matches null, and nothing on a non-null field).
// - FIELD_FROM and FIELD_TO together are an inclusive range; either alone,
//   or a value it can't convert, matches nothing.
// - A field the entity doesn't have matches nothing — unless its value is
//   null (always true of an unknown field) or a wildcard (it reads as "null").
function requestPredicate(request: Row, source: Source): Predicate | undefined {
  const checks: Predicate[] = [];
  const byField = new Map<string, Array<[string, unknown]>>();
  for (const [key, value] of Object.entries(request)) {
    if (typeof value === 'string' && value.includes(WILDCARD)) {
      const field = source.fields.get(key);
      checks.push((row) => matchesWildcard(kotlinString(row[key], field), value));
      continue;
    }
    const field = fieldOfKey(key, source);
    byField.set(field, [...(byField.get(field) ?? []), [key, value]]);
  }
  for (const [name, values] of byField) {
    const field = source.fields.get(name);
    if (!field) {
      if (values[0][1] === null) continue;
      return undefined;
    }
    if (values.length === 1) {
      const [key, value] = values[0];
      if (key !== name) return undefined; // an incomplete range
      if (value === null) {
        if (!field.nullable) return undefined;
        checks.push((row) => row[name] === undefined || row[name] === null);
        continue;
      }
      const expected = convertValue(value, field);
      checks.push((row) => compareValues(row[name], expected) === 0);
      continue;
    }
    if (values.length > 2) return undefined;
    const lookup = new Map(values);
    const range = [lookup.get(`${name}_FROM`), lookup.get(`${name}_TO`)].map((value) => {
      if (value === undefined || value === null) return undefined;
      try {
        return convertValue(value, field);
      } catch {
        return undefined;
      }
    });
    const [from, to] = range;
    if (from === undefined || to === undefined) return undefined;
    checks.push((row) => {
      const value = row[name];
      if (value === undefined || value === null) return false;
      return compareValues(value, from) >= 0 && compareValues(value, to) <= 0;
    });
  }
  return (row) => checks.every((check) => check(row));
}

function plainReply(
  resourceName: string,
  requestReply: RequestReplyDef,
  message: GenesisMessage,
  config: MockServerConfig,
  store: Store,
  user: RequestingUser,
): Row {
  const details: Row = isPlainObject(message.DETAILS) ? message.DETAILS : {};
  const rowLimit =
    readInteger(details.MAX_ROWS, 'lenient') ?? requestReply.rowLimit ?? DEFAULT_ROW_LIMIT;
  // Flowable.take(-1)'s IllegalArgumentException.
  if (rowLimit < 0) throw requestFailed(`count >= 0 required but it was ${rowLimit}`);
  // REQUEST is top level only (DETAILS.REQUEST is never read). An array runs
  // each of its objects in turn; anything else is one request, empty when
  // missing.
  const requests: Row[] = Array.isArray(message.REQUEST)
    ? message.REQUEST.filter((request) => request !== null && request !== undefined).map(
        (request) => (isPlainObject(request) ? request : {}),
      )
    : [isPlainObject(message.REQUEST) ? message.REQUEST : {}];

  const source = sourceOf(requestReply, config, store);
  // Here a criteria that doesn't compile fails the whole request with its
  // message (handleThrowable: REQUEST_FAILED).
  const criteria = criteriaFilter(readText(details.CRITERIA_MATCH), source, (error) =>
    requestFailed(error.message),
  );
  const fieldNames = requestReplyFieldNames(requestReply, config, store);
  // The rows come through the table's (or the view's root table's) primary
  // key, as getBulk reads them (recorded: auth's RIGHT, in CODE order).
  const ordered = [...store.getResourceRows(source.name)].sort(
    requestReply.order ?? rowComparator(byPrimaryKey(source)),
  );
  const candidates = requestReply.filter ? ordered.filter(requestReply.filter) : ordered;

  const reply = requests.flatMap((request) => {
    const matches = requestPredicate(request, source);
    if (!matches) return [];
    const rows: Row[] = [];
    for (const row of applyCriteria(candidates.filter(matches), criteria)) {
      if (rows.length >= rowLimit) break;
      const permitted = permittedQueryRow(row, requestReply.auth, user, fieldNames);
      if (permitted) rows.push(permitted);
    }
    return rows;
  });

  // In RequestReply.kt's order: SOURCE_REF straight after MESSAGE_TYPE
  // (connection.reply overwrites it with the same value, keeping its place).
  return {
    MESSAGE_TYPE: `REP_${resourceName}`,
    SOURCE_REF: message.SOURCE_REF,
    PARAMETRIC_TYPE: parametricTypeOf(source.name, config, store),
    REPLY: wireRows(reply, source),
  };
}

// --- resolvers (custom request servers) -------------------------------------

function resolverReply(
  resourceName: string,
  requestReply: RequestReplyDef,
  message: GenesisMessage,
  config: MockServerConfig,
  store: Store,
  user: RequestingUser,
): Row {
  // The resolver owns REQUEST interpretation entirely — no generic
  // REQUEST/CRITERIA filtering is applied on top, since a computed reply's
  // rows needn't echo the request's fields at all. Its rows go out exactly as
  // returned.
  const request: Row =
    (typeof message.REQUEST === 'object' && message.REQUEST) || message.DETAILS?.REQUEST || {};
  const resolved = requestReply.resolver!({ request, message, user, store, config });
  const typed = requestReply.typed;
  if (!typed) {
    return {
      MESSAGE_TYPE: `REP_${resourceName}`,
      REPLY: applyResourceAuth(resolved, requestReply.auth, user),
    };
  }
  // A typed server (AbstractCustomReqRep.kt:62-92, CustomReqRep.kt:34-38
  // @v8.15.29): CRITERIA_MATCH compiled against the reply class, a criteria
  // GSF can't compile failing the request (GENERIC_ERROR, with its text); the
  // criteria over each whole row, then the resource's auth, then at most
  // MAX_ROWS of them (Flow.take, which refuses a count below 1), with no
  // paging.
  const details: Row = message.DETAILS ?? {};
  const maxRows =
    readInteger(details.MAX_ROWS, 'lenient') ?? requestReply.rowLimit ?? DEFAULT_ROW_LIMIT;
  if (maxRows < 1) throw new Error(`Requested element count ${maxRows} should be positive`);
  const fields = typedFields(typed.reply.fields);
  const replyClass: Source = {
    name: typed.reply.className,
    fields: new Map(fields.map((field) => [field.name, field])),
    pkFields: [],
    types: new Map(fields.map((field) => [field.name, field.type])),
  };
  const criteria = criteriaFilter(
    readText(details.CRITERIA_MATCH),
    replyClass,
    (error) => new Error(error.message),
  );
  const rows = applyResourceAuth(applyCriteria(resolved, criteria), requestReply.auth, user);
  return {
    REPLY: rows.slice(0, maxRows),
    MESSAGE_TYPE: `REP_${resourceName}`,
    PARAMETRIC_TYPE: typed.reply.className,
  };
}

// A failed request is a MSG_NACK. A NackError keeps its items. Anything else
// is what each kind of server makes of an unexpected exception: a custom
// request server's GENERIC_ERROR with the message (AbstractCustomReqRep.kt), a
// plain one's REQUEST_FAILED with the message (RequestReply.kt
// handleThrowable), a criteria-only one's INTERNAL_ERROR.
function failureReply(error: unknown, requestReply: RequestReplyDef): Row {
  if (error instanceof NackError) return serviceMsgNack(error.errors, error.warnings);
  const message = error instanceof Error ? error.message : String(error);
  const code = requestReply.resolver
    ? ErrorCode.GENERIC_ERROR
    : requestReply.criteriaOnly
      ? ErrorCode.INTERNAL_ERROR
      : ErrorCode.REQUEST_FAILED;
  const nack = NackError.standard(
    code,
    code === ErrorCode.INTERNAL_ERROR ? undefined : message || undefined,
  );
  return serviceMsgNack(nack.errors, nack.warnings);
}

export function handleRequestReply(
  message: GenesisMessage,
  connection: Connection,
  config: MockServerConfig,
  store: Store,
): void {
  const resourceName = message.MESSAGE_TYPE.replace(/^REQ_/, '');
  const requestReply = config.requestReplies?.[resourceName];

  if (!requestReply || (!requestReply.resolver && !requestReply.source)) {
    connection.reply(
      message,
      isLegacyFidelity(config)
        ? {
            MESSAGE_TYPE: `REP_${resourceName}`,
            ERROR: [
              {
                CODE: ERROR_CODE.UNKNOWN_RESOURCE,
                TEXT: `Unknown request-reply resource: ${resourceName}`,
              },
            ],
          }
        : resourceNotMappedNack(resourceName),
    );
    return;
  }

  if (isLegacyFidelity(config)) {
    legacyRequestReply(resourceName, requestReply, message, connection, config, store);
    return;
  }

  const user = requestingUser(connection);
  let reply: Row;
  try {
    if (requestReply.resolver) {
      reply = resolverReply(resourceName, requestReply, message, config, store, user);
    } else if (requestReply.criteriaOnly) {
      reply = criteriaOnlyReply(resourceName, requestReply, message, config, store, user);
    } else {
      reply = plainReply(resourceName, requestReply, message, config, store, user);
    }
  } catch (error) {
    reply = failureReply(error, requestReply);
  }
  connection.reply(message, reply);
}

// --- fidelity: 'legacy' ------------------------------------------------------
// What this engine answered up to 15.47: REQUEST from the top level or
// DETAILS, '*', null and '' matching anything, an array value matching any of
// its values, CRITERIA_MATCH failing open, every row (no MAX_ROWS, OFFSET or
// ORDER_BY), no PARAMETRIC_TYPE or MORE_ROWS, rows without RECORD_ID and
// TIMESTAMP, and errors as a REP_ with ERROR.

function matchesLegacyRequestFields(row: Row, request: Row): boolean {
  return Object.entries(request).every(([field, value]) => {
    if (value === undefined || value === null || value === WILDCARD || value === '') return true;
    if (Array.isArray(value)) return value.length === 0 || value.includes(row[field]);
    return row[field] === value;
  });
}

function legacyRequestReply(
  resourceName: string,
  requestReply: RequestReplyDef,
  message: GenesisMessage,
  connection: Connection,
  config: MockServerConfig,
  store: Store,
): void {
  const user = requestingUser(connection);
  const request: Row =
    (typeof message.REQUEST === 'object' && message.REQUEST) || message.DETAILS?.REQUEST || {};
  try {
    let rows: Row[];
    if (requestReply.resolver) {
      rows = requestReply.resolver({ request, message, user, store, config });
    } else {
      rows = store.getResourceRows(requestReply.source!);
      if (Object.keys(request).length > 0) {
        rows = rows.filter((row) => matchesLegacyRequestFields(row, request));
      }
      const criteria = message.DETAILS?.CRITERIA_MATCH ?? message.CRITERIA_MATCH;
      if (criteria) rows = filterByCriteria(rows, criteria, { lenient: true });
      if (requestReply.filter) rows = rows.filter(requestReply.filter);
    }
    rows = applyResourceAuth(rows, requestReply.auth, user).map(withoutRecordFields);
    connection.reply(message, { MESSAGE_TYPE: `REP_${resourceName}`, REPLY: rows });
  } catch (error) {
    connection.reply(message, { MESSAGE_TYPE: `REP_${resourceName}`, ...nackPayload(error, true) });
  }
}