import crypto from 'node:crypto';
import type { Predicate } from '../db/criteria.ts';
import type { GenesisMessage, Row, SocketLike } from '../types.ts';

const WS_OPEN = 1;

// One row of a subscription's view (see Subscription.pending / held).
export interface ViewRow {
  // pending: the row as a page will send it, stamped as an INSERT. held: the
  // values the client holds, without DETAILS — what a MODIFY is diffed
  // against.
  row: Row;
  // The DETAILS.ROW_REF it is sent under.
  rowRef: unknown;
  // The stored (or view) row it was built from: what the snapshot order
  // sorts and what a waiting row is placed by.
  source: Row;
  // The TIMESTAMP of the version the client holds (the RECORD_ID when the
  // row has none). A moving view evicts the held row with the oldest one, as
  // MovingViewClientIndex's timestampIndex does.
  timestamp: unknown;
}

export interface Subscription {
  resourceName: string;
  // Computes the DETAILS.ROW_REF value for a pushed row (its RECORD_ID as a
  // string; in legacy mode the pk value or joined composite key) — see
  // rowRefFor in protocol/rowUpdate.ts.
  getRowRef: (row: Row) => unknown;
  predicate: Predicate;
  // SEQUENCE_ID of the last QUERY_UPDATE sent on this subscription (the
  // snapshot is 1); every later page or live push takes the next one.
  sequenceId: number;
  // The client's view, tracked like the real dataserver's client index (set
  // on 'gsf'-fidelity subscriptions; legacy ones have none). Every row is in
  // at most one of:
  // - pending: matched but not sent yet — past this page, or past MAX_VIEW —
  //   in delivery order, stamped and permission-stripped: the rows the real
  //   dataserver's next read would find past its read position. MORE_ROWS
  //   drains it, maxRows at a time, while the view has room.
  // - held: sent to the client (on a page, or pushed as an INSERT), with the
  //   values it was last sent. Only these get MODIFY and DELETE pushes.
  // Both are keyed by String(ROW_REF) (see dataLogon.ts pendingKey).
  pending?: Map<string, ViewRow>;
  held?: Map<string, ViewRow>;
  maxRows?: number;
  maxView?: number;
  // DETAILS.MOVING_VIEW (GSF's default: true). A new row for a full view is
  // still pushed, with a DELETE of the oldest held row in the same frame;
  // without it, a full view takes no new rows.
  movingView?: boolean;
  // The snapshot order, over source rows, and the source of the last row a
  // page delivered: the real dataserver's read position. A row the client
  // doesn't hold waits for a page only when it sorts after it.
  order?: (a: Row, b: Row) => number;
  cursor?: Row;
  // The columns this subscription's rows carry (the query's fields block or
  // declared columns, narrowed by DETAILS.FIELDS); undefined: rows as stored.
  fieldNames?: string[];
  // Turns a row into its wire form (BigDecimals as scaled strings); unset
  // when the query has nothing to encode, and under fidelity: 'legacy'.
  encode?: (row: Row) => Row;
}

// What a verbose HEARTBEAT_PONG reports per subscription (genesis-router
// QuerySubscription.kt / DataserverResource.kt @v8.15.29), all epoch millis.
interface SubscriptionStats {
  createdAt: number;
  lastReceived: number;
  lastSent: number;
  dataLogon: Row;
}

