Task Queue Management

/**
 * Client for worker versioning operations (Experimental)
 */
class TaskQueueClient extends BaseClient {
  /** Update build ID compatibility for a task queue */
  updateBuildIdCompatibility(
    taskQueue: string,
    operation: BuildIdOperation
  ): Promise<void>;

  /** Get build ID compatibility sets for a task queue */
  getBuildIdCompatability(taskQueue: string): Promise<WorkerBuildIdVersionSets>;

  /** Get worker task reachability information */
  getReachability(options: ReachabilityOptions): Promise<ReachabilityResponse>;

  /** Raw gRPC access to WorkflowService */
  readonly workflowService: WorkflowService;
}

import { TaskQueueClient } from "@temporalio/client";

const taskQueueClient = new TaskQueueClient();

// Add new build ID in new default set
await taskQueueClient.updateBuildIdCompatibility('my-task-queue', {
  operation: 'addNewIdInNewDefaultSet',
  buildId: 'build-v1.2.0',
});

// Add compatible build ID to existing set
await taskQueueClient.updateBuildIdCompatibility('my-task-queue', {
  operation: 'addNewCompatibleVersion',
  buildId: 'build-v1.2.1',
  existingCompatibleBuildId: 'build-v1.2.0',
});

// Get current build ID compatibility
const compatibility = await taskQueueClient.getBuildIdCompatability('my-task-queue');
console.log('Default set:', compatibility.defaultSet);

// Check reachability
const reachability = await taskQueueClient.getReachability(
  ['build-v1.2.0', 'build-v1.2.1'],
  ['my-task-queue']
);

/**
 * Operations for build ID management
 */
type BuildIdOperation =
  | AddNewIdInNewDefaultSet
  | AddNewCompatibleVersion
  | PromoteSetByBuildId
  | PromoteBuildIdWithinSet
  | MergeSets;

/**
 * Add new build ID in new default set
 */
interface AddNewIdInNewDefaultSet {
  operation: 'addNewIdInNewDefaultSet';
  buildId: string;
}

/**
 * Add compatible build ID version to existing set
 */
interface AddNewCompatibleVersion {
  operation: 'addNewCompatibleVersion';
  buildId: string;
  existingCompatibleBuildId: string;
  makeSetDefault?: boolean;
}

/**
 * Promote set containing build ID to default
 */
interface PromoteSetByBuildId {
  operation: 'promoteSetByBuildId';
  buildId: string;
}

/**
 * Promote build ID within its set to be the default
 */
interface PromoteBuildIdWithinSet {
  operation: 'promoteBuildIdWithinSet';
  buildId: string;
}

/**
 * Merge two sets of compatible build IDs
 */
interface MergeSets {
  operation: 'mergeSets';
  primaryBuildId: string;
  secondaryBuildId: string;
}

// Gradual rollout workflow
const taskQueue = 'production-queue';

// 1. Add new build ID in non-default set for testing
await taskQueueClient.updateBuildIdCompatibility(taskQueue, {
  operation: 'addNewIdInNewDefaultSet',
  buildId: 'v2.0.0',
});

// 2. Add compatible patch version
await taskQueueClient.updateBuildIdCompatibility(taskQueue, {
  operation: 'addNewCompatibleVersion',
  buildId: 'v2.0.1',
  existingCompatibleBuildId: 'v2.0.0',
});

// 3. Promote new version to default after validation
await taskQueueClient.updateBuildIdCompatibility(taskQueue, {
  operation: 'promoteSetByBuildId',
  buildId: 'v2.0.1',
});

/**
 * Sets of compatible build IDs for a task queue
 */
interface WorkerBuildIdVersionSets {
  /** Default build ID set (receives new tasks) */
  defaultSet?: BuildIdVersionSet;
  /** All build ID sets including non-default */
  allSets: BuildIdVersionSet[];
}

/**
 * Single set of compatible build IDs
 */
interface BuildIdVersionSet {
  /** Build IDs in this compatibility set */
  buildIds: string[];
  /** Default build ID within this set */
  defaultBuildId?: string;
}

/**
 * Stand-in for unversioned workers
 */
const UnversionedBuildId: unique symbol;

/**
 * Type for unversioned build ID symbol
 */
type UnversionedBuildIdType = typeof UnversionedBuildId;

/**
 * Options for reachability queries
 */
type ReachabilityOptions = {
  /** Include reachability for unversioned workers */
  includeUnversioned?: boolean;
};

/**
 * Options for querying build ID reachability
 */
interface ReachabilityOptions {
  /** List of build IDs to query (supports UnversionedBuildId symbol) */
  buildIds: (string | UnversionedBuildIdType)[];
  /** Task queues to check (optional) */
  taskQueues?: string[];
  /** Type of reachability to check (optional) */
  reachability?: ReachabilityType;
}

/**
 * Response from reachability queries
 */
interface ReachabilityResponse {
  /** Reachability information per build ID */
  buildIdReachability: BuildIdReachability[];
}

/**
 * Reachability info for specific build ID
 */
