Eventstream CLI

az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com" \
  --headers "Content-Type=application/json" \
  --body '{"displayName": "my-eventstream", "description": "IoT sensor pipeline"}'

# 1. Build eventstream.json content (topology)
TOPOLOGY_JSON='{"compatibilityLevel":"1.1","sources":[...],"streams":[...],"operators":[...],"destinations":[...]}'

# 2. Build eventstreamProperties.json (optional — controls retention and throughput)
PROPERTIES_JSON='{"retentionTimeInDays":1,"eventThroughputLevel":"Low"}'

# 3. Base64-encode both (no line wraps)
TOPOLOGY_B64=$(echo -n "$TOPOLOGY_JSON" | base64 -w 0)
PROPERTIES_B64=$(echo -n "$PROPERTIES_JSON" | base64 -w 0)

# 4. Submit via Items API
az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/items" \
  --resource "https://api.fabric.microsoft.com" \
  --headers "Content-Type=application/json" \
  --body "{
    \"displayName\": \"my-eventstream\",
    \"type\": \"Eventstream\",
    \"definition\": {
      \"parts\": [
        {
          \"path\": \"eventstream.json\",
          \"payload\": \"${TOPOLOGY_B64}\",
          \"payloadType\": \"InlineBase64\"
        },
        {
          \"path\": \"eventstreamProperties.json\",
          \"payload\": \"${PROPERTIES_B64}\",
          \"payloadType\": \"InlineBase64\"
        }
      ]
    }
  }"

{
  "name": "FilterHighTemp",
  "type": "Filter",
  "inputNodes": [{"name": "my-stream"}],
  "properties": {
    "conditions": [{
      "column": {
        "node": null,
        "columnName": "temperature",
        "columnPath": null,
        "expressionType": "ColumnReference"
      },
      "operatorType": "GreaterThan",
      "value": {
        "dataType": "Float",
        "value": "30.0",
        "expressionType": "Literal"
      }
    }]
  }
}

az rest --method post `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/pause" `
  --resource "https://api.fabric.microsoft.com"

$resumeBodyPath = Join-Path ([IO.Path]::GetTempPath()) "eventstream-resume.json"
$resumeBody = @{ startType = "WhenLastStopped" } | ConvertTo-Json -Compress
[IO.File]::WriteAllText($resumeBodyPath, $resumeBody, [Text.UTF8Encoding]::new($false))
try {
  az rest --method post `
    --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/resume" `
    --resource "https://api.fabric.microsoft.com" `
    --headers "Content-Type=application/json" `
    --body "@$resumeBodyPath"
} finally {
  Remove-Item $resumeBodyPath -ErrorAction SilentlyContinue
}

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

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

# 2. Create empty Eventstream
$esBody = @{ displayName = "SensorIngestion"; description = "IoT sensor pipeline" } | ConvertTo-Json -Compress
$bodyFile = Join-Path ([IO.Path]::GetTempPath()) "es_create.json"
[IO.File]::WriteAllText($bodyFile, $esBody, [System.Text.UTF8Encoding]::new($false))
$created = az rest --method post `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --headers "Content-Type=application/json" `
  --body "@$bodyFile" | ConvertFrom-Json

# 3. Get the created Eventstream ID from response
$esId = $created.id
if (-not $esId) { throw "Eventstream creation did not return an ID" }

# 4. Build topology — DefaultStream uses inputNodes (not parentName)
$topology = @{
    compatibilityLevel = "1.0"
    sources = @(@{
        name = "SampleSource"
        type = "SampleData"
        properties = @{ type = "Bicycles" }
    })
    streams = @(@{
        name = "SensorIngestion-stream"
        type = "DefaultStream"
        properties = @{}
        inputNodes = @(@{ name = "SampleSource" })
    })
    operators = @()
    destinations = @()
}
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))

# 5. Deploy definition
$defBody = @{
    definition = @{
        parts = @(@{
            path = "eventstream.json"
            payload = $topologyB64
            payloadType = "InlineBase64"
        })
    }
} | ConvertTo-Json -Depth 5 -Compress
$defFile = Join-Path ([IO.Path]::GetTempPath()) "es_def.json"
[IO.File]::WriteAllText($defFile, $defBody, [System.Text.UTF8Encoding]::new($false))
# updateDefinition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$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/updateDefinition" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $defFile -Raw -Encoding UTF8)

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 }
    $succeeded = $false
    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') { $succeeded = $true; Write-Host "Update succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "updateDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "updateDefinition LRO timed out" }
}

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' 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 current 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 existing topology
$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

