foreign/node/src/wire/topic/update-topic.command.ts

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

import { serializeIdentifier, type Id } from '../identifier.utils.js';
import { dedupeOptions, serializeOptions, type OptionEntry } from '../options.utils.js';
import { HeaderValue } from '../message/header.utils.js';
import { deserializeVoidResponse } from '../../client/client.utils.js';
import { wrapCommand } from '../command.utils.js';
import { COMMAND_CODE } from '../command.code.js';
import {
  type CompressionAlgorithm as CompressionAlgorithmT,
  CompressionAlgorithm,
  compressionAlgorithmName,
  isValidCompressionAlgorithm
} from './topic.utils.js';


/**
 * Parameters for the update topic command.
 */
export type UpdateTopic = {
  /** Stream identifier (ID or name) */
  streamId: Id,
  /** Topic identifier (ID or name) */
  topicId: Id,
  /** New topic name (1-255 bytes) */
  name: string,
  /** Compression algorithm (None or Gzip) */
  compressionAlgorithm?: CompressionAlgorithmT,
  /**
   * Message expiry time in microseconds. `0n` is not "unlimited": it omits the
   * key, so the topic keeps whatever expiry it currently has. Resetting a key
   * back to the server default is not expressible on an update.
   */
  messageExpiry?: bigint,
  /**
   * Maximum topic size in bytes. `0n` is not "unlimited": it omits the key, so
   * the topic keeps whatever cap it currently has. Resetting a key back to the
   * server default is not expressible on an update.
   */
  maxTopicSize?: bigint,
  /**
   * Option keys with no field of their own. The server refuses any key an update
   * may not change, by name; a key left out keeps its current value.
   */
  options?: OptionEntry[]
};

/**
 * Update topic command definition.
 * Updates a topic's configuration.
 */
export const UPDATE_TOPIC = {
  code: COMMAND_CODE.UpdateTopic,

  serialize: ({
    streamId,
    topicId,
    name,
    compressionAlgorithm = CompressionAlgorithm.None,
    messageExpiry = 0n,
    maxTopicSize = 0n,
    options: extraOptions = []
  }: UpdateTopic) => {
    const streamIdentifier = serializeIdentifier(streamId);
    const topicIdentifier = serializeIdentifier(topicId);
    const bName = Buffer.from(name)

    if (bName.length < 1 || bName.length > 255)
      throw new Error('Topic name should be between 1 and 255 bytes');
    if(!isValidCompressionAlgorithm(compressionAlgorithm))
      throw new Error(`updateTopic: invalid compressionAlgorithm (${compressionAlgorithm})`);

    // Settings ride the options block. A default value means the caller did not
    // set the key, so it is omitted and the server leaves the current value be.
    // Caller keys first so a typed field below overwrites one of them.
    const options: OptionEntry[] = [...extraOptions];
    if (compressionAlgorithm !== CompressionAlgorithm.None)
      options.push({
        key: 'compression_algorithm',
        value: HeaderValue.String(compressionAlgorithmName(compressionAlgorithm))
      });
    if (messageExpiry !== 0n)
      options.push({
        key: 'message_expiry',
        value: HeaderValue.Uint64(messageExpiry)
      });
    if (maxTopicSize !== 0n)
      options.push({
        key: 'max_topic_size',
        value: HeaderValue.Uint64(maxTopicSize)
      });

    const b = Buffer.allocUnsafe(1);
    b.writeUInt8(bName.length, 0);

    return Buffer.concat([
      streamIdentifier,
      topicIdentifier,
      b,
      bName,
      serializeOptions(dedupeOptions(options)),
    ]);
  },

  deserialize: deserializeVoidResponse
};


/**
 * Executable update topic command function.
 */
export const updateTopic = wrapCommand<UpdateTopic, boolean>(UPDATE_TOPIC);