// The reporting product's events and computed request servers, as its scripts
// at the version the showcase pins run them (reporting v8.15.3,
// scripts/platform-preset/seeds/reporting@8.15.3/scripts:
// reporting-eventhandler.kts below as EH, reporting-reqrep.kts as RR), with
// what they call in the product's ReportingManager (reporting-app 8.15.3).
//
// A report run writes its GENERATED_REPORT row, with the file name, the
// status and a file id, but no file: the datasources aren't run and nothing
// is rendered. Delivering it to a notify destination raises the notification
// genesis-notify routes; an SFTP destination fails, as the product's TODO()
// does.
import { randomUUID } from 'node:crypto';
import { compareRecordIds, compareValues } from '../db/ordering.ts';
import { primaryKeyIndexName } from '../db/schema.ts';
import { checkUnique } from '../handlers/crudEvents.ts';
import { requestServerReplyProperties } from '../handlers/jsonSchema.ts';
import { listResources } from '../handlers/resources.ts';
import { NackError } from '../protocol/errors.ts';
import type {
  EventHandler,
  EventHandlerCtx,
  EventSchemaOverride,
  FieldDef,
  RequestReplyDef,
  Row,
} from '../types.ts';
import { cronExpressionError } from './cronExpression.ts';
import { newFileStorageId } from './fileServer.ts';
import {
  hostZone,
  knownZone,
  jodaFormat,
  jodaIso,
  JVM_ZONE_IDS,
  kotlinString,
  zoneIdError,
} from './javaTime.ts';
import { dto, nullable } from './notify.ts';
import type { RaisedNotification } from './notifyDispatch.ts';

// --- schemas ---------------------------------------------------------------------

const EVENT_CLASS = 'global.genesis.reporting.message.event';
const DAO_EVENT_CLASS = 'global.genesis.gen.dao.event';
// The product's enums are its tables' own: InputType is
// global.genesis.gen.dao.enums.reporting.report_datasource.InputType.
const enumOf = (table: string, name: string, values: string[], more: Partial<FieldDef> = {}) =>
  dto(name, 'ENUM', {
    values,
    enumClass: `global.genesis.gen.dao.enums.reporting.${table}.${name
      .toLowerCase()
      .replace(/(^|_)([a-z])/g, (_match, _sep, letter: string) => letter.toUpperCase())}`,
    ...more,
  });
const optional = { optional: true };
const text = (name: string) => dto(name, 'STRING', nullable);

// DatasourceConfigurationInsert: the transformer configuration (any object
// here: its expression tree isn't described) and ReportDatasourceInsert
// unwrapped.
const DATASOURCE: FieldDef = dto('DATA_SOURCES', 'ARRAY', {
  items: {
    type: 'OBJECT',
    className: `${EVENT_CLASS}.DatasourceConfigurationInsert`,
    nullable: false,
    fields: [
      dto('TRANSFORMER_CONFIGURATION', 'MAP', nullable),
      dto('KEY', 'STRING'),
      dto('NAME', 'STRING'),
      enumOf('report_datasource', 'INPUT_TYPE', ['REQ_REP', 'DATA_PIPELINE'], optional),
      enumOf('report_datasource', 'OUTPUT_TYPE', ['TABLE', 'RECORD_STREAM', 'CHART'], optional),
      text('GROUP_KEY'),
      enumOf(
        'report_datasource',
        'GROUPING_STRATEGY',
        ['NONE', 'MULTI_SHEET', 'LOOKUP_TABLE'],
        optional,
      ),
    ],
  },
});
const DESTINATION_IDS = dto('DESTINATION_IDS', 'ARRAY', { items: { type: 'STRING' }, ...optional });
const PARAMETERS = dto('PARAMETERS', 'ARRAY', {
  ...optional,
  items: {
    type: 'OBJECT',
    className: `${DAO_EVENT_CLASS}.ReportParameterInsert`,
    nullable: false,
    fields: [
      dto('PARAMETER_NAME', 'STRING'),
      text('REPORT_DATASOURCE'),
      text('REFERENCE_FIELD'),
      enumOf('report_parameter', 'SOURCE', ['User Input', 'Lookup'], optional),
      text('LOOKUP_RESOURCE'),
      text('LOOKUP_DATA_FIELD'),
      text('LOOKUP_DISPLAY_FIELD'),
    ],
  },
});
const SCHEDULES = dto('SCHEDULES', 'ARRAY', {
  ...optional,
  items: {
    type: 'OBJECT',
    className: 'global.genesis.reporting.message.common.ReportSchedule',
    nullable: false,
    fields: [dto('CRON_EXPRESSION', 'STRING'), dto('TIME_ZONE', 'STRING')],
  },
});
const OUTPUT_FORMATS = ['CSV', 'PDF', 'HTML', 'XLSX'];
// The fields of a generated DAO event class (global.genesis.gen.dao.event) a
// DTO unwraps: described with the DAO's annotations, its STRING lengths
// (reporting-tables-dictionary.kts) checked.
const fromDao =
  (lengths: Record<string, number>) =>
  (field: FieldDef): FieldDef => ({
    ...field,
    dao: true,
    ...(lengths[field.name] ? { maxSize: lengths[field.name] } : {}),
  });