interface BuildIdReachability {
  /** Build ID or UnversionedBuildId symbol */
  buildId: string | typeof UnversionedBuildId;
  /** Task queues and their reachability info */
  taskQueueReachability: TaskQueueReachability[];
}

/**
 * Reachability info for a task queue
 */
interface TaskQueueReachability {
  /** Task queue name */
  taskQueue: string;
  /** Types of reachability */
  reachability: ReachabilityType[];
}

/**
 * Types of reachability
 */
const ReachabilityType = {
  /** Can process new workflow tasks */
  NEW_WORKFLOWS: 'NEW_WORKFLOWS',
  /** Can process existing workflow tasks */
  EXISTING_WORKFLOWS: 'EXISTING_WORKFLOWS',
  /** Can process open workflows */
  OPEN_WORKFLOWS: 'OPEN_WORKFLOWS',
  /** Can process closed workflows */
  CLOSED_WORKFLOWS: 'CLOSED_WORKFLOWS',
} as const;

type ReachabilityType = typeof ReachabilityType[keyof typeof ReachabilityType];

// Check if old build ID can be safely removed
const reachability = await taskQueueClient.getReachability(['old-build-v1.0.0']);

const canRemove = reachability.buildIdReachability.every(buildInfo =>
  buildInfo.taskQueueReachability.every(tqInfo =>
    !tqInfo.reachability.includes(ReachabilityType.OPEN_WORKFLOWS) &&
    !tqInfo.reachability.includes(ReachabilityType.NEW_WORKFLOWS)
  )
);

if (canRemove) {
  console.log('Safe to remove old build ID');
} else {
  console.log('Old build ID still needed for open workflows');
}

/**
 * Build ID not found error
 */
class BuildIdNotFoundError extends Error {
  readonly buildId: string;
  readonly taskQueue: string;

  constructor(buildId: string, taskQueue: string) {
    super(`Build ID '${buildId}' not found in task queue '${taskQueue}'`);
    this.buildId = buildId;
    this.taskQueue = taskQueue;
  }
}

class SafeDeploymentManager {
  private taskQueueClient: TaskQueueClient;

  constructor() {
    this.taskQueueClient = new TaskQueueClient();
  }

  async deployNewVersion(taskQueue: string, buildId: string) {
    // 1. Add new build ID in non-default set
    await this.taskQueueClient.updateBuildIdCompatibility(taskQueue, {
      operation: 'addNewIdInNewDefaultSet',
      buildId,
    });

    console.log(`Deployed ${buildId} to ${taskQueue} (non-default)`);

    // 2. Monitor and validate the new version
    await this.validateDeployment(taskQueue, buildId);

    // 3. Promote to default after validation
    await this.taskQueueClient.updateBuildIdCompatibility(taskQueue, {
      operation: 'promoteSetByBuildId',
      buildId,
    });

    console.log(`Promoted ${buildId} to default for ${taskQueue}`);
  }

  async validateDeployment(taskQueue: string, buildId: string): Promise<void> {
    // Wait and check metrics, health, etc.
    await new Promise(resolve => setTimeout(resolve, 60000)); // Wait 1 minute

    // Check reachability
    const reachability = await this.taskQueueClient.getReachability([buildId], [taskQueue]);

    const hasNewWorkflowCapability = reachability.buildIdReachability.some(info =>
      info.buildId === buildId &&
      info.taskQueueReachability.some(tq =>
        tq.reachability.includes(ReachabilityType.NEW_WORKFLOWS)
      )
    );

    if (!hasNewWorkflowCapability) {
      throw new Error(`Build ${buildId} cannot process new workflows`);
    }
  }

  async rollback(taskQueue: string, fallbackBuildId: string) {
    await this.taskQueueClient.updateBuildIdCompatibility(taskQueue, {
      operation: 'promoteSetByBuildId',
      buildId: fallbackBuildId,
    });

    console.log(`Rolled back to ${fallbackBuildId} for ${taskQueue}`);
  }
}

class BuildIdCleanup {
  private taskQueueClient: TaskQueueClient;

  constructor() {
    this.taskQueueClient = new TaskQueueClient();
  }

  async cleanupOldBuildIds(taskQueue: string, buildIds: string[]) {
    const reachability = await this.taskQueueClient.getReachability(buildIds, [taskQueue]);

    for (const buildInfo of reachability.buildIdReachability) {
      if (typeof buildInfo.buildId === 'string') {
        const canCleanup = buildInfo.taskQueueReachability.every(tqInfo =>
          !tqInfo.reachability.includes(ReachabilityType.OPEN_WORKFLOWS) &&
          !tqInfo.reachability.includes(ReachabilityType.NEW_WORKFLOWS)
        );

        if (canCleanup) {
          console.log(`Build ID ${buildInfo.buildId} can be safely removed`);
          // Note: Actual removal would require additional task queue management APIs
        } else {
          console.log(`Build ID ${buildInfo.buildId} still needed for active workflows`);
        }
      }
    }
  }
}