Jasny przepływ kodu:
od Command-Event
do Fluent Method Chaining
i Monad w C#
@walenciukC
od Command-Event do Fluent Method Chaining i Monad w C#
Jak napisać kod, który będzie zrozumiały nawet dla twojego product ownera czy scrum mastera?
Wniosek 1:
Nazwy klas, metod, folderów, projektów kierują nasz wzrok i mózg do konkretnego fragmentu kodu, który nas interesuje
W Domain Driven Design mamy "Ubiquitous Language", czyli wspólny język między biznesem a programistami
Tak żebyś wiedział, gdzie szukać tego w kodzie, gdy ktoś biznesowo opisze Ci problem z daną aplikacją
Wniosek 2:
Spodobało mi się to, że w tym kodzie nie wyrzucałem wyjątków.
Miałem "Rezultaty", które mówiły mi, czy dane polecenie się wykonało, czy nie, i co poszło nie tak
Nie sterowałem przepływem wyjątkami
Wniosek 3:
"Zdarzenia" na pewno nie powinny sterować przepływem.
"Zdarzenia" czy strumień zdarzeń powinny dawać możliwość podpinania dodatkowego kodu,
gdy dane polecenie się wykonało.
Wniosek 4:
Nieważne "jak", muszę mieć w jednym miejscu w kodzie widok przepływu kodu
Wniosek 5:
O ile to możliwe, stworzyć polecenia jak klocki Lego, którym mogę dowolnie zmieniać kolejność.
Chociaż wiem, że ma to swoją cenę
Wniosek 6:
Zobaczyłem potrzebę podziału przepływu per "jeden element" i per "cały batch", a w tym rozwiązaniu średnio to było widać
O ile po nazwie można było wszystko znaleźć, to najbardziej bolesny był brak czytelności i zrozumienia przepływu kodu
Zaraz, naprawdę bez wyjątków?
Wyjątki można wyrzucać tylko wtedy, gdy nastąpił błąd krytyczny i chcecie wyłączyć bądź zablokować całą aplikację
Warto stworzyć sobie "wajhę", która będzie wrzucała wyjątki zamiast błędnych "rezultatów", aby debugowało się wygodniej
Scentralizowany i widoczny przepływ kodu to jest to
To usunie potrzebę kminienia, jak projekty, foldery i pliki powinny być ułożone, bo będzie to widać w jednym miejscu, na starcie aplikacji
Wniosek 7:
Nie ma co trzymać się sztywno pewnych zasad. Railway Oriented Programming powiedziałby nam,
że metody muszą się nazywać "Bind", "Map", "Match" i mieć konkretne sygnatury.
Albo mieć klasy "Maybe", "Either"
Ale to nie jest ważne. Ważne jest, żeby mieć czytelny przepływ kodu
Tutaj wstaw wykład na temat "Monad"
Jednak wiem, że ten "wykład" nic nie wniesie
var filePipeline = Flow.Start()
.Then(StartTimingCommand.Run(bus))
.Then(OpenFileCommand.Run(bus))
.Then(ParseCsvCommand.Run(bus))
.Then(ValidateRowsCommand.Run(bus, app.Rules))
.Then(DeduplicationCommand.Run(bus))
.ThenBranch("SchemaRoute", branch => branch
.When("Orders", ctx => ctx.Schema == "Orders", pipelineOrdersBranch)
.When("Sales", ctx => ctx.Schema == "Sales", pipelineSalesBranch)
.When("Users", ctx => ctx.Schema == "Users", pipelineUsersBranch)
.When("Products", ctx => ctx.Schema == "Products", pipelineProductsBranch))
.ThenWaitForAll("Enrichment",
GeoLookupCommand.Run(bus),
AuditStampCommand.Run(bus))
.Then(MapEntitiesCommand.Run(bus))
.Then(SaveBatchCommand.Run(bus))
.Build();
var filePipeline = Flow.Start()
.Then(new TimingCommand(bus))
.Then(new OpenFileCommand(bus))
.Then(new ParseCsvCommand(bus))
.Then(new ValidateRowsCommand(bus, app.Rules))
.Then(new DeduplicationCommand(bus))
.ThenBranch("SchemaRoute", branch => branch
.When("Orders", ctx => ctx.Schema == "Orders", ordersBranch)
.When("Sales", ctx => ctx.Schema == "Sales", salesBranch)
.When("Users", ctx => ctx.Schema == "Users", usersBranch)
.When("Products", ctx => ctx.Schema == "Products", productsBranch))
.ThenWaitForAll("Enrichment",
new GeoLookupCommand(bus), // np. geolokalizacja regionu
new AuditStampCommand(bus)) // np. audit timestamp
.Then(new MapEntitiesCommand(bus))
.Then(new SaveBatchCommand(bus))
.Build();
var flow = Flow.Start()
// cross-cutting
.Then<StartTimingCommand>()
// wspólne kroki wejściowe
.Then<OpenFileCSVCommand>() // → ctx.FileContext
.Then<ParseCsvCommand>() // → ctx.ParsedFile
.Then<ValidateRowsCommand>() // → ctx.ValidatedFile | ctx.NoItemsReason
.Then<DeduplicationCommand>() // → ctx.ValidatedFile | ctx.NoItemsReason
// rozgałęzienie per schemat
.ThenBranch("SchemaRoute", branch => branch
.When("Orders", ctx => ctx.FileContext?.Schema == "Orders",
then => then.Then<CalcOrderTotalsCommand>())
.When("Sales", ctx => ctx.FileContext?.Schema == "Sales",
then => then.Then<CurrencyConvertCommand>())
.When("Users", ctx => ctx.FileContext?.Schema == "Users",
then => then.Then<ValidateEmailCommand>()) // | ctx.NoItemsReason
.When("Products", ctx => ctx.FileContext?.Schema == "Products",
then => then.Then<ValidateStockCommand>()))
// wspólne kroki wyjściowe
.Then<MapEntitiesCommand>() // → ctx.Entities
// równoległe wzbogacenie
.ThenWaitForAll<GeoLookupCommand, AuditStampCommand>("Enrichment")
.Then<SaveBatchCommand>() // → ctx.ImportResult
// BuildScoped — każde wywołanie = nowy scope DI
.BuildScoped(services);
public class BatchPipeline
{
private readonly ScopedFlow _flow;
private readonly IPipelineEventBus _bus;
public BatchPipeline(ScopedFlow flow, IPipelineEventBus bus)
{
_flow = flow;
_bus = bus;
}
public async Task<Result<BatchReport>> ProcessBatchAsync(IEnumerable<string> filePaths)
{
var paths = filePaths.ToArray();
var missing = paths.Where(p => !File.Exists(p)).ToArray();
if (missing.Length > 0)
return Result<BatchReport>.Fail(
$"Paths not found: {string.Join(", ", missing.Select(Path.GetFileName))}");
var imports = new List<ImportResult>();
var errors = new List<FileError>();
var skipped = new List<FileSkipped>();
foreach (var path in paths)
{
Console.WriteLine($"\n Plik: {Path.GetFileName(path)}");
var ctx = new PipelineContext { FilePath = path };
await _flow.RunAsync(ctx);
if (ctx.Failed)
errors.Add(new FileError(path, ctx.Error!));
else if (ctx.IsEmpty)
skipped.Add(new FileSkipped(path, ctx.NoItemsReason!));
else
imports.Add(ctx.ImportResult!);
ctx.Trace.Dump();
}
_bus.Publish(new PipelineEvent.BatchCompleted(
paths, imports.Count, errors.Count));
return Result<BatchReport>.Ok(new BatchReport(
Total: paths.Length,
Succeeded: imports.Count,
Skipped: skipped.Count,
Failed: errors.Count,
Imports: imports,
SkippedFiles: skipped,
Errors: errors));
}
}
public abstract record PipelineEvent(string FilePath, DateTime Timestamp)
{
public record StepStarted(
string FilePath, string StepName
) : PipelineEvent(FilePath, DateTime.UtcNow);
public record StepCompleted(
string FilePath, string StepName, TimeSpan Duration, string Detail
) : PipelineEvent(FilePath, DateTime.UtcNow);
public record StepFailed(
string FilePath, string StepName, string Error
) : PipelineEvent(FilePath, DateTime.UtcNow);
public record StepNoItems(
string FilePath, string StepName, string Reason
) : PipelineEvent(FilePath, DateTime.UtcNow);
public record FileCompleted(
string FilePath, int RowsSaved, int RowsSkipped
) : PipelineEvent(FilePath, DateTime.UtcNow);
public record BatchCompleted(
string[] FilePaths, int Succeeded, int Failed
) : PipelineEvent("batch", DateTime.UtcNow);
}
public interface IPipelineEventBus
{
void Publish(PipelineEvent evt);
IDisposable Subscribe(Action<PipelineEvent> handler);
}
public class PipelineEventBus : IPipelineEventBus
{
private readonly List<Action<PipelineEvent>> _handlers = [];
public void Publish(PipelineEvent evt)
{
foreach (var handler in _handlers)
handler(evt);
}
public IDisposable Subscribe(Action<PipelineEvent> handler)
{
_handlers.Add(handler);
return new Unsubscriber(() => _handlers.Remove(handler));
}
private sealed class Unsubscriber(Action onDispose) : IDisposable
{
public void Dispose() => onDispose();
}
}
public static class EventSubscribers
{
/// <summary>
/// Loguje każdy krok na konsolę z ikonami i kolorami.
/// </summary>
public static Action<PipelineEvent> ConsoleLogger => evt =>
{
switch (evt)
{
case PipelineEvent.StepStarted e:
Console.ForegroundColor = ConsoleColor.DarkGray;
Console.WriteLine($" => {e.StepName}...");
Console.ResetColor();
break;
case PipelineEvent.StepCompleted e:
Console.ForegroundColor = ConsoleColor.Green;
Console.Write($" OK {e.StepName,-22}");
Console.ResetColor();
Console.WriteLine($" {e.Duration.TotalMilliseconds,6:F1}ms {e.Detail}");
break;
case PipelineEvent.StepFailed e:
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($" OH NO {e.StepName,-22} {e.Error}");
Console.ResetColor();
break;
case PipelineEvent.StepNoItems e:
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine($" ○ {e.StepName,-22} {e.Reason}");
Console.ResetColor();
break;
case PipelineEvent.FileCompleted e:
Console.ForegroundColor = ConsoleColor.Cyan;
Console.WriteLine($" => {Path.GetFileName(e.FilePath)}: " +
$"{e.RowsSaved} saved, {e.RowsSkipped} skipped");
Console.ResetColor();
break;
case PipelineEvent.BatchCompleted e:
Console.WriteLine();
Console.ForegroundColor = e.Failed == 0 ? ConsoleColor.Green : ConsoleColor.Yellow;
Console.WriteLine($" Batch: {e.Succeeded} ok / {e.Failed} errors " +
$"({e.FilePaths.Length} total)");
Console.ResetColor();
break;
}
};
/// <summary>
/// Zbiera metryki do analizy — czasy kroków, error rate itp.
/// </summary>
public class MetricsCollector
{
private readonly List<PipelineEvent.StepCompleted> _completed = [];
private readonly List<PipelineEvent.StepFailed> _failed = [];
public IReadOnlyList<PipelineEvent.StepCompleted> Completed => _completed;
public IReadOnlyList<PipelineEvent.StepFailed> Failed => _failed;
public Action<PipelineEvent> Handler => evt =>
{
if (evt is PipelineEvent.StepCompleted sc) _completed.Add(sc);
if (evt is PipelineEvent.StepFailed sf) _failed.Add(sf);
};
public void PrintSummary()
{
Console.WriteLine("\n── Metrics ──────────────────────────────");
var byStep = _completed.GroupBy(c => c.StepName);
foreach (var group in byStep)
{
var avg = group.Average(c => c.Duration.TotalMilliseconds);
var max = group.Max(c => c.Duration.TotalMilliseconds);
Console.WriteLine($" {group.Key,-22} avg={avg:F1}ms max={max:F1}ms n={group.Count()}");
}
Console.WriteLine($" Errors: {_failed.Count}");
}
}
}
using Microsoft.Extensions.DependencyInjection;
using CsvBatchImporterDI;
using CsvBatchImporterDI.Commands;
using CsvBatchImporterDI.Events;
using CsvBatchImporterDI.Models;
using CsvBatchImporterDI.Pipeline;
var services = CompositionRoot.CreateServices();
var bus = services.GetRequiredService<PipelineEventBus>();
var options = services.GetRequiredService<PipelineOptions>();
var env = services.GetRequiredService<EnvironmentInfo>();
var metrics = services.GetRequiredService<EventSubscribers.MetricsCollector>();
bus.Subscribe(EventSubscribers.ConsoleLogger);
bus.Subscribe(metrics.Handler);
var flow = Flow.Start()
// cross-cutting
.Then<StartTimingCommand>()
// wspólne kroki wejściowe
.Then<OpenFileCSVCommand>() // → ctx.FileContext
.Then<ParseCsvCommand>() // → ctx.ParsedFile
.Then<ValidateRowsCommand>() // → ctx.ValidatedFile | ctx.NoItemsReason
.Then<DeduplicationCommand>() // → ctx.ValidatedFile | ctx.NoItemsReason
// rozgałęzienie per schemat
.ThenBranch("SchemaRoute", branch => branch
.When("Orders", ctx => ctx.FileContext?.Schema == "Orders",
then => then.Then<CalcOrderTotalsCommand>())
.When("Sales", ctx => ctx.FileContext?.Schema == "Sales",
then => then.Then<CurrencyConvertCommand>())
.When("Users", ctx => ctx.FileContext?.Schema == "Users",
then => then.Then<ValidateEmailCommand>()) // | ctx.NoItemsReason
.When("Products", ctx => ctx.FileContext?.Schema == "Products",
then => then.Then<ValidateStockCommand>()))
// wspólne kroki wyjściowe
.Then<MapEntitiesCommand>() // → ctx.Entities
// równoległe wzbogacenie
.ThenWaitForAll<GeoLookupCommand, AuditStampCommand>("Enrichment")
.Then<SaveBatchCommand>() // → ctx.ImportResult
// BuildScoped — każde wywołanie = nowy scope DI
.BuildScoped(services);
var input = SampleBatchFiles.Create();
ConsoleBatchReporter.PrintStartup(env, options, flow, input);
var batch = new BatchPipeline(flow, bus);
var result = await batch.ProcessBatchAsync(input.ExistingPaths);
ConsoleBatchReporter.PrintReport(result, input.MissingPaths, metrics);
namespace CsvBatchImporterDI;
public static class SampleBatchFiles
{
public static BatchInput Create()
{
var baseDir = AppContext.BaseDirectory;
var dataDir = Path.Combine(baseDir, "SampleData");
var filePaths = new[]
{
Path.Combine(dataDir, "users.csv"), // → sukces
Path.Combine(dataDir, "users_all_invalid.csv"), // → NoItemsToProcess (all rows fail validation)
Path.Combine(dataDir, "orders.csv"), // → sukces
Path.Combine(dataDir, "sales.csv"), // → sukces
Path.Combine(dataDir, "invoices_unknown_schema.csv"), // → Failure (unknown schema)
Path.Combine(dataDir, "missing_file.csv"), // → pominięty (pre-filter)
};
return new BatchInput(
ExistingPaths: filePaths.Where(File.Exists).ToArray(),
MissingPaths: filePaths.Where(p => !File.Exists(p)).ToArray());
}
}
public record BatchInput(string[] ExistingPaths, string[] MissingPaths);
using CsvBatchImporterDI.Events;
using CsvBatchImporterDI.Models;
using CsvBatchImporterDI.Pipeline;
namespace CsvBatchImporterDI;
public static class ConsoleBatchReporter
{
public static void PrintStartup(
EnvironmentInfo env,
PipelineOptions options,
ScopedFlow flow,
BatchInput input)
{
Console.WriteLine("╔══════════════════════════════════════════════╗");
Console.WriteLine("║ CSV Batch Importer [Flow + Scoped DI] ║");
Console.WriteLine("╚══════════════════════════════════════════════╝");
Console.Write($"\nEnvironment: {env.Name} | BreakOnError: ");
if (options.BreakOnError)
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine("true ← debugger zatrzyma się na błędzie");
}
else
{
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("false ← tryb produkcyjny");
}
Console.ResetColor();
if (input.MissingPaths.Length > 0)
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine("\nPominięte (nie znaleziono):");
foreach (var missing in input.MissingPaths)
Console.WriteLine($" ✗ {Path.GetFileName(missing)}");
Console.ResetColor();
}
Console.WriteLine($"\nPipeline: {string.Join(" → ", flow.GetRegistered())}");
Console.WriteLine($"Batch: {input.ExistingPaths.Length} plików\n");
Console.WriteLine("─────────────────────────────────────────────");
}
public static void PrintReport(
Result<BatchReport> result,
string[] missingPaths,
EventSubscribers.MetricsCollector metrics)
{
Console.WriteLine("\n─────────────────────────────────────────────");
Console.WriteLine("RAPORT BATCHA");
Console.WriteLine("─────────────────────────────────────────────");
if (!result.IsSuccess)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($"Batch zakończony błędem: {result.Error}");
Console.ResetColor();
return;
}
var report = result.Value!;
if (report.Succeeded > 0)
{
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine($"\nZaimportowane ({report.Succeeded}):");
Console.ResetColor();
foreach (var import in report.Imports)
{
var skip = import.RowsSkipped > 0 ? $" ({import.RowsSkipped} pominięto)" : "";
Console.WriteLine($" + {Path.GetFileName(import.FilePath)}: {import.RowsSaved} wierszy{skip}");
}
}
if (report.Skipped > 0)
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine($"\nPominięte — brak wierszy ({report.Skipped}):");
Console.ResetColor();
foreach (var skipped in report.SkippedFiles)
Console.WriteLine($" ○ {Path.GetFileName(skipped.Path)}: {skipped.Reason}");
}
var allErrors = report.Errors
.Concat(missingPaths.Select(p => new FileError(p, "File not found")))
.ToList();
if (allErrors.Count > 0)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($"\nBłędy ({allErrors.Count}):");
Console.ResetColor();
foreach (var error in allErrors)
Console.WriteLine($" - {Path.GetFileName(error.Path)}: {error.Reason}");
}
var total = report.Total + missingPaths.Length;
Console.WriteLine($"\nŁącznie: {total} | OK: {report.Succeeded} | Puste: {report.Skipped} | Błędy: {allErrors.Count}");
Console.ForegroundColor = allErrors.Count == 0 && report.Skipped == 0 ? ConsoleColor.Green
: report.Succeeded > 0 ? ConsoleColor.Yellow : ConsoleColor.Red;
Console.WriteLine(allErrors.Count == 0 && report.Skipped == 0 ? "Batch OK."
: report.Succeeded > 0 ? "Częściowy sukces." : "Batch failed.");
Console.ResetColor();
metrics.PrintSummary();
}
}
using System.Diagnostics;
using CsvBatchImporterDI.Events;
using CsvBatchImporterDI.Models;
using CsvBatchImporterDI.Pipeline;
namespace CsvBatchImporterDI.Commands;
/// <summary>
/// Bazowa klasa dla kroków biznesowych.
/// Automatycznie: pomija się gdy ctx.Failed, mierzy czas, emituje eventy, zapisuje trace.
/// Podklasa implementuje tylko Execute.
/// </summary>
public abstract class Command(IPipelineEventBus bus) : ICommand
{
public abstract string Name { get; }
public async Task InvokeAsync(PipelineContext ctx, FlowDelegate next)
{
if (ctx.Halted)
{
await next(ctx);
return;
}
bus.Publish(new PipelineEvent.StepStarted(ctx.FilePath, Name));
var sw = Stopwatch.StartNew();
try
{
Execute(ctx);
}
catch (Exception ex)
{
ctx.Error = $"{Name}: {ex.Message}";
}
sw.Stop();
if (ctx.Failed)
{
ctx.Trace.Add(Name, ctx.Error!, sw.Elapsed, false);
bus.Publish(new PipelineEvent.StepFailed(ctx.FilePath, Name, ctx.Error!));
}
else if (ctx.IsEmpty)
{
ctx.Trace.Add(Name, ctx.NoItemsReason!, sw.Elapsed, true);
bus.Publish(new PipelineEvent.StepNoItems(ctx.FilePath, Name, ctx.NoItemsReason!));
}
else
{
var detail = GetDetail(ctx);
ctx.Trace.Add(Name, detail, sw.Elapsed, true);
bus.Publish(new PipelineEvent.StepCompleted(ctx.FilePath, Name, sw.Elapsed, detail));
}
await next(ctx);
}
/// <summary>Logika kroku — ustaw pola na ctx, lub ustaw ctx.Error w razie błędu.</summary>
protected abstract void Execute(PipelineContext ctx);
/// <summary>Krótki opis wyniku do trace'a.</summary>
protected virtual string GetDetail(PipelineContext ctx) => "ok";
}
using Microsoft.Extensions.DependencyInjection;
using CsvBatchImporterDI.Models;
namespace CsvBatchImporterDI.Pipeline;
// ── Delegate ──────────────────────────────────────────────────────
public delegate Task FlowDelegate(PipelineContext ctx);
// ── Command interface ─────────────────────────────────────────────
public interface ICommand
{
string Name { get; }
Task InvokeAsync(PipelineContext ctx, FlowDelegate next);
}
// ── Flow builder ──────────────────────────────────────────────────
public class Flow
{
private readonly List<Func<IServiceProvider, ICommand>> _factories = [];
private readonly List<string> _names = [];
private Flow() { }
public static Flow Start() => new();
/// <summary>Dodaj krok — instancja podana jawnie.</summary>
public Flow Then(ICommand command)
{
_factories.Add(_ => command);
_names.Add(command.Name);
return this;
}
/// <summary>Dodaj krok — typ rozwiązywany z DI per scope.</summary>
public Flow Then<T>() where T : ICommand
{
_factories.Add(sp => sp.GetRequiredService<T>());
_names.Add(typeof(T).Name.Replace("Command", ""));
return this;
}
/// <summary>Warunkowy krok z DI.</summary>
public Flow ThenIf<T>(Func<PipelineContext, bool> predicate) where T : ICommand
{
_factories.Add(sp => new ConditionalCommand(predicate, sp.GetRequiredService<T>()));
_names.Add($"If({typeof(T).Name.Replace("Command", "")})");
return this;
}
/// <summary>Rozgałęzienie — sub-flow budowane ze scope'u w runtime.</summary>
public Flow ThenBranch(string name, Action<BranchBuilder> configure)
{
var branch = new BranchBuilder(name);
configure(branch);
_factories.Add(_ => branch.Build());
_names.Add(name);
return this;
}
/// <summary>Równoległe wykonanie z DI.</summary>
public Flow ThenWaitForAll<T1, T2>(string name)
where T1 : ICommand
where T2 : ICommand
{
_factories.Add(sp => new WaitForAllCommand(name, [
sp.GetRequiredService<T1>(),
sp.GetRequiredService<T2>()
]));
_names.Add(name);
return this;
}
/// <summary>BuildScoped — każde wywołanie RunAsync tworzy nowy scope.</summary>
public ScopedFlow BuildScoped(IServiceProvider rootProvider)
=> new(rootProvider, _factories, _names);
/// <summary>Build bez scope'u (testy, singleton flow).</summary>
public FlowDelegate Build()
=> BuildWith(new EmptyServiceProvider());
internal FlowDelegate BuildWith(IServiceProvider sp)
{
var commands = _factories.Select(f => f(sp)).ToList();
FlowDelegate terminal = _ => Task.CompletedTask;
return commands.AsEnumerable().Reverse()
.Aggregate(terminal, (next, cmd) => ctx => cmd.InvokeAsync(ctx, next));
}
internal List<Func<IServiceProvider, ICommand>> Factories => _factories;
public IReadOnlyList<string> GetRegistered() => _names;
private class EmptyServiceProvider : IServiceProvider
{
public object? GetService(Type serviceType) => null;
}
internal class EmptyServiceProviderPublic : IServiceProvider
{
public object? GetService(Type serviceType) => null;
}
}
// ── Scoped flow ───────────────────────────────────────────────────
public class ScopedFlow
{
private readonly IServiceProvider _rootProvider;
private readonly List<Func<IServiceProvider, ICommand>> _factories;
private readonly IReadOnlyList<string> _names;
internal ScopedFlow(
IServiceProvider rootProvider,
List<Func<IServiceProvider, ICommand>> factories,
List<string> names)
{
_rootProvider = rootProvider;
_factories = factories;
_names = names;
}
/// <summary>
/// Nowy scope DI → zbuduj łańcuch → wykonaj → dispose scope.
/// Scoped serwisy (DbContext, UnitOfWork) żyją per plik.
/// </summary>
public async Task RunAsync(PipelineContext ctx)
{
await using var scope = _rootProvider.CreateAsyncScope();
ctx.ScopedServices = scope.ServiceProvider;
var commands = _factories.Select(f => f(scope.ServiceProvider)).ToList();
FlowDelegate terminal = _ => Task.CompletedTask;
var pipeline = commands.AsEnumerable().Reverse()
.Aggregate(terminal, (next, cmd) => c => cmd.InvokeAsync(c, next));
await pipeline(ctx);
}
public IReadOnlyList<string> GetRegistered() => _names;
}
// ── Warunkowy wrapper ─────────────────────────────────────────────
internal class ConditionalCommand(
Func<PipelineContext, bool> predicate,
ICommand inner) : ICommand
{
public string Name => $"If({inner.Name})";
public async Task InvokeAsync(PipelineContext ctx, FlowDelegate next)
{
if (predicate(ctx))
await inner.InvokeAsync(ctx, next);
else
await next(ctx);
}
}
// ── Branch — sub-flow budowany ze scoped providera w runtime ──────
public class BranchBuilder
{
private readonly string _name;
private readonly List<(string label, Func<PipelineContext, bool> predicate, Flow subFlow)> _branches = [];
internal BranchBuilder(string name) => _name = name;
/// <summary>Dodaj gałąź — lambda konfiguruje sub-flow, bez jawnego Flow.Start().</summary>
public BranchBuilder When(string label, Func<PipelineContext, bool> predicate, Action<Flow> configure)
{
var subFlow = Flow.Start();
configure(subFlow);
_branches.Add((label, predicate, subFlow));
return this;
}
internal BranchCommand Build() => new(_name, _branches);
}
internal class BranchCommand(
string name,
List<(string label, Func<PipelineContext, bool> predicate, Flow subFlow)> branches) : ICommand
{
public string Name => name;
public async Task InvokeAsync(PipelineContext ctx, FlowDelegate next)
{
foreach (var (label, predicate, subFlow) in branches)
{
if (predicate(ctx))
{
ctx.Trace.Add(Name, $"→ {label}", TimeSpan.Zero, true);
// Buduj sub-flow z tego samego scoped providera
var sp = ctx.ScopedServices ?? new Flow.EmptyServiceProviderPublic();
var subPipeline = subFlow.BuildWith(sp);
await subPipeline(ctx);
await next(ctx);
return;
}
}
ctx.Trace.Add(Name, "→ (no match)", TimeSpan.Zero, true);
await next(ctx);
}
}
// ── WaitForAll ────────────────────────────────────────────────────
internal class WaitForAllCommand(string name, ICommand[] commands) : ICommand
{
public string Name => name;
public async Task InvokeAsync(PipelineContext ctx, FlowDelegate next)
{
if (ctx.Failed) { await next(ctx); return; }
var sw = System.Diagnostics.Stopwatch.StartNew();
var tasks = commands.Select(cmd =>
{
FlowDelegate noop = _ => Task.CompletedTask;
return cmd.InvokeAsync(ctx, noop);
});
await Task.WhenAll(tasks);
sw.Stop();
var names = string.Join(" + ", commands.Select(c => c.Name));
ctx.Trace.Add(Name, $"parallel: {names}", sw.Elapsed, !ctx.Failed);
await next(ctx);
}
}
/// <summary>
/// Mutowalny kontekst przepływu — każde polecenie czyta/pisze swoje pola.
/// Jeśli Error != null, kolejne polecenia powinny się pominąć.
/// </summary>
public class PipelineContext
{
public required string FilePath { get; init; }
// Wyniki poszczególnych kroków — nullable bo wypełniane stopniowo
public FileContext? FileContext { get; set; }
public ParsedFile? ParsedFile { get; set; }
public ValidatedFile? ValidatedFile { get; set; }
public List<MappedEntity>? Entities { get; set; }
public ImportResult? ImportResult { get; set; }
// Błąd — jeśli ustawiony, pipeline się zatrzymuje
public string? Error { get; set; }
public bool Failed => Error is not null;
// Trzecia ścieżka — brak elementów do przetworzenia (nie błąd!)
public string? NoItemsReason { get; set; }
public bool IsEmpty => NoItemsReason is not null;
// Cokolwiek przerwało pipeline
public bool Halted => Failed || IsEmpty;
// Metadane do debugowania i raportowania
public PipelineTrace Trace { get; } = new();
// Dodatkowe dane z kroków schema-specific (np. OrdersTotal, CurrencyConverted)
public Dictionary<string, string> Metadata { get; } = [];
// Scoped DI provider — ustawiany przez ScopedFlow, dostępny dla sub-flow
public IServiceProvider? ScopedServices { get; set; }
}
Z wbudowaną obsługą statystyk, drukowania przepływu w konsoli i tworzenia pliku logu per nazwany pipeline dzięki Serilog,
oraz z Circuit Breakerem, który blokuje cały pipeline, jeśli było wcześniej więcej błędów niż "N"
/// <summary>
/// KafkaFlow middleware for batched inventory sync messages.
/// Uses AddBatching + WithManualMessageCompletion.
/// Commits the batch offset only on successful pipeline execution.
/// Register as Transient.
/// </summary>
public class InventorySyncBatchHandler : IMessageMiddleware
{
private readonly InventoryPipelineFactory _factory;
private readonly ILogger<InventorySyncBatchHandler> _logger;
private readonly PipelineLogsOptions _pipelineLogsOptions;
public InventorySyncBatchHandler(InventoryPipelineFactory factory,
ILogger<InventorySyncBatchHandler> logger,
PipelineLogsOptions pipelineLogsOptions)
{
_factory = factory;
_logger = logger;
_pipelineLogsOptions = pipelineLogsOptions;
}
public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
{
var batch = context.GetMessagesBatch();
_logger.LogInformation
("Received inventory sync batch: {Count} messages", batch.Count);
var syncMessages = new List<KafkaInventorySyncMessage>();
foreach (var msg in batch)
{
if (msg.Message.Value is KafkaInventorySyncMessage syncMsg)
{
syncMessages.Add(syncMsg);
}
else
{
_logger.LogWarning("Unexpected message type in inventory batch: {Type}",
msg.Message.Value?.GetType().Name ?? "null");
}
}
if (!syncMessages.Any())
{
_logger.LogWarning("No valid inventory sync messages in batch — skipping");
return;
}
var pipelineContext = BuildContext(syncMessages);
var name = _pipelineLogsOptions.Pipelines[nameof(InventorySyncBatchHandler)];
var result = await _factory.Create(name).ExecuteAsync(pipelineContext);
if (result.IsSuccess)
{
batch.Commit();
_logger.LogInformation(
"Inventory batch SUCCESS | BatchId: {Id} | Persisted: {P}/{T} | Warehouses: {W} | Duration: {D}ms",
pipelineContext.BatchId,
pipelineContext.PersistedItems.Count,
pipelineContext.Items.Count,
pipelineContext.ByWarehouse.Count,
result.ExecutionStats?.TotalDuration.TotalMilliseconds ?? 0);
}
else if (result.WasTerminatedEarly)
{
_logger.LogWarning("Inventory batch terminated early: {Reason}. Batch NOT committed.",
result.TerminationReason);
}
else
{
_logger.LogError(
"Inventory batch FAILED: {Error}. Batch NOT committed — {Count} messages will be re-delivered.",
result.ErrorMessage, batch.Count);
}
}
private static InventorySyncContext BuildContext(List<KafkaInventorySyncMessage> messages)
{
var context = new InventorySyncContext
{
ProcessingStartedAt = DateTime.UtcNow,
BatchSize = messages.Count
};
foreach (var msg in messages)
{
context.Items.Add(new ProcessableInventoryItem { Original = msg });
}
return context;
}
}
/// <summary>
/// Builds inventory sync pipelines on demand.
/// Register as Transient.
/// </summary>
public class InventoryPipelineFactory
{
private readonly GlobalStatisticsTracker _stats;
private readonly IServiceProvider _serviceProvider;
private readonly ILogger<InventoryPipelineFactory> _logger;
public InventoryPipelineFactory(
GlobalStatisticsTracker stats,
IServiceProvider serviceProvider,
ILogger<InventoryPipelineFactory> logger)
{
_stats = stats;
_serviceProvider = serviceProvider;
_logger = logger;
}
/// <summary>
/// Creates a fully built inventory sync pipeline with the given name.
/// Each call returns a distinct pipeline instance — use once per handler.
/// </summary>
public CommandPipeline<InventorySyncContext> Create(string pipelineName)
{
var pipeline = new CommandPipeline<InventorySyncContext>(
_stats, _serviceProvider, pipelineName);
pipeline.AddEventHandler(new KafkaPipelineEventHandler());
new CommandChainBuilder<InventorySyncContext>(pipeline)
.Then<ValidateInventoryCommand>()
.Then<AggregateInventoryCommand>()
.Then<PersistInventoryCommand>()
.Build();
_logger.LogInformation(
"[{Pipeline}] Inventory pipeline built: Validate → Aggregate → Persist",
pipelineName);
return pipeline;
}
}
using KafkaFlow;
using KafkaFlow.Serializer;
using KafkaPipeline.ConsumerDI;
using KafkaPipeline.ConsumerDI.Pipeline;
using KafkaPipeline.ConsumerDI.Pipeline.Handlers;
using KafkaPipeline.Shared;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Serilog;
using Serilog.Events;
namespace KafkaPipeline.ConsumerDI;
public class Program
{
private const string InventoryConsumerName = "inventory-consumer";
public static async Task Main(string[] args)
{
var configuration = new ConfigurationBuilder()
.SetBasePath(Directory.GetCurrentDirectory())
.AddJsonFile("appsettings.json", optional: false)
.AddEnvironmentVariables()
.Build();
var pipelineLogs = configuration
.GetSection(PipelineLogsOptions.Section)
.Get<PipelineLogsOptions>() ?? new PipelineLogsOptions();
Log.Logger = BuildLogger(pipelineLogs);
try
{
Log.Information("╔════════════════════════════════════════════════════════════════╗");
Log.Information("║ Kafka Pipeline Consumer — DI Edition ║");
Log.Information("║ All commands resolved from IServiceScope per execution ║");
Log.Information("║ Broker: localhost:19092 ║");
Log.Information("║ Topics: ║");
Log.Information("║ • {Topic,-46}║", KafkaTopics.OrdersInbound + " (linear pipeline)");
Log.Information("║ • {Topic,-46}║", KafkaTopics.PaymentsInbound + " (branching pipeline)");
Log.Information("║ • {Topic,-46}║", KafkaTopics.InventorySync + " (batch pipeline)");
Log.Information("╚════════════════════════════════════════════════════════════════╝");
var host = Host
.CreateDefaultBuilder(args)
.UseSerilog()
.ConfigureServices((context, services) =>
{
services.Configure<PipelineLogsOptions>(
context.Configuration.GetSection(PipelineLogsOptions.Section));
// Register all pipeline services, commands, repositories, handlers, factories
services.AddSingleton<PipelineLogsOptions>(pipelineLogs);
services.AddKafkaPipelineServices();
services.AddKafkaFlowHostedService(kafka => kafka
.UseMicrosoftLog()
.AddCluster(cluster => cluster
.WithBrokers(new[] { "localhost:19092" })
.CreateTopicIfNotExists(KafkaTopics.InventorySync, numberOfPartitions: 3, replicationFactor: 1)
// ── Consumer : Inventory Sync (batch pipeline) ───────────────────────
.AddConsumer(consumer => consumer
.Topic(KafkaTopics.InventorySync)
.WithGroupId(KafkaConsumerGroups.InventorySyncGroup)
.WithName(InventoryConsumerName)
.WithBufferSize(200)
.WithWorkersCount(1)
.WithAutoOffsetReset(AutoOffsetReset.Earliest)
.WithManualMessageCompletion()
.AddMiddlewares(middlewares => middlewares
.AddDeserializer<JsonCoreDeserializer>()
.AddBatching(50, TimeSpan.FromSeconds(5))
.Add<InventorySyncBatchHandler>()
)
)
)
);
})
.Build();
try
{
var inventoryFactory = host.Services.GetRequiredService<InventoryPipelineFactory>();
Log.Information("All pipeline factories built successfully");
}
catch (Exception ex)
{
Log.Fatal(ex, "Failed to build pipeline factories");
return;
}
await host.RunAsync();
}
catch (Exception ex)
{
Log.Fatal(ex, "Consumer terminated unexpectedly");
}
finally
{
Log.Information("Consumer shutting down...");
await Log.CloseAndFlushAsync();
}
}
private static ILogger BuildLogger(PipelineLogsOptions options)
{
var config = new LoggerConfiguration()
.MinimumLevel.Debug()
.MinimumLevel.Override("Microsoft", LogEventLevel.Warning)
.MinimumLevel.Override("KafkaFlow", LogEventLevel.Warning)
.Enrich.FromLogContext()
.WriteTo.Console(
outputTemplate: "[{Timestamp:HH:mm:ss.fff} {Level:u3}] {Message:lj}{NewLine}{Exception}")
.WriteTo.File(
path: Path.Combine(options.Directory, "log.txt"),
rollingInterval: RollingInterval.Hour,
outputTemplate: "{Timestamp:yyyy-MM-dd HH:mm:ss.fff} [{Level:u3}] [{CommandPipelineName}] [{ExecutionId}] {Message:lj}{NewLine}{Exception}");
// Per-pipeline sub-loggers — filter by CommandPipelineName enriched via LogContext
foreach (var pipelineName in options.Pipelines.Values.Distinct())
{
var name = pipelineName; // local copy for lambda closure
var logPath = Path.Combine(options.Directory, name, "log.txt");
config = config.WriteTo.Logger(lc => lc
.Filter.ByIncludingOnly(evt =>
evt.Properties.TryGetValue("CommandPipelineName", out var value)
&& value.ToString().Trim('"') == name)
.WriteTo.File(
path: logPath,
rollingInterval: RollingInterval.Hour,
outputTemplate: "{Timestamp:yyyy-MM-dd HH:mm:ss.fff} [{Level:u3}] [{ExecutionId}] {Message:lj}{NewLine}{Exception}"));
}
return config.CreateLogger();
}
}
public class InventorySyncContext
{
public string BatchId { get; set; } = Guid.NewGuid().ToString("N")[..8];
public DateTime ProcessingStartedAt { get; set; } = DateTime.UtcNow;
public int BatchSize { get; set; }
public List<ProcessableInventoryItem> Items { get; set; } = new();
public List<ProcessableInventoryItem> ValidatedItems { get; set; } = new();
public Dictionary<string, List<ProcessableInventoryItem>> ByWarehouse { get; set; } = new();
public List<ProcessableInventoryItem> PersistedItems { get; set; } = new();
public List<ProcessableInventoryItem> FailedItems { get; set; } = new();
}
public class InventorySyncContext
{
public string BatchId { get; set; } = Guid.NewGuid().ToString("N")[..8];
public DateTime ProcessingStartedAt { get; set; } = DateTime.UtcNow;
public int BatchSize { get; set; }
public List<ProcessableInventoryItem> Items { get; set; } = new();
public List<ProcessableInventoryItem> FailedItems { get; set; } = new();
}
OneOf<User, Error>
OneOf<User, NotFound, ValidationError>
OneOf<Success, Failure>
using OneOf;
namespace CsvImporter;
/// <summary>
/// Extension methods dla OneOf<TSuccess, TFailure, TEmpty> — trzy ścieżki:
/// T0 = sukces → kontynuuj łańcuch
/// T1 = failure → przerwij (błąd)
/// T2 = empty → przerwij (brak elementów, nie błąd)
///
/// Wzorowane na OneOf.Chaining, rozszerzone o trzeci wariant.
/// </summary>
public static class OneOfTripleExtensions
{
/// <summary>
/// Then — wykonaj następny krok jeśli sukces, inaczej propaguj failure/empty.
/// </summary>
public static async Task<OneOf<TSuccess, TFailure, TEmpty>> Then<TSuccess, TFailure, TEmpty>(
this Task<OneOf<TSuccess, TFailure, TEmpty>> currentResult,
Func<TSuccess, Task<OneOf<TSuccess, TFailure, TEmpty>>> nextJob)
{
var result = await currentResult;
if (result.IsT1) return result.AsT1;
if (result.IsT2) return result.AsT2;
return await nextJob(result.AsT0);
}
/// <summary>
/// Then (async-safe) — wersja poprawnie awaitująca.
/// </summary>
public static async Task<OneOf<TSuccess, TFailure, TEmpty>> ThenAsync<TSuccess, TFailure, TEmpty>(
this Task<OneOf<TSuccess, TFailure, TEmpty>> currentResult,
Func<TSuccess, Task<OneOf<TSuccess, TFailure, TEmpty>>> nextJob)
{
var result = await currentResult;
if (result.IsT1) return result.AsT1; // failure
if (result.IsT2) return result.AsT2; // empty
return await nextJob(result.AsT0);
}
/// <summary>
/// IfThen — wykonaj krok TYLKO gdy predykat == true i jest sukces.
/// </summary>
public static async Task<OneOf<TSuccess, TFailure, TEmpty>> IfThen<TSuccess, TFailure, TEmpty>(
this Task<OneOf<TSuccess, TFailure, TEmpty>> currentResult,
Func<TSuccess, bool> condition,
Func<TSuccess, Task<OneOf<TSuccess, TFailure, TEmpty>>> nextJob)
{
var result = await currentResult;
if (result.IsT1) return result.AsT1;
if (result.IsT2) return result.AsT2;
var success = result.AsT0;
if (!condition(success)) return success;
return await nextJob(success);
}
/// <summary>
/// Sync Then — dla kroków synchronicznych.
/// </summary>
public static async Task<OneOf<TSuccess, TFailure, TEmpty>> Then<TSuccess, TFailure, TEmpty>(
this Task<OneOf<TSuccess, TFailure, TEmpty>> currentResult,
Func<TSuccess, OneOf<TSuccess, TFailure, TEmpty>> nextJob)
{
var result = await currentResult;
if (result.IsT1) return result.AsT1;
if (result.IsT2) return result.AsT2;
return nextJob(result.AsT0);
}
}
using Microsoft.Extensions.DependencyInjection;
using CsvImporter.Commands;
namespace CsvImporter;
public static class CompositionRoot
{
public static IServiceProvider CreateServices()
{
var services = new ServiceCollection();
// Kroki pipeline'u — każdy jako serwis DI
services.AddScoped<FileCommands>();
services.AddScoped<ValidationCommands>();
services.AddScoped<MappingCommands>();
services.AddScoped<PersistenceCommands>();
// Schema-specific
services.AddScoped<OrderCommands>();
services.AddScoped<SalesCommands>();
services.AddScoped<UserCommands>();
services.AddScoped<ProductCommands>();
// Enrichment
services.AddScoped<EnrichmentCommands>();
return services.BuildServiceProvider();
}
}
using Microsoft.Extensions.DependencyInjection;
using CsvImporter;
using CsvImporter.Models;
using CsvImporter.Commands;
var services = CompositionRoot.CreateServices();
var input = SampleBatchFiles.Create();
ConsoleBatchReporter.PrintStartup(input);
// ═══════════════════════════════════════════════════
// PIPELINE — cały przepływ w jednym miejscu
// ═══════════════════════════════════════════════════
foreach (var path in input.ExistingPaths)
{
Console.WriteLine($"\n Plik: {Path.GetFileName(path)}");
await using var scope = services.CreateAsyncScope();
var sp = scope.ServiceProvider;
var file = sp.GetRequiredService<FileCommands>();
var validation = sp.GetRequiredService<ValidationCommands>();
var mapping = sp.GetRequiredService<MappingCommands>();
var persistence = sp.GetRequiredService<PersistenceCommands>();
var orders = sp.GetRequiredService<OrderCommands>();
var sales = sp.GetRequiredService<SalesCommands>();
var users = sp.GetRequiredService<UserCommands>();
var products = sp.GetRequiredService<ProductCommands>();
var enrichment = sp.GetRequiredService<EnrichmentCommands>();
var result = await ImportContext.Create(path)
// wspólne kroki wejściowe
.ThenAsync(file.OpenFile)
.ThenAsync(file.ParseCsv)
.ThenAsync(validation.ValidateRows) // → może: NoItemsToProcess
.ThenAsync(validation.Deduplicate) // → może: NoItemsToProcess
// rozgałęzienie per schemat
.IfThen(ctx => ctx.Schema == "Orders", orders.CalcTotals)
.IfThen(ctx => ctx.Schema == "Sales", sales.ConvertCurrency)
.IfThen(ctx => ctx.Schema == "Users", users.ValidateEmails) // → może: NoItemsToProcess
.IfThen(ctx => ctx.Schema == "Products", products.ValidateStock)
// wspólne kroki wyjściowe
.ThenAsync(mapping.MapEntities)
.ThenAsync(enrichment.GeoLookup)
.ThenAsync(enrichment.AuditStamp)
.ThenAsync(persistence.SaveBatch);
ConsoleBatchReporter.PrintItemResult(path, result);
}
using System.Diagnostics;
using CsvImporter.Models;
namespace CsvImporter.Commands;
public class FileCommands
{
private static readonly string[] SupportedExtensions = [".csv", ".tsv"];
private static readonly Dictionary<string, char> DelimiterHints = new() { { ".tsv", '\t' } };
private static readonly string[][] KnownSchemas =
[
["id", "name", "email"],
["id", "product", "quantity", "price"],
["id", "title", "category", "stock"],
["date", "region", "amount", "currency"]
];
private static readonly string[] SchemaNames = ["Users", "Orders", "Products", "Sales"];
public Task<PipelineResult> OpenFile(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
if (!File.Exists(ctx.FilePath))
return Task.FromResult<PipelineResult>(
new FileNotFoundFailure(ctx.FilePath));
ctx.DetectedFile = new FileInfo(ctx.FilePath);
ctx.Extension = ctx.DetectedFile.Extension.ToLowerInvariant();
if (!SupportedExtensions.Contains(ctx.Extension))
return Task.FromResult<PipelineResult>(
new UnsupportedTypeFailure(ctx.Extension));
var bytes = File.ReadAllBytes(ctx.FilePath);
ctx.Encoding = DetectEncoding(bytes);
if (DelimiterHints.TryGetValue(ctx.Extension, out var hint))
ctx.Delimiter = hint;
else
{
var firstLine = File.ReadLines(ctx.FilePath).FirstOrDefault() ?? "";
var delim = InferDelimiter(firstLine);
if (!delim.HasValue)
return Task.FromResult<PipelineResult>(
new DelimiterFailure());
ctx.Delimiter = delim.Value;
}
ctx.Header = File.ReadLines(ctx.FilePath)
.FirstOrDefault()?.Split(ctx.Delimiter)
.Select(c => c.Trim().ToLowerInvariant()).ToArray() ?? [];
for (var i = 0; i < KnownSchemas.Length; i++)
{
if (KnownSchemas[i].All(col => ctx.Header.Contains(col)))
{
ctx.Schema = SchemaNames[i];
sw.Stop();
ctx.AddTrace("OpenFile", $"schema={ctx.Schema}, delim='{ctx.Delimiter}'", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
return Task.FromResult<PipelineResult>(
new UnknownSchemaFailure(ctx.Header));
}
public Task<PipelineResult> ParseCsv(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var lines = File.ReadAllLines(ctx.FilePath);
if (lines.Length < 2)
return Task.FromResult<PipelineResult>(
new EmptyFileFailure());
ctx.ParsedRows = lines.Skip(1)
.Select((line, i) => new CsvRow(i + 2, line.Split(ctx.Delimiter))).ToList();
sw.Stop();
ctx.AddTrace("ParseCsv", $"{ctx.ParsedRows.Count} data rows", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
private static string DetectEncoding(byte[] bytes)
{
if (bytes.Length >= 3 && bytes[0] == 0xEF && bytes[1] == 0xBB && bytes[2] == 0xBF) return "UTF-8 BOM";
if (bytes.Length >= 2 && bytes[0] == 0xFF && bytes[1] == 0xFE) return "UTF-16 LE";
return bytes.Any(b => b > 0x7F) ? "UTF-8 (extended)" : "UTF-8";
}
private static char? InferDelimiter(string line)
{
char[] candidates = [',', ';', '\t', '|'];
return candidates.OrderByDescending(c => line.Count(ch => ch == c))
.FirstOrDefault(c => line.Contains(c));
}
}
using System.Diagnostics;
using CsvImporter.Models;
namespace CsvImporter.Commands;
public class ValidationCommands
{
public Task<PipelineResult> ValidateRows(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var headerCount = ctx.Header.Length;
var errors = new List<RowError>();
var valid = new List<CsvRow>();
foreach (var row in ctx.ParsedRows)
{
var rowErrors = new List<string>();
if (row.Fields.Length != headerCount)
rowErrors.Add($"Expected {headerCount} columns, got {row.Fields.Length}");
var empty = row.Fields.Select((f, i) => (f, i))
.Where(x => string.IsNullOrWhiteSpace(x.f))
.Select(x => x.i + 1).ToList();
if (empty.Count > 0)
rowErrors.Add($"Empty field(s) at columns: {string.Join(", ", empty)}");
if (rowErrors.Count > 0)
errors.AddRange(rowErrors.Select(e => new RowError(row.LineNumber, e)));
else
valid.Add(row);
}
ctx.ValidRows = valid;
ctx.ValidationErrors = errors;
sw.Stop();
ctx.AddTrace("ValidateRows",
errors.Count > 0 ? $"{valid.Count} ok, {errors.Count} errors" : $"all {valid.Count} valid",
sw.Elapsed);
// TUTAJ: jeśli walidacja odrzuciła WSZYSTKIE wiersze → NoItemsToProcess
if (valid.Count == 0)
return Task.FromResult<PipelineResult>(
new NoItemsToProcess("ValidateRows",
$"All {ctx.ParsedRows.Count} rows failed validation ({errors.Count} errors)",
ctx));
return Task.FromResult<PipelineResult>(ctx);
}
public Task<PipelineResult> Deduplicate(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var before = ctx.ValidRows.Count;
ctx.ValidRows = ctx.ValidRows
.GroupBy(r => string.Join("|", r.Fields))
.Select(g => g.First()).ToList();
var removed = before - ctx.ValidRows.Count;
sw.Stop();
ctx.AddTrace("Deduplicate", $"{ctx.ValidRows.Count} unique ({removed} removed)", sw.Elapsed);
// Deduplikacja usunęła wszystko → NoItemsToProcess
if (ctx.ValidRows.Count == 0)
return Task.FromResult<PipelineResult>(
new NoItemsToProcess("Deduplicate",
$"All {before} rows were duplicates", ctx));
return Task.FromResult<PipelineResult>(ctx);
}
}
public class MappingCommands
{
public Task<PipelineResult> MapEntities(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var header = File.ReadLines(ctx.FilePath)
.First().Split(ctx.Delimiter).Select(h => h.Trim()).ToArray();
ctx.Entities = ctx.ValidRows.Select(row =>
{
var fields = header.Zip(row.Fields.Select(f => f.Trim()))
.ToDictionary(p => p.First, p => p.Second);
return new MappedEntity(
fields.GetValueOrDefault("id", Guid.NewGuid().ToString()), fields);
}).ToList();
sw.Stop();
ctx.AddTrace("MapEntities", $"{ctx.Entities.Count} entities → '{ctx.Schema}'", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class PersistenceCommands
{
public async Task<PipelineResult> SaveBatch(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
await Task.Delay(30);
ctx.Result = new ImportResult(ctx.FilePath, ctx.Entities.Count, ctx.ValidationErrors.Count);
sw.Stop();
ctx.AddTrace("SaveBatch", $"{ctx.Result.RowsSaved} saved, {ctx.Result.RowsSkipped} skipped", sw.Elapsed);
return ctx;
}
}
using System.Diagnostics;
using CsvImporter.Models;
namespace CsvImporter.Commands;
public class OrderCommands
{
public Task<PipelineResult> CalcTotals(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var qtyIdx = Array.IndexOf(ctx.Header, "quantity");
var priceIdx = Array.IndexOf(ctx.Header, "price");
var total = 0m;
foreach (var row in ctx.ValidRows)
if (decimal.TryParse(row.Fields[qtyIdx], out var qty) &&
decimal.TryParse(row.Fields[priceIdx], out var price))
total += qty * price;
ctx.Metadata["OrdersTotal"] = total.ToString("F2");
sw.Stop();
ctx.AddTrace("CalcOrderTotals", $"total={total:F2}", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class SalesCommands
{
private static readonly Dictionary<string, decimal> Rates = new()
{ { "USD", 1.0m }, { "EUR", 1.08m }, { "GBP", 1.27m }, { "PLN", 0.25m } };
public Task<PipelineResult> ConvertCurrency(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var amountIdx = Array.IndexOf(ctx.Header, "amount");
var currencyIdx = Array.IndexOf(ctx.Header, "currency");
var converted = 0;
foreach (var row in ctx.ValidRows)
{
var currency = row.Fields[currencyIdx].Trim().ToUpperInvariant();
if (currency != "USD" && decimal.TryParse(row.Fields[amountIdx], out var amount) &&
Rates.TryGetValue(currency, out var rate))
{
var fields = row.Fields.ToArray();
fields[amountIdx] = (amount * rate).ToString("F2");
fields[currencyIdx] = "USD";
ctx.ValidRows[ctx.ValidRows.IndexOf(row)] = new CsvRow(row.LineNumber, fields);
converted++;
}
}
ctx.Metadata["CurrencyConverted"] = converted.ToString();
sw.Stop();
ctx.AddTrace("ConvertCurrency", $"{converted} rows → USD", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class UserCommands
{
public Task<PipelineResult> ValidateEmails(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var emailIdx = Array.IndexOf(ctx.Header, "email");
var valid = new List<CsvRow>();
foreach (var row in ctx.ValidRows)
{
var email = row.Fields[emailIdx].Trim();
if (!email.Contains('@') || !email.Contains('.'))
ctx.ValidationErrors.Add(new RowError(row.LineNumber, $"Invalid email: '{email}'"));
else
valid.Add(row);
}
ctx.ValidRows = valid;
sw.Stop();
ctx.AddTrace("ValidateEmails", $"{valid.Count} valid emails", sw.Elapsed);
// Wszystkie emaile niepoprawne → NoItemsToProcess
if (valid.Count == 0)
return Task.FromResult<PipelineResult>(
new NoItemsToProcess("ValidateEmails", "All emails invalid", ctx));
return Task.FromResult<PipelineResult>(ctx);
}
}
public class ProductCommands
{
public Task<PipelineResult> ValidateStock(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var stockIdx = Array.IndexOf(ctx.Header, "stock");
var corrected = 0;
foreach (var row in ctx.ValidRows)
if (int.TryParse(row.Fields[stockIdx].Trim(), out var stock) && stock < 0)
{
var fields = row.Fields.ToArray();
fields[stockIdx] = "0";
ctx.ValidRows[ctx.ValidRows.IndexOf(row)] = new CsvRow(row.LineNumber, fields);
corrected++;
}
ctx.Metadata["StockCorrected"] = corrected.ToString();
sw.Stop();
ctx.AddTrace("ValidateStock", $"{corrected} negative stocks → 0", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class EnrichmentCommands
{
public async Task<PipelineResult> GeoLookup(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var regions = new Dictionary<string, string>
{ { "North", "52.23°N" }, { "South", "50.06°N" }, { "East", "51.25°N" }, { "West", "51.10°N" } };
await Task.Delay(20);
var matched = 0;
foreach (var e in ctx.Entities)
if (e.Fields.TryGetValue("region", out var r) && regions.TryGetValue(r, out var c))
{ e.Fields["geo_coords"] = c; matched++; }
ctx.Metadata["GeoMatched"] = matched.ToString();
sw.Stop();
ctx.AddTrace("GeoLookup", $"{matched} geocoded", sw.Elapsed);
return ctx;
}
public Task<PipelineResult> AuditStamp(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var stamp = DateTime.UtcNow.ToString("O");
foreach (var e in ctx.Entities)
{ e.Fields["_imported_at"] = stamp; e.Fields["_source_file"] = Path.GetFileName(ctx.FilePath); }
ctx.Metadata["AuditStamp"] = stamp;
sw.Stop();
ctx.AddTrace("AuditStamp", $"stamped {ctx.Entities.Count} entities", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
using CsvImporter.Models;
namespace CsvImporter;
public static class ConsoleBatchReporter
{
public static void PrintStartup(BatchInput input)
{
Console.WriteLine("╔══════════════════════════════════════════════╗");
Console.WriteLine("║ CSV Batch Importer [OneOf.Chaining] ║");
Console.WriteLine("╚══════════════════════════════════════════════╝");
if (input.MissingPaths.Length > 0)
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine("\nPominięte (nie znaleziono):");
foreach (var missing in input.MissingPaths)
Console.WriteLine($" ✗ {Path.GetFileName(missing)}");
Console.ResetColor();
}
Console.WriteLine($"\nBatch: {input.ExistingPaths.Length} plików\n");
Console.WriteLine("─────────────────────────────────────────────");
}
public static void PrintItemResult(string path, PipelineResult result)
{
result.Switch(
ctx =>
{
PrintTrace(ctx.Trace, ConsoleColor.Green);
var skip = ctx.Result!.RowsSkipped > 0 ? $" ({ctx.Result.RowsSkipped} pominięto)" : "";
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine($" + OK: {ctx.Result.RowsSaved} wierszy{skip}");
Console.ResetColor();
},
failure =>
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($" ✗ {Path.GetFileName(path)}: {failure}");
Console.ResetColor();
},
noItems =>
{
PrintTrace(noItems.Context.Trace, ConsoleColor.DarkGray);
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine($" ○ PUSTE: {noItems.Reason}");
Console.ResetColor();
});
}
private static void PrintTrace(IEnumerable<TraceEntry> trace, ConsoleColor color)
{
foreach (var entry in trace)
{
Console.ForegroundColor = color;
Console.Write($" ✓ {entry.Step,-22}");
Console.ResetColor();
Console.WriteLine($" {entry.Duration.TotalMilliseconds,6:F1}ms {entry.Detail}");
}
}
}
using System.Diagnostics;
namespace CsvBatchImporterCSharp15Union.Models;
/// <summary>
/// Natywny union type z C# 15 preview.
/// Zastępuje OneOf<ImportContext, Failure, NoItemToProcess>.
/// </summary>
public union PipelineResult(ImportContext, Failure, NoItemToProcess);
/// <summary>
/// Obiekt stanu przepływu — niesiony przez cały łańcuch Then().
///
/// Trzy możliwe wyniki:
/// ImportContext — sukces, kontynuuj
/// Failure — błąd, przerwij
/// NoItemToProcess — pusta kolekcja po filtrze, przerwij (ale to nie błąd)
/// </summary>
public record ImportContext
{
public required string FilePath { get; init; }
public FileInfo? DetectedFile { get; set; }
public string Extension { get; set; } = "";
public string Encoding { get; set; } = "UTF-8";
public char Delimiter { get; set; } = ',';
public string Schema { get; set; } = "Unknown";
public string[] Header { get; set; } = [];
public List<CsvRow> ParsedRows { get; set; } = [];
public List<CsvRow> ValidRows { get; set; } = [];
public List<RowError> ValidationErrors { get; set; } = [];
public List<MappedEntity> Entities { get; set; } = [];
public ImportResult? Result { get; set; }
public Dictionary<string, string> Metadata { get; set; } = [];
public List<TraceEntry> Trace { get; set; } = [];
public void AddTrace(string step, string detail, TimeSpan duration)
=> Trace.Add(new TraceEntry(step, detail, duration));
/// <summary>Punkt wejścia łańcucha — tworzy kontekst opakowany w PipelineResult.</summary>
public static Task<PipelineResult> Create(string filePath)
=> Task.FromResult<PipelineResult>(new ImportContext { FilePath = filePath });
}
public record CsvRow(int LineNumber, string[] Fields);
public record RowError(int LineNumber, string Reason);
public record MappedEntity(string Id, Dictionary<string, string> Fields);
public record ImportResult(string FilePath, int RowsSaved, int RowsSkipped);
public record TraceEntry(string Step, string Detail, TimeSpan Duration);
public record FileError(string Path, string Reason);
public record BatchReport(int Total, int Succeeded, int Failed,
List<ImportResult> Imports, List<FileError> Errors);
namespace CsvBatchImporterCSharp15Union.Models;
/// <summary>
/// Bazowy typ błędu — wariant Failure w PipelineResult.
/// Pierwszy Failure zwrócony przez krok przerywa cały łańcuch.
/// </summary>
public record Failure(string Step, string Reason)
{
public override string ToString() => $"[{Step}] {Reason}";
}
// Konkretne typy — można matchować pattern matchingiem
public record FileNotFoundFailure(string Path)
: Failure("OpenFile", $"File not found: {Path}");
public record UnsupportedTypeFailure(string Extension)
: Failure("DetectType", $"Unsupported file type: '{Extension}'");
public record DelimiterFailure()
: Failure("DetectDelimiter", "Cannot detect delimiter. Tried: , ; \\t |");
public record UnknownSchemaFailure(string[] Columns)
: Failure("DetectSchema", $"Unknown schema. Columns: [{string.Join(", ", Columns)}]");
public record EmptyFileFailure()
: Failure("ParseCsv", "File is empty or has only a header row.");
public record ValidationFailure(int ErrorCount)
: Failure("ValidateRows", $"{ErrorCount} validation errors");
public record MappingFailure(string Detail)
: Failure("MapEntities", Detail);
public record SaveFailure(string Detail)
: Failure("SaveBatch", Detail);
// ── Trzecia ścieżka — ani błąd, ani sukces ──────────────────────
/// <summary>
/// Filtrowanie usunęło wszystkie wiersze — dalszy przepływ nie ma sensu.
/// To NIE jest błąd (plik był poprawny), to informacja o pustym wyniku.
/// </summary>
public record NoItemToProcess(string Step, string Reason, ImportContext Context)
{
public override string ToString() => $"[{Step}] {Reason}";
}
using Microsoft.Extensions.DependencyInjection;
using CsvBatchImporterCSharp15Union;
using CsvBatchImporterCSharp15Union.Commands;
using CsvBatchImporterCSharp15Union.Models;
using CsvBatchImporterCSharp15Union.Pipeline;
var services = CompositionRoot.CreateServices();
var input = SampleBatchFiles.Create();
ConsoleBatchReporter.PrintStartup(input);
// ═══════════════════════════════════════════════════
// PIPELINE — cały przepływ w jednym miejscu
// ═══════════════════════════════════════════════════
//
// Trzy ścieżki wynikowe:
// ImportContext → sukces, plik zaimportowany
// Failure → błąd (zły format, nieznany schemat, ...)
// NoItemToProcess → plik poprawny, ale 0 wierszy po filtrowaniu
//
foreach (var path in input.ExistingPaths)
{
Console.WriteLine($"\n Plik: {Path.GetFileName(path)}");
await using var scope = services.CreateAsyncScope();
var sp = scope.ServiceProvider;
var file = sp.GetRequiredService<FileCommands>();
var validation = sp.GetRequiredService<ValidationCommands>();
var mapping = sp.GetRequiredService<MappingCommands>();
var persistence = sp.GetRequiredService<PersistenceCommands>();
var orders = sp.GetRequiredService<OrderCommands>();
var sales = sp.GetRequiredService<SalesCommands>();
var users = sp.GetRequiredService<UserCommands>();
var products = sp.GetRequiredService<ProductCommands>();
var enrichment = sp.GetRequiredService<EnrichmentCommands>();
var result = await ImportContext.Create(path)
// wspólne kroki wejściowe
.ThenAsync(file.OpenFile)
.ThenAsync(file.ParseCsv)
.ThenAsync(validation.ValidateRows) // → może: NoItemToProcess
.ThenAsync(validation.Deduplicate) // → może: NoItemToProcess
// rozgałęzienie per schemat
.IfThen(ctx => ctx.Schema == "Orders", orders.CalcTotals)
.IfThen(ctx => ctx.Schema == "Sales", sales.ConvertCurrency)
.IfThen(ctx => ctx.Schema == "Users", users.ValidateEmails) // → może: NoItemToProcess
.IfThen(ctx => ctx.Schema == "Products", products.ValidateStock)
// wspólne kroki wyjściowe
.ThenAsync(mapping.MapEntities)
.ThenWaitForAll("Enrichment", enrichment.GeoLookup, enrichment.AuditStamp)
.ThenAsync(persistence.SaveBatch);
ConsoleBatchReporter.PrintItemResult(path, result);
}
using System.Diagnostics;
using CsvBatchImporterCSharp15Union.Models;
namespace CsvBatchImporterCSharp15Union.Commands;
public class FileCommands
{
private static readonly string[] SupportedExtensions = [".csv", ".tsv"];
private static readonly Dictionary<string, char> DelimiterHints = new() { { ".tsv", '\t' } };
private static readonly string[][] KnownSchemas =
[
["id", "name", "email"],
["id", "product", "quantity", "price"],
["id", "title", "category", "stock"],
["date", "region", "amount", "currency"]
];
private static readonly string[] SchemaNames = ["Users", "Orders", "Products", "Sales"];
public Task<PipelineResult> OpenFile(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
if (!File.Exists(ctx.FilePath))
return Task.FromResult<PipelineResult>(
new FileNotFoundFailure(ctx.FilePath));
ctx.DetectedFile = new FileInfo(ctx.FilePath);
ctx.Extension = ctx.DetectedFile.Extension.ToLowerInvariant();
if (!SupportedExtensions.Contains(ctx.Extension))
return Task.FromResult<PipelineResult>(
new UnsupportedTypeFailure(ctx.Extension));
var bytes = File.ReadAllBytes(ctx.FilePath);
ctx.Encoding = DetectEncoding(bytes);
if (DelimiterHints.TryGetValue(ctx.Extension, out var hint))
ctx.Delimiter = hint;
else
{
var firstLine = File.ReadLines(ctx.FilePath).FirstOrDefault() ?? "";
var delim = InferDelimiter(firstLine);
if (!delim.HasValue)
return Task.FromResult<PipelineResult>(
new DelimiterFailure());
ctx.Delimiter = delim.Value;
}
ctx.Header = File.ReadLines(ctx.FilePath)
.FirstOrDefault()?.Split(ctx.Delimiter)
.Select(c => c.Trim().ToLowerInvariant()).ToArray() ?? [];
for (var i = 0; i < KnownSchemas.Length; i++)
{
if (KnownSchemas[i].All(col => ctx.Header.Contains(col)))
{
ctx.Schema = SchemaNames[i];
sw.Stop();
ctx.AddTrace("OpenFile", $"schema={ctx.Schema}, delim='{ctx.Delimiter}'", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
return Task.FromResult<PipelineResult>(
new UnknownSchemaFailure(ctx.Header));
}
public Task<PipelineResult> ParseCsv(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var lines = File.ReadAllLines(ctx.FilePath);
if (lines.Length < 2)
return Task.FromResult<PipelineResult>(
new EmptyFileFailure());
ctx.ParsedRows = lines.Skip(1)
.Select((line, i) => new CsvRow(i + 2, line.Split(ctx.Delimiter))).ToList();
sw.Stop();
ctx.AddTrace("ParseCsv", $"{ctx.ParsedRows.Count} data rows", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
private static string DetectEncoding(byte[] bytes)
{
if (bytes.Length >= 3 && bytes[0] == 0xEF && bytes[1] == 0xBB && bytes[2] == 0xBF) return "UTF-8 BOM";
if (bytes.Length >= 2 && bytes[0] == 0xFF && bytes[1] == 0xFE) return "UTF-16 LE";
return bytes.Any(b => b > 0x7F) ? "UTF-8 (extended)" : "UTF-8";
}
private static char? InferDelimiter(string line)
{
char[] candidates = [',', ';', '\t', '|'];
return candidates.OrderByDescending(c => line.Count(ch => ch == c))
.FirstOrDefault(c => line.Contains(c));
}
}
using System.Diagnostics;
using CsvBatchImporterCSharp15Union.Models;
namespace CsvBatchImporterCSharp15Union.Commands;
public class ValidationCommands
{
public Task<PipelineResult> ValidateRows(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var headerCount = ctx.Header.Length;
var errors = new List<RowError>();
var valid = new List<CsvRow>();
foreach (var row in ctx.ParsedRows)
{
var rowErrors = new List<string>();
if (row.Fields.Length != headerCount)
rowErrors.Add($"Expected {headerCount} columns, got {row.Fields.Length}");
var empty = row.Fields.Select((f, i) => (f, i))
.Where(x => string.IsNullOrWhiteSpace(x.f))
.Select(x => x.i + 1).ToList();
if (empty.Count > 0)
rowErrors.Add($"Empty field(s) at columns: {string.Join(", ", empty)}");
if (rowErrors.Count > 0)
errors.AddRange(rowErrors.Select(e => new RowError(row.LineNumber, e)));
else
valid.Add(row);
}
ctx.ValidRows = valid;
ctx.ValidationErrors = errors;
sw.Stop();
ctx.AddTrace("ValidateRows",
errors.Count > 0 ? $"{valid.Count} ok, {errors.Count} errors" : $"all {valid.Count} valid",
sw.Elapsed);
// TUTAJ: jeśli walidacja odrzuciła WSZYSTKIE wiersze → NoItemToProcess
if (valid.Count == 0)
return Task.FromResult<PipelineResult>(
new NoItemToProcess("ValidateRows",
$"All {ctx.ParsedRows.Count} rows failed validation ({errors.Count} errors)",
ctx));
return Task.FromResult<PipelineResult>(ctx);
}
public Task<PipelineResult> Deduplicate(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var before = ctx.ValidRows.Count;
ctx.ValidRows = ctx.ValidRows
.GroupBy(r => string.Join("|", r.Fields))
.Select(g => g.First()).ToList();
var removed = before - ctx.ValidRows.Count;
sw.Stop();
ctx.AddTrace("Deduplicate", $"{ctx.ValidRows.Count} unique ({removed} removed)", sw.Elapsed);
// Deduplikacja usunęła wszystko → NoItemToProcess
if (ctx.ValidRows.Count == 0)
return Task.FromResult<PipelineResult>(
new NoItemToProcess("Deduplicate",
$"All {before} rows were duplicates", ctx));
return Task.FromResult<PipelineResult>(ctx);
}
}
public class MappingCommands
{
public Task<PipelineResult> MapEntities(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var header = File.ReadLines(ctx.FilePath)
.First().Split(ctx.Delimiter).Select(h => h.Trim()).ToArray();
ctx.Entities = ctx.ValidRows.Select(row =>
{
var fields = header.Zip(row.Fields.Select(f => f.Trim()))
.ToDictionary(p => p.First, p => p.Second);
return new MappedEntity(
fields.GetValueOrDefault("id", Guid.NewGuid().ToString()), fields);
}).ToList();
sw.Stop();
ctx.AddTrace("MapEntities", $"{ctx.Entities.Count} entities → '{ctx.Schema}'", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class PersistenceCommands
{
public async Task<PipelineResult> SaveBatch(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
await Task.Delay(30);
ctx.Result = new ImportResult(ctx.FilePath, ctx.Entities.Count, ctx.ValidationErrors.Count);
sw.Stop();
ctx.AddTrace("SaveBatch", $"{ctx.Result.RowsSaved} saved, {ctx.Result.RowsSkipped} skipped", sw.Elapsed);
return ctx;
}
}
using System.Diagnostics;
using CsvBatchImporterCSharp15Union.Models;
namespace CsvBatchImporterCSharp15Union.Commands;
public class OrderCommands
{
public Task<PipelineResult> CalcTotals(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var qtyIdx = Array.IndexOf(ctx.Header, "quantity");
var priceIdx = Array.IndexOf(ctx.Header, "price");
var total = 0m;
foreach (var row in ctx.ValidRows)
if (decimal.TryParse(row.Fields[qtyIdx], out var qty) &&
decimal.TryParse(row.Fields[priceIdx], out var price))
total += qty * price;
ctx.Metadata["OrdersTotal"] = total.ToString("F2");
sw.Stop();
ctx.AddTrace("CalcOrderTotals", $"total={total:F2}", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class SalesCommands
{
private static readonly Dictionary<string, decimal> Rates = new()
{ { "USD", 1.0m }, { "EUR", 1.08m }, { "GBP", 1.27m }, { "PLN", 0.25m } };
public Task<PipelineResult> ConvertCurrency(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var amountIdx = Array.IndexOf(ctx.Header, "amount");
var currencyIdx = Array.IndexOf(ctx.Header, "currency");
var converted = 0;
foreach (var row in ctx.ValidRows)
{
var currency = row.Fields[currencyIdx].Trim().ToUpperInvariant();
if (currency != "USD" && decimal.TryParse(row.Fields[amountIdx], out var amount) &&
Rates.TryGetValue(currency, out var rate))
{
var fields = row.Fields.ToArray();
fields[amountIdx] = (amount * rate).ToString("F2");
fields[currencyIdx] = "USD";
ctx.ValidRows[ctx.ValidRows.IndexOf(row)] = new CsvRow(row.LineNumber, fields);
converted++;
}
}
ctx.Metadata["CurrencyConverted"] = converted.ToString();
sw.Stop();
ctx.AddTrace("ConvertCurrency", $"{converted} rows → USD", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class UserCommands
{
public Task<PipelineResult> ValidateEmails(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var emailIdx = Array.IndexOf(ctx.Header, "email");
var valid = new List<CsvRow>();
foreach (var row in ctx.ValidRows)
{
var email = row.Fields[emailIdx].Trim();
if (!email.Contains('@') || !email.Contains('.'))
ctx.ValidationErrors.Add(new RowError(row.LineNumber, $"Invalid email: '{email}'"));
else
valid.Add(row);
}
ctx.ValidRows = valid;
sw.Stop();
ctx.AddTrace("ValidateEmails", $"{valid.Count} valid emails", sw.Elapsed);
// Wszystkie emaile niepoprawne → NoItemToProcess
if (valid.Count == 0)
return Task.FromResult<PipelineResult>(
new NoItemToProcess("ValidateEmails", "All emails invalid", ctx));
return Task.FromResult<PipelineResult>(ctx);
}
}
public class ProductCommands
{
public Task<PipelineResult> ValidateStock(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var stockIdx = Array.IndexOf(ctx.Header, "stock");
var corrected = 0;
foreach (var row in ctx.ValidRows)
if (int.TryParse(row.Fields[stockIdx].Trim(), out var stock) && stock < 0)
{
var fields = row.Fields.ToArray();
fields[stockIdx] = "0";
ctx.ValidRows[ctx.ValidRows.IndexOf(row)] = new CsvRow(row.LineNumber, fields);
corrected++;
}
ctx.Metadata["StockCorrected"] = corrected.ToString();
sw.Stop();
ctx.AddTrace("ValidateStock", $"{corrected} negative stocks → 0", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
public class EnrichmentCommands
{
public async Task<PipelineResult> GeoLookup(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var regions = new Dictionary<string, string>
{ { "North", "52.23°N" }, { "South", "50.06°N" }, { "East", "51.25°N" }, { "West", "51.10°N" } };
await Task.Delay(20);
var matched = 0;
foreach (var e in ctx.Entities)
if (e.Fields.TryGetValue("region", out var r) && regions.TryGetValue(r, out var c))
{ e.Fields["geo_coords"] = c; matched++; }
ctx.Metadata["GeoMatched"] = matched.ToString();
sw.Stop();
ctx.AddTrace("GeoLookup", $"{matched} geocoded", sw.Elapsed);
return ctx;
}
public Task<PipelineResult> AuditStamp(ImportContext ctx)
{
var sw = Stopwatch.StartNew();
var stamp = DateTime.UtcNow.ToString("O");
foreach (var e in ctx.Entities)
{ e.Fields["_imported_at"] = stamp; e.Fields["_source_file"] = Path.GetFileName(ctx.FilePath); }
ctx.Metadata["AuditStamp"] = stamp;
sw.Stop();
ctx.AddTrace("AuditStamp", $"stamped {ctx.Entities.Count} entities", sw.Elapsed);
return Task.FromResult<PipelineResult>(ctx);
}
}
using CsvBatchImporterCSharp15Union.Models;
namespace CsvBatchImporterCSharp15Union;
public static class ConsoleBatchReporter
{
public static void PrintStartup(BatchInput input)
{
Console.WriteLine("╔══════════════════════════════════════════════╗");
Console.WriteLine("║ CSV Batch Importer [C# 15 Union] ║");
Console.WriteLine("╚══════════════════════════════════════════════╝");
if (input.MissingPaths.Length > 0)
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine("\nPominięte (nie znaleziono):");
foreach (var missing in input.MissingPaths)
Console.WriteLine($" ✗ {Path.GetFileName(missing)}");
Console.ResetColor();
}
Console.WriteLine($"\nBatch: {input.ExistingPaths.Length} plików\n");
Console.WriteLine("─────────────────────────────────────────────");
}
public static void PrintItemResult(string path, PipelineResult result)
{
switch (result)
{
case ImportContext ctx:
{
PrintTrace(ctx.Trace, ConsoleColor.Green);
var skip = ctx.Result!.RowsSkipped > 0 ? $" ({ctx.Result.RowsSkipped} pominięto)" : "";
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine($" + OK: {ctx.Result.RowsSaved} wierszy{skip}");
Console.ResetColor();
break;
}
case Failure failure:
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($" ✗ {Path.GetFileName(path)}: [{failure.Step}] {failure.Reason}");
Console.ResetColor();
break;
}
case NoItemToProcess noItems:
{
PrintTrace(noItems.Context.Trace, ConsoleColor.DarkGray);
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine($" ○ PUSTE: {noItems.Reason}");
Console.ResetColor();
break;
}
}
}
private static void PrintTrace(IEnumerable<TraceEntry> trace, ConsoleColor color)
{
foreach (var entry in trace)
{
Console.ForegroundColor = color;
Console.Write($" ✓ {entry.Step,-22}");
Console.ResetColor();
Console.WriteLine($" {entry.Duration.TotalMilliseconds,6:F1}ms {entry.Detail}");
}
}
}
using CsvBatchImporterCSharp15Union.Models;
namespace CsvBatchImporterCSharp15Union.Pipeline;
/// <summary>
/// Extension methods dla PipelineResult union — trzy ścieżki:
/// ImportContext = sukces → kontynuuj łańcuch
/// Failure = failure → przerwij (błąd)
/// NoItemToProcess = empty → przerwij (brak elementów, nie błąd)
///
/// Wzorowane na monadycznym chainingu, ale bez zewnętrznego pakietu.
/// </summary>
public static class UnionPipelineExtensions
{
/// <summary>
/// Then — wykonaj następny krok jeśli sukces, inaczej propaguj failure/empty.
/// </summary>
public static async Task<PipelineResult> ThenAsync(
this Task<PipelineResult> currentResult,
Func<ImportContext, Task<PipelineResult>> nextJob)
{
var result = await currentResult;
return result switch
{
ImportContext ctx => await nextJob(ctx),
Failure failure => failure,
NoItemToProcess noItems => noItems,
null => throw new InvalidOperationException("PipelineResult cannot be null.")
};
}
/// <summary>
/// IfThen — wykonaj krok TYLKO gdy predykat == true i jest sukces.
/// </summary>
public static async Task<PipelineResult> IfThen(
this Task<PipelineResult> currentResult,
Func<ImportContext, bool> condition,
Func<ImportContext, Task<PipelineResult>> nextJob)
{
var result = await currentResult;
return result switch
{
ImportContext ctx when condition(ctx) => await nextJob(ctx),
ImportContext ctx => ctx,
Failure failure => failure,
NoItemToProcess noItems => noItems,
null => throw new InvalidOperationException("PipelineResult cannot be null.")
};
}
/// <summary>
/// ThenWaitForAll — wykonaj kilka niezależnych kroków równolegle jeśli bieżący wynik to sukces.
/// </summary>
public static async Task<PipelineResult> ThenWaitForAll(
this Task<PipelineResult> currentResult,
string name,
params Func<ImportContext, Task<PipelineResult>>[] nextJobs)
{
var result = await currentResult;
return result switch
{
ImportContext ctx => await RunAll(ctx, name, nextJobs),
Failure failure => failure,
NoItemToProcess noItems => noItems,
null => throw new InvalidOperationException("PipelineResult cannot be null.")
};
}
/// <summary>
/// Sync Then — dla kroków synchronicznych.
/// </summary>
public static async Task<PipelineResult> Then(
this Task<PipelineResult> currentResult,
Func<ImportContext, PipelineResult> nextJob)
{
var result = await currentResult;
return result switch
{
ImportContext ctx => nextJob(ctx),
Failure failure => failure,
NoItemToProcess noItems => noItems,
null => throw new InvalidOperationException("PipelineResult cannot be null.")
};
}
private static async Task<PipelineResult> RunAll(
ImportContext ctx,
string name,
Func<ImportContext, Task<PipelineResult>>[] nextJobs)
{
if (nextJobs.Length == 0)
return ctx;
var results = await Task.WhenAll(nextJobs.Select(nextJob => nextJob(ctx)));
foreach (var result in results)
{
switch (result)
{
case Failure failure:
return failure;
case NoItemToProcess noItems:
return noItems;
case ImportContext:
break;
case null:
throw new InvalidOperationException($"{name} returned null PipelineResult.");
}
}
return ctx;
}
}
using AiContentTeam;
using AiContentTeam.Guardrails;
using AiContentTeam.Middleware;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.DevUI;
using Microsoft.Agents.AI.Hosting;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using ModelContextProtocol.Client;
var builder = WebApplication.CreateBuilder(args);
// ── IChatClient ───────────────────────────────────────────────────────────────
IChatClient chatClient = new Anthropic.AnthropicClient()
.AsIChatClient("claude-haiku-4-5");
builder.Services.AddChatClient(chatClient);
// ── MCP tools ────────────────────────────────────────────────────────────────
builder.Services.AddSingleton<IList<McpClientTool>>(sp =>
{
return Task.Run(async () =>
{
var mcpClient = await McpClient.CreateAsync(
new HttpClientTransport(new HttpClientTransportOptions
{
Endpoint = new Uri("http://localhost:2001/mcp")
}));
return await mcpClient.ListToolsAsync();
}).GetAwaiter().GetResult();
});
// ── Skills provider ───────────────────────────────────────────────────────────
#pragma warning disable MAAI001
builder.Services.AddSingleton(
new AgentSkillsProvider(Path.Combine(AppContext.BaseDirectory, "skills")));
#pragma warning restore MAAI001
// ── Guardrails ────────────────────────────────────────────────────────────────
var contentPolicy = new ContentPolicyGuardrail();
var brandSafety = new BrandSafetyGuardrail(
competitors: ["CompetitorA", "CompetitorB"],
requireCTA: true);
var outputLength = new OutputLengthGuardrail(maxChars: 3000);
// ── Agenty w DI ───────────────────────────────────────────────────────────────
builder.Services.AddKeyedSingleton<AIAgent>("Researcher", (sp, _) =>
{
var chat = sp.GetRequiredService<IChatClient>();
var mcpTools = sp.GetRequiredService<IList<McpClientTool>>();
return GuardrailFactory.WrapWithFullPipeline(
new ChatClientAgent(chat, new ChatClientAgentOptions
{
Name = "Researcher",
ChatOptions = new()
{
Instructions = """
Jesteś research agentem. Twoim JEDYNYM zadaniem jest:
1. UŻYJ NARZĘDZIA GetYouTubeTranscript aby pobrać transkrypcję z podanego URL
2. Na podstawie wyniku wyodrębnij dokładnie 5 kluczowych tez jako bullet pointy
ZASADY:
- ZAWSZE wywołuj narzędzie GetYouTubeTranscript
- Odpowiadaj po polsku
- Każdą tezę pisz w osobnej linii zaczynającej się od "• "
""",
Tools = [.. mcpTools]
}
}),
enableLogging: true,
enableRetry: true);
});
builder.Services.AddKeyedSingleton<AIAgent>("Copywriter", (sp, _) =>
{
var chat = sp.GetRequiredService<IChatClient>();
var skills = sp.GetRequiredService<AgentSkillsProvider>();
return GuardrailFactory.WrapWithFullPipeline(
new ChatClientAgent(chat, new ChatClientAgentOptions
{
Name = "Copywriter",
AIContextProviders = [skills],
ChatOptions = new()
{
Instructions = """
Jesteś copywriterem specjalizującym się w treściach dla .NET developerów.
Na podstawie tez od Researchera napisz post na LinkedIn.
ZASADY:
- 150-250 słów, max 3 emoji
- Mocny hook w 1. zdaniu
- Zakończ pytaniem do społeczności
- Napisz TYLKO treść posta, bez meta-komentarza
- Odpowiadaj po polsku
"""
}
}),
enableLogging: true,
enableRetry: true,
contentPolicy, brandSafety, outputLength);
});
builder.Services.AddKeyedSingleton<AIAgent>("Strateg", (sp, _) =>
{
var chat = sp.GetRequiredService<IChatClient>();
var skills = sp.GetRequiredService<AgentSkillsProvider>();
return GuardrailFactory.WrapWithFullPipeline(
new ChatClientAgent(chat, new ChatClientAgentOptions
{
Name = "Strateg",
AIContextProviders = [skills],
ChatOptions = new()
{
Instructions = """
Jesteś strategiem social media dla społeczności .NET developerów.
Przeanalizuj temat i odpowiedz STRUKTURALNIE:
ANALIZA TRENDÓW: [2-3 zdania]
REKOMENDOWANY HOOK: [jedno zdanie]
SUGEROWANE HASHTAGI: [max 5, format #tag]
UWAGI O ZAANGAŻOWANIU: [1-2 zdania]
Odpowiadaj po polsku.
""",
Tools = [AIFunctionFactory.Create(GetLinkedInTrends)]
}
}),
enableLogging: true,
enableRetry: true,
outputLength);
});
builder.Services.AddKeyedSingleton<AIAgent>("Redaktor", (sp, _) =>
{
var chat = sp.GetRequiredService<IChatClient>();
return GuardrailFactory.WrapWithFullPipeline(
new ChatClientAgent(chat, new ChatClientAgentOptions
{
Name = "Redaktor",
ChatOptions = new()
{
Instructions = """
Jesteś redaktorem naczelnym. Dostałeś draft i analizę strategiczną.
Zastosuj rekomendacje, popraw błędy językowe.
Napisz TYLKO finalną treść posta po polsku.
"""
}
}),
enableLogging: true,
enableRetry: true,
contentPolicy, brandSafety, outputLength);
});
// ══════════════════════════════════════════════════════════════════════════════
// content-pipeline-dev — BEZ HITL, do DevUI
// ══════════════════════════════════════════════════════════════════════════════
builder.AddWorkflow("content-pipeline-dev", (sp, key) =>
{
var chat = sp.GetRequiredService<IChatClient>();
var researcherAgent = sp.GetRequiredKeyedService<AIAgent>("Researcher");
var copywriterAgent = sp.GetRequiredKeyedService<AIAgent>("Copywriter");
var strategAgent = sp.GetRequiredKeyedService<AIAgent>("Strateg");
var redaktorAgent = sp.GetRequiredKeyedService<AIAgent>("Redaktor");
// Entry point: ChatClientAgent (DevUI wymaga tego)
var entryAgent = new ChatClientAgent(chat, new ChatClientAgentOptions
{
Name = "InputParser",
ChatOptions = new()
{
Instructions = """
Jesteś parserem inputu. Użytkownik poda URL YouTube lub temat.
Wyodrębnij URL i powtórz go dokładnie.
Jeśli nie ma URL, powtórz temat dosłownie.
Nie dodawaj żadnych komentarzy ani wyjaśnień.
Odpowiedz JEDNYM zdaniem lub samym URL-em.
"""
}
});
// Bridge: string → ResearchRequest (bo ResearcherExecutor oczekuje ResearchRequest)
var inputBridge = new InputBridgeExecutor();
// Executory
var researcher = new ResearcherExecutor(researcherAgent);
var copywriter = new CopywriterExecutor(copywriterAgent);
var strateg = new StrategExecutor(strategAgent);
var aggregator = new AggregatorExecutor();
var redaktor = new RedaktorExecutor(redaktorAgent);
var publisher = new PublishExecutor();
IEnumerable<ExecutorBinding> fanInSources = [copywriter, strateg];
// Graf: prosty, bez HITL
// W workflow zamiast:
// entryAgent → inputBridge → researcher(executor)
// Daj:
// entryAgent → researcherAgent(ChatClientAgent) → copywriter + strateg
return new WorkflowBuilder(entryAgent)
.WithName(key)
.AddEdge(entryAgent, researcherAgent) // ← sam agent, nie executor
.AddFanOutEdge(researcherAgent, [copywriterAgent, strategAgent])
.AddFanInBarrierEdge(
(IEnumerable<ExecutorBinding>)[copywriterAgent, strategAgent],
redaktorAgent)
.WithOutputFrom(redaktorAgent)
.Build();
}).AddAsAIAgent();
// ── OpenAI + DevUI ────────────────────────────────────────────────────────────
builder.Services.AddOpenAIResponses();
builder.Services.AddOpenAIConversations();
builder.Logging.SetMinimumLevel(LogLevel.Debug);
// ── Build & Map ───────────────────────────────────────────────────────────────
var app = builder.Build();
// Middleware na wyjątki — PRZED endpointami
app.Use(async (context, next) =>
{
try
{
await next();
}
catch (Exception ex)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($"[EXCEPTION] {ex}");
Console.ResetColor();
throw;
}
});
app.MapOpenAIResponses();
app.MapOpenAIConversations();
if (app.Environment.IsDevelopment())
{
app.MapDevUI();
}
app.MapGet("/health", () => Results.Ok(new
{
status = "healthy",
pipeline = "AiContentTeam — DevUI",
devui = "/devui",
features = new[] { "No HITL (dev)", "Middleware", "Guardrails", "Fan-out/Fan-in" }
}));
// ── Start ─────────────────────────────────────────────────────────────────────
app.Urls.Add("http://localhost:5050");
Console.ForegroundColor = ConsoleColor.Cyan;
Console.WriteLine("╔════════════════════════════════════════════════════════════╗");
Console.WriteLine("║ 🚀 AiContentTeam — DevUI Server ║");
Console.WriteLine("║ ║");
Console.WriteLine("║ DevUI: http://localhost:5050/devui ║");
Console.WriteLine("║ Health: http://localhost:5050/health ║");
Console.WriteLine("║ ║");
Console.WriteLine("║ Workflow: content-pipeline-dev (no HITL) ║");
Console.WriteLine("║ ║");
Console.WriteLine("║ Pipeline: ║");
Console.WriteLine("║ InputParser → Bridge → Researcher ║");
Console.WriteLine("║ → Copywriter + Strateg (fan-out) ║");
Console.WriteLine("║ → Aggregator → Redaktor → Publisher ║");
Console.WriteLine("║ ║");
Console.WriteLine("║ Middleware: Logging, Retry, Guardrails, Function Audit ║");
Console.WriteLine("╚════════════════════════════════════════════════════════════╝");
Console.ResetColor();
app.Run();
// ── Tool helper ───────────────────────────────────────────────────────────────
[System.ComponentModel.Description("Sprawdza popularne trendy na LinkedIn w kategorii tech/AI")]
static string GetLinkedInTrends(
[System.ComponentModel.Description("Kategoria, np. 'AI', '.NET'")] string category)
{
return $"""
Top trendy w kategorii "{category}" (symulacja):
1. Microsoft Agent Framework — wysoka aktywność
2. .NET 10 Preview — umiarkowane zaangażowanie
3. Multi-agent orchestration — rosnący trend
""";
}
// ══════════════════════════════════════════════════════════════════════════════
// InputBridgeExecutor — konwertuje string z ChatClientAgent na ResearchRequest
// ══════════════════════════════════════════════════════════════════════════════
/// <summary>
/// Bridge między ChatClientAgent (zwraca string) a ResearcherExecutor (oczekuje ResearchRequest).
/// Potrzebny bo DevUI wymaga ChatClientAgent jako entry point workflow.
/// Produkcyjny Program.cs (konsolowy) nie potrzebuje tego — tam InputParserExecutor
/// jest entry pointem i bezpośrednio tworzy ResearchRequest.
/// </summary>
public partial class InputBridgeExecutor() : Executor<string, ResearchRequest>("InputBridge")
{
public override ValueTask<ResearchRequest> HandleAsync(
string message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine($"[InputBridge] Parsing: '{message}'");
return ValueTask.FromResult(new ResearchRequest(
YouTubeUrl: message.Trim(),
TargetAudience: "polscy .NET developerzy, seniorzy i architekci"));
}
}
Ale jeśli chcesz:
widzieć przepływ procesu w jednym miejscu, bez sterowania wyjątkami i bez eventów udających kontrolę flow, to już wiesz, jak to zrobić
Jeśli Twój proces ma:
Widoczny flow, jawne rezultaty, wymienne kroki, eventy tylko jako skutki uboczne, osobny batch/item context, to jesteś blisko czytelnego kodu.