const REPORT_CONFIG_LENGTHS: Record<string, number> = {
  NAME: 256,
  DESCRIPTION: 10_000,
  FILE_NAME: 256,
  OUTPUT_DIRECTORY: 1024,
};
const savedReport = fromDao({
  REPORT_COLUMNS: 10_000,
  REPORT_DATASOURCE: 200,
  REPORT_DESCRIPTION: 10_000,
});
// ReportConfigInsert unwrapped: defaults everywhere but FILE_NAME and NAME.
const CONFIG_INSERT: FieldDef[] = [
  dto('AUTO_DISPATCH', 'BOOLEAN', optional),
  text('DESCRIPTION'),
  text('DOCUMENT_TEMPLATE_ID'),
  dto('FILE_NAME', 'STRING'),
  text('ID'),
  dto('NAME', 'STRING'),
  text('OUTPUT_DIRECTORY'),
  enumOf('report_config', 'OUTPUT_FORMAT', OUTPUT_FORMATS, optional),
].map(fromDao(REPORT_CONFIG_LENGTHS));
// ReportConfigModify unwrapped: a modify class has no defaults.
const CONFIG_MODIFY: FieldDef[] = [
  dto('AUTO_DISPATCH', 'BOOLEAN'),
  dto('DESCRIPTION', 'STRING', { nullable: true }),
  dto('DOCUMENT_TEMPLATE_ID', 'STRING', { nullable: true }),
  dto('FILE_NAME', 'STRING'),
  dto('ID', 'STRING'),
  dto('NAME', 'STRING'),
  dto('OUTPUT_DIRECTORY', 'STRING', { nullable: true }),
  enumOf('report_config', 'OUTPUT_FORMAT', OUTPUT_FORMATS),
].map(fromDao(REPORT_CONFIG_LENGTHS));
const byId = (table: string, className: string): EventSchemaOverride => ({
  table,
  pkOnly: true,
  className: `global.genesis.gen.dao.${className}`,
});

const SCHEMAS: Record<string, EventSchemaOverride> = {
  // SavedReportsInsertWrapper: its own REPORT_DATA_SOURCE, SavedReportsInsert
  // unwrapped.
  EVENT_SAVED_REPORTS_INSERT: {
    className: `${EVENT_CLASS}.SavedReportsInsertWrapper`,
    fields: [
      text('REPORT_DATA_SOURCE'),
      ...[
        text('REPORT_COLUMNS'),
        text('REPORT_DATASOURCE'),
        text('REPORT_DESCRIPTION'),
        text('REPORT_ID'),
        dto('REPORT_NAME', 'STRING'),
      ].map(savedReport),
    ],
  },
  EVENT_SAVED_REPORTS_UPDATE: {
    className: `${EVENT_CLASS}.SavedReportsModifyWrapper`,
    fields: [
      savedReport(dto('REPORT_COLUMNS', 'STRING', { nullable: true })),
      text('REPORT_DATA_SOURCE'),
      savedReport(text('REPORT_DATASOURCE')),
      savedReport(dto('REPORT_DESCRIPTION', 'STRING', { nullable: true })),
      savedReport(dto('REPORT_ID', 'STRING')),
      savedReport(dto('REPORT_NAME', 'STRING')),
    ],
  },
  EVENT_SAVED_REPORTS_DELETE: byId('SAVED_REPORTS', 'SavedReports.ByReportId'),
  EVENT_REPORT_CONFIG_INSERT: {
    className: `${EVENT_CLASS}.ReportingConfigurationInsert`,
    fields: [DATASOURCE, DESTINATION_IDS, PARAMETERS, ...CONFIG_INSERT, SCHEDULES],
  },
  EVENT_REPORT_CONFIG_UPDATE: {
    className: `${EVENT_CLASS}.ReportingConfigurationModify`,
    fields: [DATASOURCE, DESTINATION_IDS, PARAMETERS, ...CONFIG_MODIFY, SCHEDULES],
  },
  EVENT_REPORT_CONFIG_DELETE: byId('REPORT_CONFIG', 'ReportConfig.ById'),
  EVENT_NOTIFY_DESTINATION_INSERT: {
    className: `${EVENT_CLASS}.NotifyDestinationInsert`,
    fields: [dto('NAME', 'STRING'), dto('TOPIC', 'STRING')],
  },
  EVENT_NOTIFY_DESTINATION_UPDATE: {
    className: `${EVENT_CLASS}.NotifyDestinationUpdate`,
    fields: [dto('ID', 'STRING'), dto('NAME', 'STRING'), dto('TOPIC', 'STRING')],
  },
  EVENT_NOTIFY_DESTINATION_DELETE: byId('REPORT_DESTINATION', 'ReportDestination.ById'),
  // SMTP in the names, SFTP in the classes and tables.
  EVENT_SMTP_DESTINATION_INSERT: {
    className: `${EVENT_CLASS}.SftpDestinationInsert`,
    fields: [dto('NAME', 'STRING'), dto('OUTPUT_DIRECTORY', 'STRING')],
  },
  EVENT_SMTP_DESTINATION_UPDATE: {
    className: `${EVENT_CLASS}.SftpDestinationUpdate`,
    fields: [dto('ID', 'STRING'), dto('NAME', 'STRING'), dto('OUTPUT_DIRECTORY', 'STRING')],
  },
  EVENT_SMTP_DESTINATION_DELETE: byId('REPORT_DESTINATION', 'ReportDestination.ById'),
  EVENT_GENERATE_REPORT: {
    className: `${EVENT_CLASS}.GenerateReport`,
    fields: [dto('ID', 'STRING'), dto('PARAMETERS', 'MAP', optional)],
  },
  EVENT_DISPATCH_REPORT: byId('GENERATED_REPORT', 'GeneratedReport.ById'),
  EVENT_GENERATED_REPORT_DELETE: byId('GENERATED_REPORT', 'GeneratedReport.ById'),
};

// --- writes ------------------------------------------------------------------------

