import * as plugins from '../../plugins.js';
import type { OpsServer } from '../classes.opsserver.js';
import * as interfaces from '../../../ts_interfaces/index.js';
import { requireOpsAuth } from '../helpers/auth.js';

const REALTIME_TAG_PREFIX = 'dcrouterOpsRealtime:user:';

export const getOpsRealtimeUserTag = (userIdArg: string): string => {
  return REALTIME_TAG_PREFIX + encodeURIComponent(userIdArg);
};

const createInitialRevisions = (): interfaces.requests.IOpsRealtimeRevisionMap => {
  return Object.fromEntries(
    interfaces.requests.opsRealtimeResources.map((resource) => [resource, 1]),
  ) as unknown as interfaces.requests.IOpsRealtimeRevisionMap;
};

export class OpsRealtimeRevisionClock {
  public readonly serverEpoch: string;
  private revisions = createInitialRevisions();

  constructor(serverEpochArg = plugins.uuid.v4()) {
    this.serverEpoch = serverEpochArg;
  }

  public getRevisions(): interfaces.requests.IOpsRealtimeRevisionMap {
    return { ...this.revisions };
  }

  public invalidate(
    resourceArg: interfaces.requests.TOpsRealtimeResource,
    optionsArg: {
      changedIds?: string[];
      reason?: string;
      timestamp?: number;
    } = {},
  ): interfaces.requests.IOpsRealtimeInvalidation {
    const revision = this.revisions[resourceArg] + 1;
    this.revisions[resourceArg] = revision;
    return {
      resource: resourceArg,
      revision,
      changedIds: optionsArg.changedIds?.length
        ? Array.from(new Set(optionsArg.changedIds))
        : undefined,
      reason: optionsArg.reason,
      timestamp: optionsArg.timestamp ?? Date.now(),
    };
  }

  public getCatchup(
    knownServerEpochArg?: string,
    knownRevisionsArg: Partial<interfaces.requests.IOpsRealtimeRevisionMap> = {},
  ): interfaces.requests.IOpsRealtimeInvalidation[] {
    const epochChanged = knownServerEpochArg !== this.serverEpoch;
    return interfaces.requests.opsRealtimeResources
      .filter((resource) => epochChanged || (knownRevisionsArg[resource] ?? 0) < this.revisions[resource])
      .map((resource) => ({
        resource,
        revision: this.revisions[resource],
        timestamp: Date.now(),
        reason: epochChanged ? 'initial-catchup' : 'revision-catchup',
      }));
  }
}

interface IWebSocketPeerForRealtime {
  id: string;
  tags: Set<string>;
}

interface IRealtimeInvalidationOptions {
  changedIds?: string[];
  reason?: string;
}

export class RealtimeHandler {
  public typedrouter = new plugins.typedrequest.TypedRouter();
  public readonly revisionClock = new OpsRealtimeRevisionClock();

  private subscribedUserIds = new Set<string>();
  private pendingInvalidations = new Map<
    interfaces.requests.TOpsRealtimeResource,
    interfaces.requests.IOpsRealtimeInvalidation
  >();
  private pushScheduled = false;
  private stopped = false;

  constructor(private opsServerRef: OpsServer) {
    this.opsServerRef.viewRouter.addTypedRouter(this.typedrouter);
    this.registerHandlers();
  }

  public invalidate(
    resourcesArg:
      | interfaces.requests.TOpsRealtimeResource
      | interfaces.requests.TOpsRealtimeResource[],
    optionsArg: IRealtimeInvalidationOptions = {},
  ): void {
    if (this.stopped) return;
    const resources = Array.isArray(resourcesArg) ? resourcesArg : [resourcesArg];
    for (const resource of new Set(resources)) {
      const invalidation = this.revisionClock.invalidate(resource, optionsArg);
      const existing = this.pendingInvalidations.get(resource);
      this.pendingInvalidations.set(resource, {
        ...invalidation,
        changedIds: Array.from(new Set([
          ...(existing?.changedIds || []),
          ...(invalidation.changedIds || []),
        ])),
      });
    }
    this.schedulePush();
  }

  public async getAuthenticatedConnections(): Promise<
    plugins.typedsocket.ISmartServeConnectionWrapper[]
  > {
    const typedsocket = this.opsServerRef.server?.typedserver?.typedsocket;
    if (!typedsocket) return [];

    const connections = new Map<string, plugins.typedsocket.ISmartServeConnectionWrapper>();
    for (const userId of Array.from(this.subscribedUserIds)) {
      const userConnections = await typedsocket.findAllTargetConnectionsByTag(
        getOpsRealtimeUserTag(userId),
      );
      if (userConnections.length === 0) {
        this.subscribedUserIds.delete(userId);
        continue;
      }
      for (const connection of userConnections) {
        connections.set(connection.peer.id, connection);
      }
    }
    return Array.from(connections.values());
  }

  public cleanup(): void {
    this.stopped = true;
    this.pendingInvalidations.clear();
    this.subscribedUserIds.clear();
  }

  private registerHandlers(): void {
    this.typedrouter.addTypedHandler(
      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_SubscribeOpsRealtime>(
        'subscribeOpsRealtime',
        async (dataArg, toolsArg) => {
          const auth = await requireOpsAuth(this.opsServerRef, dataArg);
          if (auth.type !== 'identity') {
            throw new plugins.typedrequest.TypedResponseError(
              'Realtime UI subscriptions require a user identity',
            );
          }

          const peer = toolsArg?.localData?.peer as IWebSocketPeerForRealtime | undefined;
          if (!peer?.id || !peer.tags) {
            throw new plugins.typedrequest.TypedResponseError(
              'Realtime subscriptions require a WebSocket connection',
            );
          }

          for (const tag of Array.from(peer.tags)) {
            if (tag.startsWith(REALTIME_TAG_PREFIX)) {
              peer.tags.delete(tag);
            }
          }
          peer.tags.add(getOpsRealtimeUserTag(auth.userId));
          this.subscribedUserIds.add(auth.userId);

          return {
            serverEpoch: this.revisionClock.serverEpoch,
            revisions: this.revisionClock.getRevisions(),
            invalidations: this.revisionClock.getCatchup(
              dataArg.knownServerEpoch,
              dataArg.knownRevisions,
            ),
          };
        },
      ),
    );
  }

  private schedulePush(): void {
    if (this.pushScheduled) return;
    this.pushScheduled = true;
    queueMicrotask(() => {
      this.pushScheduled = false;
      void this.flushPushes();
    });
  }

  private async flushPushes(): Promise<void> {
    if (this.stopped || this.pendingInvalidations.size === 0) return;
    const invalidations = Array.from(this.pendingInvalidations.values());
    this.pendingInvalidations.clear();

    let connections: plugins.typedsocket.ISmartServeConnectionWrapper[];
    try {
      connections = await this.getAuthenticatedConnections();
    } catch {
      return;
    }

    const typedsocket = this.opsServerRef.server?.typedserver?.typedsocket;
    if (!typedsocket) return;

    await Promise.allSettled(
      connections.map(async (connection) => {
        const request =
          typedsocket.createTypedRequest<interfaces.requests.IReq_PushOpsRealtimeInvalidation>(
            'pushOpsRealtimeInvalidation',
            connection,
          );
        await request.fire({
          serverEpoch: this.revisionClock.serverEpoch,
          invalidations,
        });
      }),
    );
  }
}