using FlinkDotNet.DataStream;

// Get Flink execution environment
var env = Flink.GetExecutionEnvironment();

// Read from Kafka
var orders = env.FromKafka("orders", "kafka:9092", "my-group");

// Transform with Flink operators
var highValueOrders = orders
    .Filter(o => o.Amount > 1000)
    .Map(o => o.ToUpperInvariant())
    .KeyBy(o => o.CustomerId);

// Write back to Kafka
highValueOrders.SinkToKafka("high-value-orders", "kafka:9092");

// Execute the job
await env.ExecuteAsync("HighValueOrderProcessor");
Console.WriteLine("Job submitted successfully!");

public record Order(
    string OrderId,
    string CustomerId,
    decimal Amount,
    string Region,
    DateTime Timestamp
);

using FlinkDotNet.DataStream;

public class OrderProcessingJob
{
    public static async Task Main(string[] args)
    {
        // Configure Flink environment
        var env = Flink.GetExecutionEnvironment();
        env.SetParallelism(4);
        env.EnableCheckpointing(TimeSpan.FromSeconds(30));

        // Read orders from Kafka
        var orders = env.FromKafka<Order>(
            topic: "orders",
            bootstrapServers: "kafka:9092",
            groupId: "order-processor"
        );

        // Filter high-value orders
        var highValueOrders = orders
            .Filter(order => order.Amount > 1000);

        // Group by region and calculate totals
        var regionalTotals = highValueOrders
            .KeyBy(order => order.Region)
            .Window(TumblingTimeWindow.Of(TimeSpan.FromMinutes(5)))
            .Reduce((a, b) => a with { Amount = a.Amount + b.Amount });

        // Write results to Kafka
        regionalTotals.SinkToKafka("regional-totals", "kafka:9092");

        // Execute the job
        var jobClient = await env.ExecuteAsync("OrderProcessingJob");
        Console.WriteLine($"Job submitted: {jobClient.JobId}");
    }
}

var env = Flink.GetExecutionEnvironment();

var input = env.FromKafka("input", "kafka:9092", "group1");
var output = input
    .Filter(msg => msg.IsValid)
    .Map(msg => msg.Transform())
    .KeyBy(msg => msg.Key);

output.SinkToKafka("output", "kafka:9092");
await env.ExecuteAsync("SimpleTransform");

var env = Flink.GetExecutionEnvironment();

var events = env.FromKafka("events", "kafka:9092", "group2");
var aggregated = events
    .KeyBy(e => e.UserId)
    .Window(TumblingTimeWindow.Of(TimeSpan.FromMinutes(1)))
    .Reduce((a, b) => new Event { Count = a.Count + b.Count });

aggregated.SinkToKafka("aggregated", "kafka:9092");
await env.ExecuteAsync("WindowedAgg");

var env = Flink.GetExecutionEnvironment();

var orders = env.FromKafka("orders", "kafka:9092", "group3");
var customers = env.FromKafka("customers", "kafka:9092", "group4");

var enriched = orders
    .Join(customers)
    .Where(order => order.CustomerId)
    .EqualTo(customer => customer.Id)
    .Window(TumblingTimeWindow.Of(TimeSpan.FromMinutes(1)))
    .Apply((order, customer) => EnrichOrder(order, customer));

enriched.SinkToKafka("enriched-orders", "kafka:9092");
await env.ExecuteAsync("OrderEnrichment");

// Enable detailed logging
var env = Flink.GetExecutionEnvironment();
env.SetLogLevel(LogLevel.Debug);

// Validate job before submission
var jobDefinition = env.GetJobDefinition();
var validation = JobDefinitionValidator.Validate(jobDefinition);
if (!validation.IsValid)
{
    Console.WriteLine($"Validation errors: {string.Join(", ", validation.Errors)}");
}

// Increase parallelism
env.SetParallelism(8);

// Tune buffer timeout
env.SetBufferTimeout(TimeSpan.FromMilliseconds(100));

// Enable operator chaining
env.EnableOperatorChaining();