// entityDb's writes: unique indexes checked (DUPLICATE_KEY), a missing row
// RECORD_NOT_FOUND, each change pushed live.
function insert(ctx: EventHandlerCtx, table: string, row: Row): Row {
  checkUnique(table, ctx.config.tables![table], row, ctx.store);
  const written = ctx.store.insertRow(table, row);
  ctx.broadcastTableChange(table, 'INSERT', ctx.store.rowKey(table, written));
  return written;
}
function modify(ctx: EventHandlerCtx, table: string, key: string, patch: Row): Row {
  const def = ctx.config.tables![table];
  const existing = ctx.store.getRow(table, key);
  if (!existing) throw NackError.recordNotFound(table, primaryKeyIndexName(table, def));
  checkUnique(table, def, { ...existing, ...patch }, ctx.store, key);
  const written = ctx.store.updateRow(table, key, patch)!;
  ctx.broadcastTableChange(table, 'MODIFY', key);
  return written;
}
function remove(ctx: EventHandlerCtx, table: string, key: string): Row {
  const existing = ctx.store.getRow(table, key);
  if (!existing || !ctx.store.deleteRow(table, key)) throw NackError.recordNotFound(table);
  ctx.broadcastTableChange(table, 'DELETE', key);
  return existing;
}
// A configuration's rows of a table. Its destinations come in DESTINATION_ID
// order, as the range read over ByReportConfigIdDestinationId gives them.
const rowsOf = (ctx: { store: EventHandlerCtx['store'] }, table: string, configId: string) => {
  const rows = ctx.store.getAllRows(table).filter((row) => row.REPORT_CONFIG_ID === configId);
  return table === 'REPORT_CONFIG_DESTINATION_MAP'
    ? rows.sort((a, b) => compareValues(a.DESTINATION_ID, b.DESTINATION_ID))
    : rows;
};
const removeAll = (ctx: EventHandlerCtx, table: string, configId: string) => {
  for (const row of rowsOf(ctx, table, configId)) remove(ctx, table, ctx.store.rowKey(table, row));
};

// --- report configurations ---------------------------------------------------------

// TransformerConfiguration (genesis-transform): every property defaulted, so
// it's stored, and read back, whole.
const transformerDefaults = (): Row => ({
  ROW_FILTERS: { OPERATION: 'AND', EXPRESSIONS: [], TYPE: 'PREDICATE_EXPRESSION' },
  INCLUDE_COLUMNS: [],
  COLUMN_TRANSFORMS: {},
  DERIVED_FIELDS: {},
  GROUP_BY_COLUMNS: { COLUMNS: [], AGGREGATION_OPERATIONS: [] },
  SORT_BY_COLUMNS: { DIRECTION: 'ASCENDING', COLUMNS: [] },
  COLUMN_FORMATS: {},
  COLUMN_RENAMES: {},
  DATE_TIME_FORMAT_TIMEZONE: null,
});
const isObject = (value: unknown): value is Row =>
  typeof value === 'object' && value !== null && !Array.isArray(value);
function transformerConfiguration(sent: unknown): Row {
  const defaults = transformerDefaults();
  if (!isObject(sent)) return defaults;
  return Object.fromEntries(
    Object.entries(defaults).map(([key, fallback]) => {
      const value = sent[key];
      if (value === undefined || value === null) return [key, fallback];
      return [
        key,
        isObject(fallback) && isObject(value) && !('TYPE' in fallback)
          ? { ...fallback, ...value }
          : value,
      ];
    }),
  );
}
// The stored column: the configuration as Jackson writes it, "null" for an
// explicit null.
const storedTransformer = (sent: unknown) =>
  sent === null ? 'null' : JSON.stringify(transformerConfiguration(sent));
// RR :119: the stored string read back, a null or "null" the defaults.
function readTransformer(stored: unknown): Row {
  if (typeof stored !== 'string') return transformerDefaults();
  try {
    return transformerConfiguration(JSON.parse(stored));
  } catch {
    return transformerDefaults();
  }
}

// validateSchedules (EH :42-67): each schedule's cron expression (log4j's
// CronExpression) and zone (ZoneId.of), every failure in one NACK.
function scheduleErrors(schedules: Row[]): string[] {
  const errors: string[] = [];
  schedules.forEach((schedule, index) => {
    const cron = cronExpressionError(String(schedule.CRON_EXPRESSION));
    if (cron !== undefined) {
      errors.push(
        `Cron expression ${schedule.CRON_EXPRESSION} in schedule at position ${index} in list is invalid: ${cron}`,
      );
    }
    const zone = zoneIdError(String(schedule.TIME_ZONE));
    if (zone !== undefined) {
      errors.push(
        `Timezone ID ${schedule.TIME_ZONE} in schedule at position ${index} in list is invalid: ${zone}`,
      );
    }
  });
  return errors;
}
function checkSchedules(details: Row): void {
  const errors = scheduleErrors(details.SCHEDULES ?? []);
  if (errors.length) {
    throw new NackError(
      errors.map((TEXT) => ({ '@type': 'StandardError', CODE: 'VALIDATION_ERROR', TEXT })),
    );
  }
}