// Wraps one WebSocket client connection: auth state and its live DATA_LOGON
// subscriptions (keyed by SOURCE_REF, since that's the client's sole correlation
// key — see README pitfall #1).
export class Connection {
  readonly ws: SocketLike;
  readonly id = crypto.randomUUID();
  authenticated = false;
  sessionAuthToken: string | undefined;
  refreshAuthToken: string | undefined;
  sessionId: string | undefined;
  userName: string | undefined;
  profile: string[] = [];
  permission: string[] = [];
  readonly subscriptions = new Map<string, Subscription>();
  // The sessionAuthToken cookie the socket was opened with, under
  // auth.cookieAuthentication; it authenticates every message (router.ts).
  cookieSessionAuthToken: string | undefined;
  // The upgrade request's HOSTNAME header and the client's address, which
  // GSF's router gives a login as DETAILS.HOSTNAME (auth.ts routerHostname).
  hostnameHeader: string | undefined;
  remoteAddress: string | undefined;
  // A verbose HEARTBEAT_PONG's CREATED_AT, LAST_RECEIVED and LAST_SENT, as
  // genesis-router's ClientHandler keeps them (ClientHandler.kt:66-68,
  // 100-113 @v8.15.29): 0 until a message other than a heartbeat goes either
  // way, since pings and pongs bypass the client handler.
  readonly createdAt = Date.now();
  lastReceived = 0;
  lastSent = 0;
  private readonly subscriptionStats = new WeakMap<Subscription, SubscriptionStats>();

  constructor(ws: SocketLike) {
    this.ws = ws;
  }

  // Every response must echo the request's SOURCE_REF verbatim, or the client's
  // pending promise for that call never resolves (silent hang, no error).
  // `track: false` for what the real router writes without its client
  // handler (pongs, the cookie router's refusals): it leaves LAST_SENT alone.
  send(payload: Row, track = true): void {
    if (this.ws.readyState !== WS_OPEN) return;
    this.ws.send(JSON.stringify(payload));
    if (!track) return;
    const now = Date.now();
    this.lastSent = now;
    const subscription =
      typeof payload.SOURCE_REF === 'string'
        ? this.subscriptions.get(payload.SOURCE_REF)
        : undefined;
    const stats = subscription && this.subscriptionStats.get(subscription);
    if (stats) stats.lastSent = now;
  }

  reply(requestMessage: GenesisMessage, payload: Row, track = true): void {
    this.send({ ...payload, SOURCE_REF: requestMessage.SOURCE_REF }, track);
  }

  // Called for every routed message but heartbeats.
  noteReceived(message: GenesisMessage): void {
    const now = Date.now();
    this.lastReceived = now;
    const subscription = this.subscriptions.get(message.SOURCE_REF ?? '');
    const stats = subscription && this.subscriptionStats.get(subscription);
    if (stats) stats.lastReceived = now;
  }

  // Called after a DATA_LOGON was handled: starts the stats of the
  // subscription it opened, if it opened one.
  noteLogon(message: GenesisMessage): void {
    const subscription = this.subscriptions.get(message.SOURCE_REF ?? '');
    if (!subscription || this.subscriptionStats.has(subscription)) return;
    const now = Date.now();
    this.subscriptionStats.set(subscription, {
      createdAt: now,
      lastReceived: now,
      lastSent: now,
      dataLogon: message.DETAILS ?? {},
    });
  }

  // A verbose HEARTBEAT_PONG's RESOURCES: the live subscriptions by resource
  // name (FullDuplexClientHandler.kt:60-66 @v8.15.29), each a serialized
  // DataserverResource.
  dataserverResources(): Record<string, Row[]> {
    const resources: Record<string, Row[]> = {};
    for (const [sourceRef, subscription] of this.subscriptions) {
      const stats = this.subscriptionStats.get(subscription);
      if (!stats) continue;
      (resources[subscription.resourceName] ??= []).push({
        NAME: subscription.resourceName,
        CREATED_AT: stats.createdAt,
        LAST_RECEIVED: stats.lastReceived,
        LAST_SENT: stats.lastSent,
        SOURCE_REF: sourceRef,
        DATA_LOGON: stats.dataLogon,
      });
    }
    return resources;
  }

  subscribe(sourceRef: string, subscription: Subscription): void {
    this.subscriptions.set(sourceRef, subscription);
  }

  unsubscribe(sourceRef: string): void {
    this.subscriptions.delete(sourceRef);
  }
}