Eventstream Consumption CLI

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com"

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com" \
  --query "value[?displayName=='my-eventstream']"

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}" \
  --resource "https://api.fabric.microsoft.com"

az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/getDefinition" \
  --resource "https://api.fabric.microsoft.com" \
  --body '{}'

# Using jq + base64 (Linux; on macOS use base64 -D instead of -d)
az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/getDefinition" \
  --resource "https://api.fabric.microsoft.com" \
  --body '{}' \
  | jq -r '.definition.parts[] | select(.path=="eventstream.json") | .payload' \
  | base64 -d | jq .

# PowerShell (Windows)
$def = az rest --method POST `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/getDefinition" `
  --resource "https://api.fabric.microsoft.com" `
  --body '{}' | ConvertFrom-Json
$payload = ($def.definition.parts | Where-Object { $_.path -eq 'eventstream.json' }).payload
[Text.Encoding]::UTF8.GetString([Convert]::FromBase64String($payload)) | ConvertFrom-Json | ConvertTo-Json -Depth 10

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/topology" \
  --resource "https://api.fabric.microsoft.com"

# Extract the sourceId for a Custom Endpoint source (use name filter if multiple exist)
SOURCE_ID=$(az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/topology" \
  --resource "https://api.fabric.microsoft.com" \
  | jq -r '[.sources[] | select(.type=="CustomEndpoint")] | if length == 0 then error("No Custom Endpoint sources found in this Eventstream") elif length > 1 then error("Multiple Custom Endpoint sources found — filter by .name") else .[0].id end') \
  || { echo "Failed to resolve Custom Endpoint source ID"; exit 1; }

if [ -z "$SOURCE_ID" ]; then echo "SOURCE_ID is empty — check topology output"; exit 1; fi

# PowerShell — extract sourceId for Custom Endpoint (fails clearly if multiple exist)
$topology = az rest --method GET `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
$customSources = @($topology.sources | Where-Object { $_.type -eq 'CustomEndpoint' })
if ($customSources.Count -eq 0) { throw "No Custom Endpoint sources found in this Eventstream" }
if ($customSources.Count -gt 1) { throw "Multiple Custom Endpoint sources found. Filter by name: $($customSources.name -join ', ')" }
$sourceId = $customSources[0].id

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/sources/${SOURCE_ID}/connection" \
  --resource "https://api.fabric.microsoft.com"

az rest --method GET `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/sources/$sourceId/connection" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

{
  "fullyQualifiedNamespace": "namespace.servicebus.windows.net",
  "eventHubName": "es_<guid>",
  "accessKeys": {
    "primaryKey": "...",
    "secondaryKey": "...",
    "primaryConnectionString": "Endpoint=sb://namespace.servicebus.windows.net/;...",
    "secondaryConnectionString": "..."
  }
}

# 1. Discover workspace ID
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }

# 2. List Eventstreams (handles pagination)
$allItems = @()
$resp = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
$allItems += $resp.value
while ($resp.continuationUri) {
    $resp = az rest --method get `
      --url $resp.continuationUri `
      --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
    $allItems += $resp.value
}
$allItems | Select-Object displayName, id, description | Format-Table

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }

$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }

# 2. Get topology (returns JSON directly — no base64 decoding needed)
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

# 3. Summarize nodes (filter nulls from arrays)
Write-Host "Sources:"
@($topo.sources | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type), id: $($_.id))"
}
Write-Host "Operators:"
@($topo.operators | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}
Write-Host "Destinations:"
@($topo.destinations | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}
Write-Host "Streams:"
@($topo.streams | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }

$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/items?type=Eventstream" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }

# 2. Get definition (handles LRO via Location header)
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/getDefinition" `
  -Headers @{ Authorization = "Bearer $token"; "Content-Type" = "application/json" } `
  -Body '{}'

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $def = $null
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') {
            $def = Invoke-RestMethod -Uri "$location/result" `
              -Headers @{ Authorization = "Bearer $token" }
            break
        } elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "getDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $def -or -not $def.definition) { throw "getDefinition LRO timed out (last status: $(if ($poll) { $poll.status } else { 'unknown' }))" }
} else {
    $def = $response.Content | ConvertFrom-Json
}