// The rows a configuration's DATA_SOURCES, DESTINATION_IDS, PARAMETERS and
// SCHEDULES make (EH :157-192).
function writeDatasources(ctx: EventHandlerCtx, configId: string, details: Row) {
  for (const source of (details.DATA_SOURCES as Row[] | undefined) ?? []) {
    insert(ctx, 'REPORT_DATASOURCE', {
      REPORT_CONFIG_ID: configId,
      KEY: source.KEY,
      NAME: source.NAME,
      INPUT_TYPE: source.INPUT_TYPE ?? 'REQ_REP',
      OUTPUT_TYPE: source.OUTPUT_TYPE ?? 'TABLE',
      TRANSFORMER_CONFIGURATION: storedTransformer(source.TRANSFORMER_CONFIGURATION),
      GROUP_KEY: source.GROUP_KEY ?? null,
      GROUPING_STRATEGY: source.GROUPING_STRATEGY ?? 'NONE',
    });
  }
}
function writeDestinations(ctx: EventHandlerCtx, configId: string, details: Row) {
  for (const DESTINATION_ID of (details.DESTINATION_IDS as string[] | undefined) ?? []) {
    insert(ctx, 'REPORT_CONFIG_DESTINATION_MAP', { REPORT_CONFIG_ID: configId, DESTINATION_ID });
  }
}
function writeParameters(ctx: EventHandlerCtx, configId: string, details: Row) {
  for (const parameter of (details.PARAMETERS as Row[] | undefined) ?? []) {
    insert(ctx, 'REPORT_PARAMETER', {
      REPORT_CONFIG_ID: configId,
      PARAMETER_NAME: parameter.PARAMETER_NAME,
      REPORT_DATASOURCE: parameter.REPORT_DATASOURCE ?? null,
      REFERENCE_FIELD: parameter.REFERENCE_FIELD ?? null,
      SOURCE: parameter.SOURCE ?? 'User Input',
      LOOKUP_RESOURCE: parameter.LOOKUP_RESOURCE ?? null,
      LOOKUP_DATA_FIELD: parameter.LOOKUP_DATA_FIELD ?? null,
      LOOKUP_DISPLAY_FIELD: parameter.LOOKUP_DISPLAY_FIELD ?? null,
    });
  }
}
// createCronRule (EH :69-95): a rule the cron scheduler would fire
// EVENT_GENERATE_REPORT {ID} on; the mock has no scheduler.
function writeSchedules(ctx: EventHandlerCtx, configId: string, details: Row, user: string) {
  ((details.SCHEDULES as Row[] | undefined) ?? []).forEach((schedule, index) => {
    const rule = insert(ctx, 'CRON_RULE', {
      NAME: randomUUID(),
      DESCRIPTION: `Schedule rule #${index} for Report Config '${details.NAME}'`,
      USER_NAME: user,
      CRON_EXPRESSION: schedule.CRON_EXPRESSION,
      TIME_ZONE: schedule.TIME_ZONE,
      RULE_STATUS: 'ENABLED',
      MESSAGE_TYPE: 'EVENT_GENERATE_REPORT',
      RESULT_EXPRESSION: `((ID = "${configId}"))`,
      PROCESS_NAME: null,
      RULE_TYPE: null,
    });
    insert(ctx, 'REPORT_CONFIG_CRON_RULE_MAP', {
      REPORT_CONFIG_ID: configId,
      CRON_RULE_NAME: rule.NAME,
    });
  });
}
// A configuration's schedules: its cron-map rows and the rules they name.
function removeSchedules(ctx: EventHandlerCtx, configId: string) {
  const maps = rowsOf(ctx, 'REPORT_CONFIG_CRON_RULE_MAP', configId);
  for (const map of maps)
    remove(
      ctx,
      'REPORT_CONFIG_CRON_RULE_MAP',
      ctx.store.rowKey('REPORT_CONFIG_CRON_RULE_MAP', map),
    );
  // filterNotNull (EH :253-255, :290-292): a rule that's gone is skipped.
  for (const map of maps) {
    if (ctx.store.getRow('CRON_RULE', map.CRON_RULE_NAME)) {
      remove(ctx, 'CRON_RULE', map.CRON_RULE_NAME);
    }
  }
}

// The configuration's columns an insert or a modify writes.
const configColumns = (details: Row, now: number, user: string): Row => ({
  NAME: details.NAME,
  DESCRIPTION: details.DESCRIPTION ?? null,
  LAST_UPDATED_AT: now,
  LAST_UPDATED_BY: user,
  OUTPUT_FORMAT: details.OUTPUT_FORMAT ?? 'CSV',
  FILE_NAME: details.FILE_NAME,
  OUTPUT_DIRECTORY: details.OUTPUT_DIRECTORY ?? null,
  DOCUMENT_TEMPLATE_ID: details.DOCUMENT_TEMPLATE_ID ?? null,
  AUTO_DISPATCH: details.AUTO_DISPATCH ?? false,
});

// --- saved reports (EH :28-40) -------------------------------------------------------

// createSavedReports: every value a string, a null "null", the columns under
// REPORT_COLUMN, the time as Joda's DateTime.toString() in the JVM's zone.
const savedReportReply = (row: Row): Row[] => [
  {
    REPORT_ID: row.REPORT_ID,
    REPORT_NAME: row.REPORT_NAME,
    CREATED_BY: row.CREATED_BY ?? 'null',
    REPORT_DESCRIPTION: row.REPORT_DESCRIPTION ?? 'null',
    REPORT_COLUMN: row.REPORT_COLUMNS ?? 'null',
    REPORT_CREATED_ON: jodaIso(Number(row.REPORT_CREATED_ON), hostZone()),
    REPORT_DATASOURCE: row.REPORT_DATASOURCE ?? 'null',
  },
];

// --- runs ----------------------------------------------------------------------------

// kotlin.NotImplementedError, which TODO() throws: an Error, not an
// Exception, so generateReport's catch lets it through.
class NotImplemented extends Error {}
// A Kotlin `!!` on null (Intrinsics.checkNotNull): an exception with no
// message.
class NullPointer extends Error {}

// The report's status, from STATUS's labels.
const PENDING = 'Pending';
const GENERATED = 'Generated';
const DELIVERED = 'Delivered';
const PARTIALLY_DELIVERED = 'Partially Delivered';
const FAILED = 'Failed';

const SUBSTITUTION = /\$\(([^:)]+)(?::([^)]*))?\)/g;

