T-HUB/apps/api/src/live/go-live.service.ts

import { BadRequestException, ConflictException, Inject, Injectable, NotFoundException, Optional } from "@nestjs/common";
import type { PoolClient } from "pg";
import { withTenant } from "../db/client.js";
import { LiveDestinationRepository, type OutputSummary } from "./live-destination.repository.js";
import { createGenericHlsResolver, GENERIC_HLS_PROVIDER, type LiveSourceResolver } from "./live-source-resolver.js";

export type TenantTransaction = <T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>) => Promise<T>;

/**
 * The media-plane target a go-live spawns: everything the relay needs, with
 * raw secret material held only inside this service's resolution path.
 */
export interface ResolvedOutputTarget {
  output: OutputSummary;
  /** Ingest endpoint for the push (destination's, or null for a disabled/generic-missing output). */
  endpointUrl: string;
  /** True when the credential came through a destination (vs the output's own handle). */
  throughDestination: boolean;
  /** Vault handle the worker's VaultHandleBroker redeems at relay start. */
  credentialHandleId: string | null;
}

/** Resolved program input for the take/start path. */
export interface ResolvedProgramSource {
  playbackHlsUrl: string;
  live: boolean;
  provider: string;
}

export interface GoLiveDependencies {
  transaction: TenantTransaction;
  /** Provider-keyed resolvers; connectors register their own, generic-hls is built in. */
  sourceResolvers: Map<string, LiveSourceResolver>;
}

export const GO_LIVE_DEPENDENCIES = Symbol("GO_LIVE_DEPENDENCIES");

/**
 * Go-live wiring (WP6/WP9): take/start resolves StreamOutput ->
 * LiveStreamDestination -> credential via the vault (same pattern as the
 * publish credential accessor), and resolves the program source through
 * connector-level resolvers. Nothing here mutates Redis or signs commands;
 * callers feed the result into MediaControlService.setDesired/director.
 */
@Injectable()
export class GoLiveService {
  private readonly resolvers: Map<string, LiveSourceResolver>;
  private readonly transaction: TenantTransaction;

  constructor(
    @Inject(LiveDestinationRepository) private readonly destinations: LiveDestinationRepository,
    @Optional() @Inject(GO_LIVE_DEPENDENCIES) dependencies?: Partial<GoLiveDependencies>,
  ) {
    const defaults: GoLiveDependencies = {
      transaction: withTenant,
      sourceResolvers: new Map([[GENERIC_HLS_PROVIDER, createGenericHlsResolver()]]),
    };
    const merged = { ...defaults, ...dependencies };
    this.resolvers = merged.sourceResolvers;
    this.transaction = merged.transaction;
  }

  registerSourceResolver(provider: string, resolver: LiveSourceResolver): void {
    this.resolvers.set(provider, resolver);
  }

  /**
   * Resolve every enabled output of a channel into media-plane targets.
   * Destination-referenced outputs resolve their credential handle through the
   * destination (aggregate boundary); handle outputs pass theirs through
   * untouched. The raw stream key never enters this surface — the worker
   * redeems the handle at relay start.
   */
  async resolveOutputs(workspaceId: string, virtualChannelId: string): Promise<ResolvedOutputTarget[]> {
    const outputs = await this.listChannelOutputs(workspaceId, virtualChannelId);
    const targets: ResolvedOutputTarget[] = [];
    for (const output of outputs) {
      if (!output.enabled) continue;
      if (output.destinationId !== null) {
        const destination = await this.destinations.get(workspaceId, output.destinationId);
        if (destination.credentialId === null) {
          throw new ConflictException(`destination ${destination.label} has no ingest credential`);
        }
        if (destination.credentialRevokedAt !== null) {
          throw new ConflictException(`destination ${destination.label} ingest credential is revoked`);
        }
        targets.push({
          output,
          endpointUrl: destination.endpointUrl,
          throughDestination: true,
          credentialHandleId: destination.credentialId,
        });
        continue;
      }
      const handle = output.credentialHandleId;
      if (handle === null) throw new ConflictException(`stream output ${output.id} has neither destination nor credential handle`);
      targets.push({ output, endpointUrl: "", throughDestination: false, credentialHandleId: handle });
    }
    return targets;
  }

