import type { AuditContext } from '../db/audit.ts';
import type { Store } from '../db/store.ts';
import type { Connection } from '../protocol/connection.ts';
import { NackError, nackPayload, warningPayload } from '../protocol/errors.ts';
import { isLegacyFidelity } from '../protocol/fidelity.ts';
import { ERROR_CODE, MESSAGE_TYPE } from '../protocol/messageTypes.ts';
import { resourceNotMappedNack } from '../protocol/msgNack.ts';
import type {
  BroadcastFn,
  BroadcastTableChangeFn,
  EventHandler,
  GenesisMessage,
  MockServerConfig,
} from '../types.ts';
import { crudEventHandler } from './crudEvents.ts';
import { eventDetails, validateEventMessage } from './eventValidation.ts';

export interface CommitEventCtx {
  broadcast: BroadcastFn;
  broadcastTableChange: BroadcastTableChangeFn;
}

// The handler serving an event: an eventHandlers entry, else the generic CRUD
// handler a table registers through TableDef.events, else
// defaultEventHandler.
function resolveEventHandler(
  eventName: string,
  config: MockServerConfig,
): EventHandler | undefined {
  return (
    config.eventHandlers?.[eventName] ??
    crudEventHandler(eventName, config) ??
    config.defaultEventHandler
  );
}

// One event's transaction: the store's (Store.begin), plus the live pushes
// the handler asks for, held back until the event has committed and acked —
// a client must never see a change that is then rolled back, and on the real
// server the dataserver's update reaches the client after the EVENT_ACK (the
// recordings' order), so a client can't count on its grid holding the change
// when the ACK arrives. Once the transaction is over, the broadcast
// functions push straight away again, so a handler can keep them for later
// (the showcase's price ticker does).
interface EventTransaction {
  broadcast: BroadcastFn;
  broadcastTableChange: BroadcastTableChangeFn;
  // Holds a callback for commit() (EventHandlerCtx.afterCommit).
  afterCommit(callback: () => void): void;
  // Keeps the writes and runs the afterCommit callbacks; the held pushes,
  // the callbacks' too, wait for flush().
  commit(): void;
  // Sends the held pushes. The event is committed whatever a push does: a
  // failing one is logged, and the rest still go out.
  flush(): void;
  // Undoes the writes and drops the held pushes. Does nothing once the
  // transaction is over.
  rollback(): void;
}

// Runs each callback; one that fails is logged, and the rest still run.
function runAll(callbacks: Array<() => void>, what: string): void {
  for (const callback of callbacks.splice(0)) {
    try {
      callback();
    } catch (error) {
      console.error(`[mock-server] Error ${what}:`, error);
    }
  }
}

function openTransaction(store: Store, ctx: CommitEventCtx, audit: AuditContext): EventTransaction {
  const pending: Array<() => void> = [];
  const committed: Array<() => void> = [];
  // open: the handler runs; committed: its writes kept, its pushes waiting
  // for flush(); over: flushed or rolled back.
  let state: 'open' | 'committed' | 'over' = 'open';
  let rolledBack = false;
  store.begin(audit);
  const hold = (push: () => void) => {
    if (state === 'over') push();
    else pending.push(push);
  };
  return {
    broadcast: (resourceName, operation, row) =>
      hold(() => ctx.broadcast(resourceName, operation, row)),
    broadcastTableChange: (tableName, operation, pk) =>
      hold(() => ctx.broadcastTableChange(tableName, operation, pk)),
    // Asked for once the handler is done (from a callback, or kept for
    // later): at once when the event committed, never when it rolled back.
    afterCommit: (callback) => {
      if (state === 'open') committed.push(callback);
      else if (!rolledBack) runAll([callback], 'after a commit');
    },
    commit() {
      // The audit rows the writes added go live with them.
      for (const { tableName, key } of store.takeAuditWrites()) {
        pending.push(() => ctx.broadcastTableChange(tableName, 'INSERT', key));
      }
      store.commit();
      state = 'committed';
      runAll(committed, 'after a commit');
    },
    flush() {
      state = 'over';
      runAll(pending, 'pushing a committed change');
    },
    rollback() {
      if (state !== 'open') return;
      state = 'over';
      rolledBack = true;
      pending.length = 0;
      committed.length = 0;
      store.rollback();
    },
  };
}

// What GSF audits an event's writes with (TransactionalEventHandlerContainer:
// entityDb.audited(message.userName, message.messageType, message.reason)).
function auditContext(message: GenesisMessage, connection: Connection): AuditContext {
  return {
    eventType: message.MESSAGE_TYPE,
    user: connection.userName ?? '',
    text: typeof message.REASON === 'string' ? message.REASON : null,
  };
}