export interface ReportingOptions {
  // Who an event runs as.
  userOf(ctx: EventHandlerCtx): string;
  // The sysdef's REPORTING_DEFAULT_TIMEZONE: the zone `$(timestamp:...)` in a
  // file name is written in. Unset: the JVM's, which is the host's here.
  timeZone?: string;
  // The sysdef's REPORTING_DATASOURCE_LIST: the request servers
  // ALL_REPORT_DATASOURCES offers. Unset or empty: all of them.
  datasources?: string[];
  // NotificationService, when genesis-notify is in the app.
  notify?: (ctx: EventHandlerCtx, user: string, notification: RaisedNotification) => void;
}

export interface ReportingBacking {
  eventHandlers: Record<string, EventHandler>;
  eventSchemas: Record<string, EventSchemaOverride>;
  requestReplies: Record<string, RequestReplyDef>;
}

export function reportingBacking(options: ReportingOptions): ReportingBacking {
  // DateTimeZone.forID of the sysdef's zone; one it can't resolve is logged
  // ("Unable to resolve timezone") and the JVM's used instead.
  const zone = () =>
    options.timeZone !== undefined && knownZone(options.timeZone) ? options.timeZone : hostZone();

  // resolveSubstitutions (reporting-app core/UtilsKt): $(timestamp:<Joda
  // pattern>), $(parameter:<key>); anything else left as written.
  const substitute = (template: unknown, parameters: Row, now: number) =>
    template === null || template === undefined
      ? template
      : String(template).replace(SUBSTITUTION, (match, type: string, argument?: string) => {
          if (type === 'timestamp') {
            return (
              (argument === undefined ? undefined : jodaFormat(argument, now, zone())) ??
              jodaIso(now, zone())
            );
          }
          if (type === 'parameter') {
            // The parameters are a Map: a key it doesn't have is null, and
            // `$(parameter)` names none (`!!`).
            if (argument === undefined) throw new NullPointer();
            const value = Object.hasOwn(parameters, argument) ? parameters[argument] : undefined;
            return value === undefined || value === null ? 'null' : kotlinString(value);
          }
          return match;
        });

  // buildParameters (ReportingManager): a parameter linked to a datasource
  // names one the configuration has, and the field it reads.
  function checkParameters(ctx: EventHandlerCtx, configId: string, sent: Row, keys: Set<string>) {
    if (!Object.keys(sent).length) return;
    const defined = new Map(
      rowsOf(ctx, 'REPORT_PARAMETER', configId).map((row) => [row.PARAMETER_NAME as string, row]),
    );
    for (const name of Object.keys(sent)) {
      const definition = defined.get(name);
      if (
        !definition ||
        definition.REPORT_DATASOURCE === null ||
        definition.REPORT_DATASOURCE === undefined
      )
        continue;
      if (!keys.has(definition.REPORT_DATASOURCE)) {
        throw new NackError(
          `Could not find datasource ${definition.REPORT_DATASOURCE} referenced by parameter ${definition.PARAMETER_NAME}`,
        );
      }
      if (definition.REFERENCE_FIELD === null || definition.REFERENCE_FIELD === undefined) {
        throw new NackError('A parameter linked to a datasource must specify a reference field.');
      }
    }
  }

  // A request-server datasource the cluster doesn't serve (ReqRepMetadataCache).
  function checkDatasources(ctx: EventHandlerCtx, sources: Row[]) {
    const served = new Set(
      listResources(ctx.config)
        .filter((resource) => resource.RESOURCE_TYPE === 'REQUEST_SERVER')
        .map((resource) => resource.RESOURCE_NAME),
    );
    for (const source of sources) {
      if (source.INPUT_TYPE === 'REQ_REP' && !served.has(source.NAME)) {
        throw new NackError(
          `Unable to resolve client for feature ${source.NAME}, unable to generate report`,
        );
      }
    }
  }

  // dispatchReport (ReportingManager): DELIVERED unless a destination is
  // gone or has no notify record; a notify one gets the report's
  // notification, an SFTP one TODO().
  function dispatch(ctx: EventHandlerCtx, report: Row, config: Row, user: string): string {
    let status = DELIVERED;
    for (const map of rowsOf(ctx, 'REPORT_CONFIG_DESTINATION_MAP', config.ID)) {
      const destination = ctx.store.getRow('REPORT_DESTINATION', map.DESTINATION_ID);
      if (!destination) {
        status = PARTIALLY_DELIVERED;
        continue;
      }
      if (destination.TYPE === 'SFTP') throw new NotImplemented();
      const notifyDestination = ctx.store
        .getViewRows('NOTIFY_REPORT_DESTINATION')
        .find((row) => row.ID === destination.ID);
      if (!notifyDestination) {
        status = PARTIALLY_DELIVERED;
        continue;
      }
      // NotifyDispatcher.send.
      if (!report.REPORT_FILE_STORAGE_ID) {
        throw new NackError('Unable to dispatch report with no file storage ID');
      }
      options.notify?.(ctx, user, {
        HEADER: `New Report Generated: ${config.DESCRIPTION ?? 'null'}`,
        BODY: `A new report has been generated with name ${config.FILE_NAME}`,
        NOTIFY_SEVERITY: 'Information',
        ROUTING_TYPE: 'TOPIC',
        ROUTING_DATA: notifyDestination.TOPIC ?? null,
        NOTIFY_ATTACHMENTS: [report.REPORT_FILE_STORAGE_ID],
      });
    }
    return status;
  }

  // GENERATE_REPORT isn't transactional: what generateReport wrote before
  // failing stays, though the event NACKs. The engine rolls the event back,
  // so the row is written again once it has.
  function keepAfterNack(ctx: EventHandlerCtx, report: Row, patch: Row) {
    const { GENERATED_REPORT_ID, RECORD_ID, TIMESTAMP, ...columns } = report;
    setImmediate(() => {
      const { store } = ctx;
      if (store.getRow('GENERATED_REPORT', GENERATED_REPORT_ID)) {
        store.updateRow('GENERATED_REPORT', GENERATED_REPORT_ID, patch);
        ctx.broadcastTableChange('GENERATED_REPORT', 'MODIFY', GENERATED_REPORT_ID);
        return;
      }
      // The row the event wrote, its id included.
      const kept = store.insertRow('GENERATED_REPORT', {
        GENERATED_REPORT_ID,
        ...columns,
        ...patch,
      });
      ctx.broadcastTableChange('GENERATED_REPORT', 'INSERT', kept.GENERATED_REPORT_ID);
    });
  }

  const userOf = options.userOf;

  const eventHandlers: Record<string, EventHandler> = {
    // EH :99-113: REPORT_DATASOURCE, else the legacy REPORT_DATA_SOURCE.
    EVENT_SAVED_REPORTS_INSERT: (details, ctx) => {
      if (ctx.validate) return {};
      const row = insert(ctx, 'SAVED_REPORTS', {
        ...(details.REPORT_ID === undefined || details.REPORT_ID === null
          ? {}
          : { REPORT_ID: details.REPORT_ID }),
        REPORT_NAME: details.REPORT_NAME,
        REPORT_DESCRIPTION: details.REPORT_DESCRIPTION ?? null,
        REPORT_DATASOURCE: details.REPORT_DATASOURCE ?? details.REPORT_DATA_SOURCE ?? null,
        REPORT_COLUMNS: details.REPORT_COLUMNS ?? null,
        CREATED_BY: userOf(ctx),
        REPORT_CREATED_ON: Date.now(),
      });
      return { generated: savedReportReply(row) };
    },
    // EH :116-129: the editor becomes the creator, now the creation time.
    EVENT_SAVED_REPORTS_UPDATE: (details, ctx) => {
      if (ctx.validate) return {};
      if (!ctx.store.getRow('SAVED_REPORTS', details.REPORT_ID)) {
        throw NackError.nack(`Report with id ${details.REPORT_ID} does not exist`);
      }
      const row = modify(ctx, 'SAVED_REPORTS', details.REPORT_ID, {
        REPORT_NAME: details.REPORT_NAME,
        REPORT_DESCRIPTION: details.REPORT_DESCRIPTION ?? null,
        REPORT_DATASOURCE: details.REPORT_DATASOURCE ?? details.REPORT_DATA_SOURCE ?? null,
        REPORT_COLUMNS: details.REPORT_COLUMNS ?? null,
        CREATED_BY: userOf(ctx),
        REPORT_CREATED_ON: Date.now(),
      });
      return { generated: savedReportReply(row) };
    },
    EVENT_SAVED_REPORTS_DELETE: (details, ctx) => {
      if (ctx.validate) return {};
      remove(ctx, 'SAVED_REPORTS', details.REPORT_ID);
      return {};
    },
    // EH :139-201
    EVENT_REPORT_CONFIG_INSERT: (details, ctx) => {
      checkSchedules(details);
      if (ctx.validate) return {};
      const user = userOf(ctx);
      const now = Date.now();
      const config = insert(ctx, 'REPORT_CONFIG', {
        ...(details.ID === undefined || details.ID === null ? {} : { ID: details.ID }),
        ...configColumns(details, now, user),
        CREATED_ON: now,
        CREATED_BY: user,
      });
      writeDatasources(ctx, config.ID, details);
      writeDestinations(ctx, config.ID, details);
      writeParameters(ctx, config.ID, details);
      writeSchedules(ctx, config.ID, details, user);
      return { generated: [{ REPORT_CONFIG_ID: config.ID }] };
    },
    // EH :204-278: the configuration modified, everything under it replaced.
    EVENT_REPORT_CONFIG_UPDATE: (details, ctx) => {
      checkSchedules(details);
      if (ctx.validate) return {};
      const user = userOf(ctx);
      const id = details.ID;
      modify(ctx, 'REPORT_CONFIG', id, configColumns(details, Date.now(), user));
      removeAll(ctx, 'REPORT_DATASOURCE', id);
      writeDatasources(ctx, id, details);
      removeAll(ctx, 'REPORT_PARAMETER', id);
      writeParameters(ctx, id, details);
      removeAll(ctx, 'REPORT_CONFIG_DESTINATION_MAP', id);
      writeDestinations(ctx, id, details);
      removeSchedules(ctx, id);
      writeSchedules(ctx, id, details, user);
      return {};
    },
    // EH :281-297: its destination maps and generated reports are left.
    EVENT_REPORT_CONFIG_DELETE: (details, ctx) => {
      if (ctx.validate) return {};
      remove(ctx, 'REPORT_CONFIG', details.ID);
      removeAll(ctx, 'REPORT_DATASOURCE', details.ID);
      removeAll(ctx, 'REPORT_PARAMETER', details.ID);
      removeSchedules(ctx, details.ID);
      return {};
    },
    ...destinationHandlers('NOTIFY', 'NOTIFY_DESTINATION', 'TOPIC'),
    ...destinationHandlers('SMTP', 'SFTP_DESTINATION', 'OUTPUT_DIRECTORY'),
    // EH :394-408, ReportingManager.generateReport.
    EVENT_GENERATE_REPORT: (details, ctx) => {
      if (ctx.validate) return {};
      const config = ctx.store.getRow('REPORT_CONFIG', details.ID);
      if (!config) {
        throw new NackError(
          `ID ${details.ID} did not correspond to any known report configuration`,
        );
      }
      const sources = rowsOf(ctx, 'REPORT_DATASOURCE', config.ID);
      if (!sources.length) {
        throw new NackError(
          'Report configuration must have at least one source of data configured',
        );
      }
      const user = userOf(ctx);
      const now = Date.now();
      const report = insert(ctx, 'GENERATED_REPORT', {
        REPORT_CONFIG_ID: config.ID,
        CREATED_BY: user,
        CREATED_ON: now,
        STATUS: PENDING,
        FILE_NAME: null,
        REPORT_FILE_STORAGE_ID: null,
        MESSAGE: null,
      });
      try {
        const parameters = (details.PARAMETERS as Row | undefined) ?? {};
        checkParameters(ctx, config.ID, parameters, new Set(sources.map((source) => source.KEY)));
        const fileName = substitute(config.FILE_NAME, parameters, now);
        // Resolved too (generateReport), though the mock writes no file there.
        substitute(config.OUTPUT_DIRECTORY, parameters, now);
        checkDatasources(ctx, sources);
        // What file-server would give the stored file.
        const stored = { ...report, REPORT_FILE_STORAGE_ID: newFileStorageId() };
        const destinations = rowsOf(ctx, 'REPORT_CONFIG_DESTINATION_MAP', config.ID);
        let status = GENERATED;
        if (destinations.length)
          status = config.AUTO_DISPATCH === true ? dispatch(ctx, stored, config, user) : PENDING;
        modify(ctx, 'GENERATED_REPORT', report.GENERATED_REPORT_ID, {
          FILE_NAME: `${fileName}.${String(config.OUTPUT_FORMAT).toLowerCase()}`,
          REPORT_FILE_STORAGE_ID: stored.REPORT_FILE_STORAGE_ID,
          STATUS: status,
        });
        return { generated: [{ GENERATED_REPORT_ID: report.GENERATED_REPORT_ID }] };
      } catch (error) {
        if (error instanceof NotImplemented) {
          keepAfterNack(ctx, report, {});
          throw NackError.standard('INTERNAL_ERROR', 'An operation is not implemented.');
        }
        if (error instanceof NullPointer) {
          keepAfterNack(ctx, report, { STATUS: FAILED, MESSAGE: null });
          throw NackError.standard('INTERNAL_ERROR');
        }
        const message =
          error instanceof NackError ? error.errors[0]?.TEXT : (error as Error).message;
        keepAfterNack(ctx, report, { STATUS: FAILED, MESSAGE: message ?? null });
        throw error;
      }
    },
    // EH :417-461: checked even under VALIDATE.
    EVENT_DISPATCH_REPORT: (details, ctx) => {
      const id = details.GENERATED_REPORT_ID;
      const report = ctx.store.getRow('GENERATED_REPORT', id);
      if (!report)
        throw new NackError(`GENERATED_REPORT ById(generatedReportId=${id}) not found in database`);
      if (report.STATUS !== PENDING) throw new NackError('Failed requirement.');
      const config = ctx.store.getRow('REPORT_CONFIG', report.REPORT_CONFIG_ID);
      if (!config)
        throw new NackError(
          `REPORT_CONFIG ById(id=${report.REPORT_CONFIG_ID}) not found in database`,
        );
      if (!rowsOf(ctx, 'REPORT_CONFIG_DESTINATION_MAP', config.ID).length) {
        throw new NackError('Unable to dispatch report with no destinations configured');
      }
      if (ctx.validate) return {};
      let status: string;
      try {
        status = dispatch(ctx, report, config, userOf(ctx));
      } catch (error) {
        if (error instanceof NotImplemented) {
          throw NackError.standard('INTERNAL_ERROR', 'An operation is not implemented.');
        }
        throw error;
      }
      modify(ctx, 'GENERATED_REPORT', id, { STATUS: status });
      return {};
    },
    // EH :465-476. The file it names isn't deleted: the mock stored none.
    EVENT_GENERATED_REPORT_DELETE: (details, ctx) => {
      if (ctx.validate) return {};
      remove(ctx, 'GENERATED_REPORT', details.GENERATED_REPORT_ID);
      return {};
    },
  };

  // --- request servers --------------------------------------------------------------

  // REPORT_CONFIG_DETAILS (RR :112-152): the configuration flattened, with
  // what's under it; nothing for an unknown id.
  const configDetails: RequestReplyDef['resolver'] = ({ request, store }) => {
    const config = store.getRow('REPORT_CONFIG', request.ID);
    if (!config) return [];
    const under = (table: string) => rowsOf({ store }, table, config.ID);
    return [
      {
        DATA_SOURCES: under('REPORT_DATASOURCE').map(
          ({ TRANSFORMER_CONFIGURATION, ...source }) => ({
            TRANSFORMER_CONFIGURATION: readTransformer(TRANSFORMER_CONFIGURATION),
            ...source,
          }),
        ),
        DESTINATION_IDS: under('REPORT_CONFIG_DESTINATION_MAP').map((map) => map.DESTINATION_ID),
        SCHEDULES: under('REPORT_CONFIG_CRON_RULE_MAP').flatMap((map) => {
          const rule = store.getRow('CRON_RULE', map.CRON_RULE_NAME);
          return rule
            ? [{ CRON_EXPRESSION: rule.CRON_EXPRESSION ?? '', TIME_ZONE: rule.TIME_ZONE ?? '' }]
            : [];
        }),
        PARAMETERS: under('REPORT_PARAMETER'),
        ...config,
      },
    ];
  };

  // REPORT_PARAMETER_DETAILS (RR :156-190): the parameters, and a JSON schema
  // for their values: a string each, or the type of the datasource field a
  // linked one names (dropped when there's none).
  const parameterDetails: RequestReplyDef['resolver'] = ({ request, store, config }) => {
    const parameters = rowsOf({ store }, 'REPORT_PARAMETER', request.ID);
    const sources = new Map(
      rowsOf({ store }, 'REPORT_DATASOURCE', request.ID).map((row) => [row.KEY, row]),
    );
    // A LinkedHashMap: in the parameters' order (an object would put
    // integer-like names first; `required` keeps it).
    const properties = new Map<string, Row>();
    for (const parameter of parameters) {
      if (parameter.REPORT_DATASOURCE === null || parameter.REPORT_DATASOURCE === undefined) {
        properties.set(parameter.PARAMETER_NAME, { type: 'string' });
        continue;
      }
      const source = sources.get(parameter.REPORT_DATASOURCE);
      if (source?.INPUT_TYPE !== 'REQ_REP') continue;
      const replyProperties = requestServerReplyProperties(source.NAME, config, store);
      if (!replyProperties) {
        throw new Error(
          `Unable to resolve client for feature ${source.NAME}, unable to generate report`,
        );
      }
      const field = String(parameter.REFERENCE_FIELD);
      if (Object.hasOwn(replyProperties, field)) {
        properties.set(parameter.PARAMETER_NAME, replyProperties[field]);
      }
    }
    return [
      {
        PARAMETERS: parameters,
        SCHEMA: {
          properties: Object.fromEntries(properties),
          required: [...properties.keys()],
          type: 'object',
        },
      },
    ];
  };

  return {
    eventHandlers,
    eventSchemas: SCHEMAS,
    requestReplies: {
      // RR :90-110: every request server the cluster lists, or the sysdef's.
      ALL_REPORT_DATASOURCES: {
        requestFields: [],
        resolver: ({ config }) =>
          listResources(config)
            .filter((resource) => resource.RESOURCE_TYPE === 'REQUEST_SERVER')
            .map((resource) => resource.RESOURCE_NAME)
            .filter((name) => !options.datasources?.length || options.datasources.includes(name))
            .map((DATASOURCE_NAME) => ({ DATASOURCE_NAME })),
        metadataRows: [{ DATASOURCE_NAME: '' }],
      },
      // RR :78-88: the reporting process's pipeline scripts define none.
      REPORTING_DATA_PIPELINES: {
        requestFields: [],
        resolver: () => [],
        fieldTypes: { NAME: 'STRING' },
      },
      REPORT_CONFIG_DETAILS: {
        requestFields: ['ID'],
        resolver: configDetails,
        metadataRows: [
          {
            DATA_SOURCES: [],
            DESTINATION_IDS: [],
            PARAMETERS: [],
            AUTO_DISPATCH: false,
            CREATED_BY: '',
            CREATED_ON: 0,
            DESCRIPTION: '',
            DOCUMENT_TEMPLATE_ID: '',
            FILE_NAME: '',
            ID: '',
            LAST_UPDATED_AT: 0,
            LAST_UPDATED_BY: '',
            NAME: '',
            OUTPUT_DIRECTORY: '',
            OUTPUT_FORMAT: 'CSV',
            SCHEDULES: [],
          },
        ],
      },
      REPORT_PARAMETER_DETAILS: {
        requestFields: ['ID'],
        resolver: parameterDetails,
        metadataRows: [{ PARAMETERS: [], SCHEMA: {} }],
      },
      // RR :192-198: the JVM's zone ids.
      VALID_TIME_ZONE: {
        requestFields: [],
        resolver: () => JVM_ZONE_IDS.map((ZONE_ID) => ({ ZONE_ID })),
        fieldTypes: { ZONE_ID: 'STRING' },
      },
      // RR :53-65: getRangeFromEnd's third argument is the number of key
      // fields, not a limit, so every report, newest first, up to MAX_ROWS.
      GENERATED_REPORTS: {
        source: 'GENERATED_REPORT_VIEW',
        requestFields: [],
        order: (a, b) => compareRecordIds(b.TIMESTAMP, a.TIMESTAMP),
      },
    },
  };
}