  /**
   * Take/start resolution of the program source for a virtual channel source
   * row: provider-keyed resolvers (twitch via Helix, generic-hls direct) map
   * {channelId | externalUrl} to {playbackHlsUrl, live}.
   */
  async resolveProgramSource(workspaceId: string, request: { provider: string; channelId?: string; externalUrl?: string }): Promise<ResolvedProgramSource> {
    if (request.provider === "") throw new BadRequestException("provider is required");
    const resolver = this.resolvers.get(request.provider);
    if (resolver === undefined) throw new NotFoundException(`no live source resolver for provider ${request.provider}`);
    const resolution = await resolver({ channelId: request.channelId, externalUrl: request.externalUrl });
    if (resolution.playbackHlsUrl === null) {
      throw new ConflictException(`live source for ${request.provider} is not live`);
    }
    return { playbackHlsUrl: resolution.playbackHlsUrl, live: resolution.live, provider: request.provider };
  }

  /**
   * Redeem the destination ingest credential (vault.open with tenant + handle
   * AAD) for a destination-referenced output. Called only by the relay spawn
   * path; the returned secret never reaches logs, audit, or HTTP responses.
   */
  async redeemDestinationCredential(workspaceId: string, destinationId: string): Promise<{ endpointUrl: string; streamKey: string; provider: string }> {
    return this.destinations.resolveIngestCredential(workspaceId, destinationId);
  }

  /**
   * Take/start: build a DesiredSpec for the go-live flow from the channel's
   * enabled StreamOutputs and a resolved program source. Destination-referenced
   * outputs contribute the destination's endpoint + credential handle (resolved
   * through the vault at relay start); a single-output channel maps onto the
   * v1 spec shape MediaControlService.setDesired expects. Channels with
   * multiple enabled outputs are a multi-fan-out concern of the worker relay
   * (WP9) and resolve per-output there.
   */
  async desiredSpecForGoLive(workspaceId: string, virtualChannelId: string, source: { provider: string; channelId?: string; externalUrl?: string },
    options?: { profile?: "720p30" | "1080p30" | "1080p60"; mode?: "passthrough" | "normalize" }): Promise<{
      desiredStatus: "RUNNING";
      source: { kind: "hls"; url: string };
      output: { endpointUrl: string; credentialHandleId: string };
      profile: "720p30" | "1080p30" | "1080p60";
      mode: "passthrough" | "normalize";
    }> {
    const [program, outputs] = await Promise.all([
      this.resolveProgramSource(workspaceId, source),
      this.resolveOutputs(workspaceId, virtualChannelId),
    ]);
    if (outputs.length === 0) throw new ConflictException("no enabled stream outputs for this channel");
    const target = outputs[0]!;
    return {
      desiredStatus: "RUNNING",
      source: { kind: "hls", url: program.playbackHlsUrl },
      output: { endpointUrl: target.endpointUrl, credentialHandleId: target.credentialHandleId ?? "" },
      profile: options?.profile ?? (target.output.profile as "720p30" | "1080p30" | "1080p60"),
      mode: options?.mode ?? (target.output.mode as "passthrough" | "normalize"),
    };
  }

  private async listChannelOutputs(workspaceId: string, virtualChannelId: string): Promise<OutputSummary[]> {
    return this.transaction(workspaceId, async (client) => {
      const result = await client.query("SELECT * FROM stream_outputs WHERE workspace_id = $1 AND virtual_channel_id = $2 ORDER BY created_at", [workspaceId, virtualChannelId]);
      return result.rows.map((row) => ({
        id: String(row.id), workspaceId: String(row.workspace_id), virtualChannelId: String(row.virtual_channel_id),
        label: String(row.label), profile: String(row.profile), mode: String(row.mode), enabled: Boolean(row.enabled),
        destinationId: row.destination_id === null ? null : String(row.destination_id),
        credentialHandleId: row.credential_handle_id === null ? null : String(row.credential_handle_id),
        createdAt: new Date(row.created_at), updatedAt: new Date(row.updated_at),
      }));
    });
  }
}