// Commit events (EVENT_<ENTITY>_INSERT/_AMEND/_DELETE, or whatever suffix
// convention your project uses) are looked up by exact MESSAGE_TYPE against
// config.eventHandlers, then the tables' generic CRUD events. A handler
// receives (details, ctx) and either returns { generated?, warnings? } or
// throws NackError(errors) on failure.
//
// A handler stands for a GSF event handler's onValidate and onCommit at once,
// and the engine gives its outcome the commit semantics of GSF's
// TransactionalEventHandlerContainer (genesis-pal-eventhandler, GSF v8.15.29,
// transactionalEventFlow/defaultEventFlow; the typed handlers'
// SyncValidatingEventHandler.process is the same):
//
//   VALIDATE: true       → onValidate only: its reply, nothing committed
//   onValidate ACKs      → onCommit runs: its reply
//   onValidate NACKs with warnings only (`result.error.isEmpty()`)
//     and IGNORE_WARNINGS: true → onCommit runs: its reply (an ACK carries
//                          no WARNING)
//   any other NACK       → that NACK, nothing committed
//
// So the handler runs in a transaction (openTransaction), and the event
// commits only when it acks: returned warnings without IGNORE_WARNINGS, a
// thrown NackError (errors, with or without warnings), any other exception,
// and VALIDATE all roll back what the handler wrote, and drop the live pushes
// it asked for. Returned warnings with IGNORE_WARNINGS commit and ack with
// the handler's GENERATED. A thrown NackError always NACKs, warnings-only
// ones included: the handler stopped there, before its writes (a GSF
// exception is always an error). A VALIDATE ack has GENERATED: [], as the
// recorded ones do: whatever the handler generated was rolled back.
export function handleCommitEvent(
  message: GenesisMessage,
  connection: Connection,
  config: MockServerConfig,
  store: Store,
  ctx: CommitEventCtx,
): void {
  const eventName = message.MESSAGE_TYPE;
  // defaultEventHandler catches anything unregistered — lets projects with
  // dozens of commit events adopt incrementally (e.g. `() => ({})` to ack
  // everything unhandled) instead of enumerating ack-only handlers.
  const handler = resolveEventHandler(eventName, config);

  // Replies use the GENERIC EVENT_ACK/EVENT_NACK — correlation is purely via
  // SOURCE_REF. See the note on MESSAGE_TYPE in protocol/messageTypes.ts:
  // real consumers (foundation-comms' kv.service, entity-management) match
  // these names exactly, so a name-suffixed <EVENT>_ACK reads as failure and
  // a <EVENT>_NACK passes as silent success.
  // An event nothing handles never reaches an event handler on the real
  // server: the router rejects it by name with MSG_NACK 404.
  if (!handler) {
    connection.reply(
      message,
      isLegacyFidelity(config)
        ? {
            MESSAGE_TYPE: MESSAGE_TYPE.EVENT_NACK,
            ERROR: [
              {
                CODE: ERROR_CODE.UNKNOWN_RESOURCE,
                TEXT: `No event handler registered for ${eventName}`,
              },
            ],
          }
        : resourceNotMappedNack(eventName),
    );
    return;
  }

  // Before any handler runs, the message is checked against the event's
  // declared schema (eventValidation.ts), as GSF's decoder does; each failure
  // is a FieldError in one EVENT_NACK. A handler's failure is an EVENT_NACK
  // with GSF's error items (see protocol/errors.ts): a thrown NackError gives
  // its own, anything else is StandardError INTERNAL_ERROR.
  // Legacy fidelity keeps what this engine did before: no schema check, no
  // transaction (a handler's writes stay whatever it replies), and warnings
  // always NACK, IGNORE_WARNINGS or not.
  const legacy = isLegacyFidelity(config);
  const validate = message.VALIDATE === true;
  const ignoreWarnings = !legacy && message.IGNORE_WARNINGS === true;
  let transaction: EventTransaction | undefined;
  try {
    // Inside the try: a store transaction someone else left open fails this
    // event (INTERNAL_ERROR) instead of leaving it unanswered.
    if (!legacy) transaction = openTransaction(store, ctx, auditContext(message, connection));
    const schemaErrors = legacy ? [] : validateEventMessage(message, connection, config, store);
    if (schemaErrors.length > 0) throw new NackError(schemaErrors);
    const details = legacy ? (message.DETAILS ?? {}) : eventDetails(message, config, store);
    const result =
      handler(details, {
        connection,
        store,
        message,
        config,
        broadcast: transaction?.broadcast ?? ctx.broadcast,
        broadcastTableChange: transaction?.broadcastTableChange ?? ctx.broadcastTableChange,
        validate,
        ignoreWarnings,
        afterCommit: transaction?.afterCommit ?? ((callback) => callback()),
      }) ?? {};
    if (result.warnings?.length && (validate || !ignoreWarnings)) {
      transaction?.rollback();
      connection.reply(message, {
        MESSAGE_TYPE: MESSAGE_TYPE.EVENT_NACK,
        ...warningPayload(result.warnings, legacy),
      });
      return;
    }
    const generated = validate && !legacy ? [] : (result.generated ?? []);
    const reply = result.reply ?? { MESSAGE_TYPE: MESSAGE_TYPE.EVENT_ACK, GENERATED: generated };
    // A reply that can't go on the wire (a BigInt, a cycle) fails the event
    // here, before anything is committed.
    JSON.stringify(reply);
    if (validate) transaction?.rollback();
    else transaction?.commit();
    try {
      connection.reply(message, reply);
    } finally {
      transaction?.flush();
    }
  } catch (error) {
    transaction?.rollback();
    connection.reply(message, {
      MESSAGE_TYPE: (error instanceof NackError && error.messageType) || MESSAGE_TYPE.EVENT_NACK,
      ...nackPayload(error, legacy),
    });
  }
}