# 3. Decode eventstreamProperties.json part (holds retention + throughput)
$propsPart = $def.definition.parts | Where-Object { $_.path -eq 'eventstreamProperties.json' } | Select-Object -First 1
if ($propsPart) {
    $props = [Text.Encoding]::UTF8.GetString(
      [Convert]::FromBase64String($propsPart.payload)) | ConvertFrom-Json
    Write-Host "Retention: $($props.retentionTimeInDays) days"
    Write-Host "Throughput Level: $($props.eventThroughputLevel)"
} else {
    # Fall back to topology-level properties (older format)
    $esPart = $def.definition.parts | Where-Object { $_.path -eq 'eventstream.json' } | Select-Object -First 1
    if (-not $esPart) { throw "eventstream.json part not found in definition" }
    $topology = [Text.Encoding]::UTF8.GetString(
      [Convert]::FromBase64String($esPart.payload)) | ConvertFrom-Json
    if ($topology.properties.retentionTimeInDays) {
        Write-Host "Retention: $($topology.properties.retentionTimeInDays) days"
    } else {
        Write-Host "Retention: (platform default — not explicitly configured)"
    }
    if ($topology.properties.eventThroughputLevel) {
        Write-Host "Throughput Level: $($topology.properties.eventThroughputLevel)"
    } else {
        Write-Host "Throughput Level: (platform default — not explicitly configured)"
    }
}

# 1. Discover workspace + Eventstream IDs (omitted for brevity)

# 2. Get topology to find Custom Endpoint source ID
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

$ceSource = @($topo.sources | Where-Object { $_.type -eq 'CustomEndpoint' }) | Select-Object -First 1
if (-not $ceSource) { throw "No CustomEndpoint source found in this Eventstream" }
$sourceId = $ceSource.id

# 3. Get connection metadata
$conn = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/sources/$sourceId/connection" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

Write-Host "Fully Qualified Namespace: $($conn.fullyQualifiedNamespace)"
Write-Host "Event Hub Name:            $($conn.eventHubName)"
Write-Host "Kafka Bootstrap Server:    $($conn.fullyQualifiedNamespace):9093"

# 1. Discover workspace + Eventstream IDs (omitted for brevity)

# 2. Get topology
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

# 3. Validate sources
@($topo.sources | Where-Object { $_ -ne $null }) | ForEach-Object {
    $src = $_
    Write-Host "Source: $($src.name) (type: $($src.type))"
    if ($src.properties.dataConnectionId) {
        Write-Host "  Cloud connection: $($src.properties.dataConnectionId)"
    }
    if ($src.type -in @('AzureEventHub','AzureEventHubExtended','AzureIoTHub','ConfluentCloud','ApacheKafka','AmazonMSKKafka') -and
        -not $src.properties.consumerGroupName) {
        Write-Host "  WARNING: No consumer group set (required for $($src.type))"
    }
}

# 4. Validate destinations
@($topo.destinations | Where-Object { $_ -ne $null }) | ForEach-Object {
    $dst = $_
    Write-Host "Destination: $($dst.name) (type: $($dst.type))"
    if (-not $dst.inputNodes -or $dst.inputNodes.Count -eq 0) {
        Write-Host "  WARNING: No input wired — destination will receive no events"
    }
    if ($dst.type -eq 'Eventhouse' -and -not $dst.properties.tableName) {
        Write-Host "  WARNING: No target table configured"
    }
    if ($dst.type -eq 'Eventhouse' -and $dst.properties.dataIngestionMode -eq 'DirectIngestion') {
        if (-not $dst.properties.connectionName -or -not $dst.properties.mappingRuleName) {
            Write-Host "  WARNING: DirectIngestion requires connectionName and mappingRuleName"
        }
    }
}

# 5. Check node count limits
$ceCount = @($topo.sources | Where-Object { $_.type -eq 'CustomEndpoint' }).Count +
           @($topo.destinations | Where-Object { $_.type -eq 'CustomEndpoint' }).Count +
           @($topo.destinations | Where-Object { $_.type -eq 'Eventhouse' -and
             $_.properties.dataIngestionMode -eq 'DirectIngestion' }).Count
Write-Host "CustomEndpoint + DI count: $ceCount / 11 limit"
if ($ceCount -gt 11) {
    Write-Host "WARNING: Exceeds limit of 11 CustomEndpoint + DirectIngestion nodes"
}