// The destination events (EH :300-389), notify ones over NOTIFY_DESTINATION
// and TOPIC, "SMTP" ones over SFTP_DESTINATION and OUTPUT_DIRECTORY.
function destinationHandlers(
  name: 'NOTIFY' | 'SMTP',
  table: string,
  column: string,
): Record<string, EventHandler> {
  const type = name === 'NOTIFY' ? 'NOTIFY' : 'SFTP';
  return {
    [`EVENT_${name}_DESTINATION_INSERT`]: (details, ctx) => {
      if (ctx.validate) return {};
      const destination = insert(ctx, 'REPORT_DESTINATION', { NAME: details.NAME, TYPE: type });
      insert(ctx, table, { DESTINATION_ID: destination.ID, [column]: details[column] });
      return {};
    },
    // Both write TYPE NOTIFY: SMTP_DESTINATION_UPDATE re-types an SFTP
    // destination (EH :371).
    [`EVENT_${name}_DESTINATION_UPDATE`]: (details, ctx) => {
      if (ctx.validate) return {};
      modify(ctx, 'REPORT_DESTINATION', details.ID, { NAME: details.NAME, TYPE: 'NOTIFY' });
      modify(ctx, table, details.ID, { [column]: details[column] });
      return {};
    },
    // The configurations' maps to it are left.
    [`EVENT_${name}_DESTINATION_DELETE`]: (details, ctx) => {
      if (ctx.validate) return {};
      remove(ctx, 'REPORT_DESTINATION', details.ID);
      remove(ctx, table, details.ID);
      return {};
    },
  };
}