///|
fn main {
  let payload = @lib.record_value([
    @lib.field("order_id", @lib.text_value("A-1001")),
    @lib.field("amount", @lib.int_value(320)),
    @lib.field("currency", @lib.text_value("CNY")),
  ])
  let event = @lib.envelope("order.created", payload, timestamp_ms=20260809).with_header(
    "source", "example",
  )
  let b1 = must_bus(@lib.bus().subscribe("audit", "order.**"))
  let b2 = must_bus(
    b1.subscribe("inventory-a", "order.created", priority=5, group="inventory"),
  )
  let b3 = must_bus(
    b2.subscribe("inventory-b", "order.created", group="inventory"),
  )
  let b4 = must_bus(
    b3.subscribe(
      "billing",
      "order.*",
      retry=@lib.retry_policy(max_attempts=1),
      guards=[@lib.HeaderEquals("source", "example")],
    ),
  )
  let schema = must_schema(
    @lib.event_schema("order-created", topic="order.*", fields=[
      @lib.field_spec("order_id", @lib.KindText, min_len=1),
      @lib.field_spec("amount", @lib.KindInt, min_int=1),
      @lib.field_spec("currency", @lib.KindText, min_len=3, max_len=3),
    ]),
  )
  let schema_report = schema.validate(event)
  let handler = fn(
    sub : @lib.Subscription,
    _event : @lib.Envelope,
  ) -> @lib.HandlerResult {
    if sub.id == "billing" {
      @lib.HandlerFail("payment gateway unavailable")
    } else {
      @lib.HandlerAck("accepted")
    }
  }
  let (next, report) = must_publish(
    b4.publish_with_mode(event, @lib.FirstPerGroup, handler),
  )
  println("EventRail example")
  println(schema_report.summary())
  println(report.status_line())
  for delivery in next.deliveries {
    println("\{delivery.subscription_id}: \{delivery.status.to_wire()}")
  }
  println("dead letters: \{next.dead_letter_count()}")
  let audit = next.audit()
  println("audit: subscriptions=\{audit.subscriptions} groups=\{audit.groups}")
  let paid = @lib.envelope("order.paid", payload, timestamp_ms=20260810).with_header(
    "source", "example",
  )
  let batch = @lib.event_batch(name="demo-batch", events=[event, paid])
  let preview = must_route_report(
    batch.route_preview(b4, mode=@lib.FirstPerGroup),
  )
  println(preview.summary())
  let projection_spec = must_projection_spec(
    @lib.projection_spec("orders-by-id", "order_id", topic="order.*", include_paths=[
      "amount", "currency", "$topic",
    ]),
  )
  let projection = must_projection(batch.project(projection_spec))
  println(projection.summary())
  let plan = @lib.bus_plan(name="example-plan")
    .add(@lib.subscription_blueprint("audit", "order.**"))
    .add(
      @lib.subscription_blueprint("billing", "order.*", guards=[
        @lib.HeaderEquals("source", "example"),
      ]),
    )
  let (_planned_bus, plan_report) = must_plan(plan.build())
  println(plan_report.summary())
  let window = must_window(
    batch.to_tape().window(@lib.event_cursor(limit=1, pattern="order.*")),
  )
  println(window.summary())
  let query = must_query(
    @lib.event_query(
      name="example-query",
      topic="order.*",
      headers=[@lib.header_filter("source", value="example")],
      limit=2,
      sort=@lib.QueryTimestampAsc,
    ),
  )
  let query_result = must_query_result(batch.query(query))
  println(query_result.summary())
  let snapshot = batch.snapshot(next, name="example-snapshot")
  println(snapshot.summary())
  let pipeline = @lib.event_pipeline("example-pipeline", batch)
    .add(@lib.PipelineValidate(schema))
    .add(@lib.PipelineRoute(b4, @lib.FirstPerGroup))
    .add(@lib.PipelineProject(projection_spec))
    .add(@lib.PipelineSnapshot(next))
  let pipeline_run = must_pipeline_run(pipeline.run())
  println(pipeline_run.summary())
  let retention_policy = must_retention_policy(
    @lib.retention_policy(
      name="example-retention",
      topic="order.*",
      max_events=1,
    ),
  )
  let (_retained_batch, retention_report) = must_retention(
    batch.apply_retention(retention_policy),
  )
  println(retention_report.summary())
  let diagnostics = @lib.diagnose_bus("example-diagnostics", next)
    .merge(@lib.diagnose_batch_schema("batch-diagnostics", batch, schema))
    .merge(@lib.diagnose_snapshot("snapshot-diagnostics", snapshot))
    .merge(@lib.diagnose_pipeline_run("pipeline-diagnostics", pipeline_run))
    .merge(@lib.diagnose_retention("retention-diagnostics", retention_report))
  println(diagnostics.summary())
}

///|
fn must_bus(result : Result[@lib.Bus, @lib.EventRailError]) -> @lib.Bus {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_schema(
  result : Result[@lib.EventSchema, @lib.EventRailError],
) -> @lib.EventSchema {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_publish(
  result : Result[(@lib.Bus, @lib.PublishReport), @lib.EventRailError],
) -> (@lib.Bus, @lib.PublishReport) {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_route_report(
  result : Result[@lib.BatchRouteReport, @lib.EventRailError],
) -> @lib.BatchRouteReport {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_projection_spec(
  result : Result[@lib.ProjectionSpec, @lib.EventRailError],
) -> @lib.ProjectionSpec {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_projection(
  result : Result[@lib.EventProjection, @lib.EventRailError],
) -> @lib.EventProjection {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_plan(
  result : Result[(@lib.Bus, @lib.PlanBuildReport), @lib.EventRailError],
) -> (@lib.Bus, @lib.PlanBuildReport) {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_window(
  result : Result[@lib.TapeWindow, @lib.EventRailError],
) -> @lib.TapeWindow {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_query(
  result : Result[@lib.EventQuery, @lib.EventRailError],
) -> @lib.EventQuery {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_query_result(
  result : Result[@lib.QueryResult, @lib.EventRailError],
) -> @lib.QueryResult {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_pipeline_run(
  result : Result[@lib.PipelineRun, @lib.EventRailError],
) -> @lib.PipelineRun {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_retention_policy(
  result : Result[@lib.RetentionPolicy, @lib.EventRailError],
) -> @lib.RetentionPolicy {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}

///|
fn must_retention(
  result : Result[(@lib.EventBatch, @lib.RetentionReport), @lib.EventRailError],
) -> (@lib.EventBatch, @lib.RetentionReport) {
  match result {
    Ok(value) => value
    Err(_) => panic()
  }
}