# 4. Add Filter operator (PascalCase name — no underscores or hyphens)
#    Column: expressionType + columnName; Value: expressionType + dataType + value
$filter = @{
    name = "FilterLowBikes"
    type = "Filter"
    inputNodes = @(@{ name = "SensorIngestion-stream" })
    properties = @{
        conditions = @(@{
            operatorType = "GreaterThan"
            column = @{
                expressionType = "ColumnReference"
                node = $null
                columnName = "No_Bikes"
                columnPath = $null
            }
            value = @{
                expressionType = "Literal"
                dataType = "BigInt"
                value = "5"
            }
        })
    }
}
$existingOps = @($topology.operators | Where-Object { $_ -ne $null })
$topology.operators = $existingOps + @($filter)

# 5. Add DerivedStream wired to filter output (makes filtered data available)
$derivedStream = @{
    name = "FilteredOutput"
    type = "DerivedStream"
    properties = @{
        inputSerialization = @{ type = "Json"; properties = @{ encoding = "UTF8" } }
    }
    inputNodes = @(@{ name = "FilterLowBikes" })
}
$existingStreams = @($topology.streams | Where-Object { $_ -ne $null })
$topology.streams = $existingStreams + @($derivedStream)

# 6. Re-encode and update
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))
$esPart.payload = $topologyB64

$defBody = @{ definition = @{ parts = $def.definition.parts } } | ConvertTo-Json -Depth 5 -Compress
$defFile = Join-Path ([IO.Path]::GetTempPath()) "es_def.json"
[IO.File]::WriteAllText($defFile, $defBody, [System.Text.UTF8Encoding]::new($false))
# updateDefinition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$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/updateDefinition" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $defFile -Raw -Encoding UTF8)

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 }
    $succeeded = $false
    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') { $succeeded = $true; Write-Host "Update succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "updateDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "updateDefinition LRO timed out" }
}

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

# 2. Build complete topology with filter + DerivedStream
$topology = @{
    compatibilityLevel = "1.0"
    sources = @(@{
        name = "CustomSource"
        type = "CustomEndpoint"
        properties = @{}
    })
    streams = @(
        @{
            name = "EventPipeline-stream"
            type = "DefaultStream"
            properties = @{}
            inputNodes = @(@{ name = "CustomSource" })
        }
        @{
            name = "FilteredEvents"
            type = "DerivedStream"
            properties = @{
                inputSerialization = @{ type = "Json"; properties = @{ encoding = "UTF8" } }
            }
            inputNodes = @(@{ name = "FilterPremium" })
        }
    )
    operators = @(@{
        name = "FilterPremium"
        type = "Filter"
        inputNodes = @(@{ name = "EventPipeline-stream" })
        properties = @{
            conditions = @(@{
                operatorType = "GreaterThan"
                column = @{
                    expressionType = "ColumnReference"
                        node = $null
                        columnName = "Amount"
                        columnPath = $null
                    }
                value = @{
                    expressionType = "Literal"
                    dataType = "BigInt"
                    value = "100"
                }
            })
        }
    })
    destinations = @()
}

# 3. Create with inline definition (single API call)
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))

$body = @{
    displayName = "EventPipeline"
    type = "Eventstream"
    definition = @{
        parts = @(@{
            path = "eventstream.json"
            payload = $topologyB64
            payloadType = "InlineBase64"
        })
    }
} | ConvertTo-Json -Depth 5 -Compress
$bodyFile = Join-Path ([IO.Path]::GetTempPath()) "es_create_full.json"
[IO.File]::WriteAllText($bodyFile, $body, [System.Text.UTF8Encoding]::new($false))

# Create-with-definition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$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/items" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $bodyFile -Raw -Encoding UTF8)

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 { 10 }
    $succeeded = $false
    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') { $succeeded = $true; Write-Host "Create succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "Create LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "Create LRO timed out" }
} else {
    Write-Host "Created: $(($response.Content | ConvertFrom-Json).displayName)"
}

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' 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. Delete
az rest --method delete `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId" `
  --resource "https://api.fabric.microsoft.com"