cloudly/ts/manager.baseos/classes.baseosmanager.ts

import * as plugins from '../plugins.js';
import type { Cloudly } from '../classes.cloudly.js';
import { logger } from '../logger.js';
import {
  BaseOsNode,
  type IBaseOsDesiredState,
  type IBaseOsNodePublic,
  type IBaseOsRuntimeInfo,
} from './classes.baseosnode.js';

interface IBaseOsRegisterRequest {
  joinToken?: string;
  nodeToken?: string;
  status?: IBaseOsRuntimeInfo;
}

interface IBaseOsRegisterResponse {
  nodeId?: string;
  nodeToken?: string;
  accepted: boolean;
  message?: string;
}

interface IBaseOsHeartbeatRequest {
  nodeToken?: string;
  status?: IBaseOsRuntimeInfo;
}

interface IBaseOsHeartbeatResponse {
  accepted: boolean;
  message?: string;
  desiredState?: IBaseOsDesiredState;
}

interface IRequestGetBaseOsNodes {
  method: 'getBaseOsNodes';
  request: {
    identity: plugins.servezoneInterfaces.data.IIdentity;
  };
  response: {
    nodes: IBaseOsNodePublic[];
  };
}

export class CloudlyBaseOsManager {
  public cloudlyRef: Cloudly;
  public typedRouter = new plugins.typedrequest.TypedRouter();

  public get db() {
    return this.cloudlyRef.mongodbConnector.smartdataDb;
  }

  public CBaseOsNode = plugins.smartdata.setDefaultManagerForDoc(this, BaseOsNode);

  constructor(cloudlyRefArg: Cloudly) {
    this.cloudlyRef = cloudlyRefArg;
    this.cloudlyRef.typedrouter.addTypedRouter(this.typedRouter);

    this.typedRouter.addTypedHandler(
      new plugins.typedrequest.TypedHandler<IRequestGetBaseOsNodes>(
        'getBaseOsNodes',
        async (requestDataArg) => {
          await plugins.smartguard.passGuardsOrReject(
            { identity: requestDataArg.identity },
            [this.cloudlyRef.authManager.adminIdentityGuard],
          );
          return {
            nodes: await this.getPublicNodes(),
          };
        },
      ),
    );
  }

  public async start() {
    logger.log('info', 'BaseOS manager started');
  }

  public async stop() {
    logger.log('info', 'BaseOS manager stopped');
  }

  public async handleRegisterHttpRequest(
    reqArg: plugins.typedserver.Request,
    resArg: plugins.typedserver.Response,
  ) {
    try {
      const requestData = await this.readJsonBody<IBaseOsRegisterRequest>(reqArg);
      const response = await this.registerNode(requestData);
      this.sendJson(resArg, 200, response);
    } catch (error) {
      this.sendJson(resArg, 400, {
        accepted: false,
        message: `BaseOS registration failed: ${(error as Error).message}`,
      } satisfies IBaseOsRegisterResponse);
    }
  }

  public async handleHeartbeatHttpRequest(
    reqArg: plugins.typedserver.Request,
    resArg: plugins.typedserver.Response,
  ) {
    try {
      const requestData = await this.readJsonBody<IBaseOsHeartbeatRequest>(reqArg);
      const response = await this.acceptHeartbeat(requestData);
      this.sendJson(resArg, 200, response);
    } catch (error) {
      this.sendJson(resArg, 400, {
        accepted: false,
        message: `BaseOS heartbeat failed: ${(error as Error).message}`,
      } satisfies IBaseOsHeartbeatResponse);
    }
  }

  public async registerNode(
    requestDataArg: IBaseOsRegisterRequest,
  ): Promise<IBaseOsRegisterResponse> {
    if (!this.isRuntimeInfo(requestDataArg.status)) {
      return {
        accepted: false,
        message: 'BaseOS runtime status is missing or invalid',
      };
    }

    if (requestDataArg.nodeToken) {
      const existingNode = await this.getNodeByToken(requestDataArg.nodeToken);
      if (existingNode) {
        await this.updateNodeRuntimeInfo(existingNode, requestDataArg.status, true);
        return {
          accepted: true,
          nodeId: existingNode.id,
        };
      }
    }

    const configuredJoinToken = await this.cloudlyRef.settingsManager.getSetting('baseosJoinToken');
    if (!configuredJoinToken) {
      return {
        accepted: false,
        message: 'BaseOS join token is not configured in Cloudly settings',
      };
    }

    if (!requestDataArg.joinToken || requestDataArg.joinToken !== configuredJoinToken) {
      return {
        accepted: false,
        message: 'BaseOS join token is invalid',
      };
    }

    const nodeToken = await this.cloudlyRef.authManager.createNewSecureToken();
    const node = await this.upsertNode(requestDataArg.status, nodeToken);
    return {
      accepted: true,
      nodeId: node.id,
      nodeToken,
    };
  }

  public async acceptHeartbeat(
    requestDataArg: IBaseOsHeartbeatRequest,
  ): Promise<IBaseOsHeartbeatResponse> {
    if (!requestDataArg.nodeToken) {
      return {
        accepted: false,
        message: 'BaseOS node token is missing',
      };
    }
    if (!this.isRuntimeInfo(requestDataArg.status)) {
      return {
        accepted: false,
        message: 'BaseOS runtime status is missing or invalid',
      };
    }

    const node = await this.getNodeByToken(requestDataArg.nodeToken);
    if (!node) {
      return {
        accepted: false,
        message: 'BaseOS node token is invalid',
      };
    }

    await this.updateNodeRuntimeInfo(node, requestDataArg.status, true);
    return {
      accepted: true,
      desiredState: node.data.desiredState || {},
    };
  }

  public async getPublicNodes(): Promise<IBaseOsNodePublic[]> {
    const nodes = await this.CBaseOsNode.getInstances({});
    return nodes.map((nodeArg) => nodeArg.toPublicNode());
  }

  private async upsertNode(statusArg: IBaseOsRuntimeInfo, nodeTokenArg: string) {
    const now = Date.now();
    let node = await this.CBaseOsNode.getInstance({
      id: statusArg.nodeId,
    }).catch(() => null);

    if (!node) {
      node = new this.CBaseOsNode({
        id: statusArg.nodeId,
        nodeToken: nodeTokenArg,
        data: {
          runtimeInfo: statusArg,
          createdAt: now,
          updatedAt: now,
          lastHeartbeatAt: now,
        },
      });
    } else {
      node.nodeToken = nodeTokenArg;
      node.data = {
        ...node.data,
        runtimeInfo: statusArg,
        updatedAt: now,
        lastHeartbeatAt: now,
      };
    }

    await node.save();
    return node;
  }

  private async updateNodeRuntimeInfo(
    nodeArg: BaseOsNode,
    statusArg: IBaseOsRuntimeInfo,
    heartbeatArg = false,
  ) {
    nodeArg.data = {
      ...nodeArg.data,
      runtimeInfo: statusArg,
      updatedAt: Date.now(),
      ...(heartbeatArg ? { lastHeartbeatAt: Date.now() } : {}),
    };
    await nodeArg.save();
  }

  private async getNodeByToken(nodeTokenArg: string) {
    const nodes = await this.CBaseOsNode.getInstances({
      nodeToken: nodeTokenArg,
    });
    return nodes[0] || null;
  }

  private isRuntimeInfo(valueArg: unknown): valueArg is IBaseOsRuntimeInfo {
    if (!valueArg || typeof valueArg !== 'object') {
      return false;
    }
    const runtimeInfo = valueArg as Partial<IBaseOsRuntimeInfo>;
    return runtimeInfo.runtime === 'baseos'
      && typeof runtimeInfo.nodeId === 'string'
      && runtimeInfo.nodeId.length > 0
      && typeof runtimeInfo.checkedAt === 'number';
  }

  private async readJsonBody<T>(reqArg: plugins.typedserver.Request): Promise<T> {
    const chunks: Buffer[] = [];
    for await (const chunk of reqArg as any) {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
    }
    const bodyString = Buffer.concat(chunks).toString('utf8').trim();
    return bodyString ? JSON.parse(bodyString) as T : {} as T;
  }

  private sendJson(
    resArg: plugins.typedserver.Response,
    statusCodeArg: number,
    bodyArg: object,
  ) {
    resArg.status(statusCodeArg);
    resArg.setHeader('Content-Type', 'application/json');
    resArg.end(JSON.stringify(bodyArg));
  }
}