ADO.NET with Npgsql

using Npgsql;

public sealed record LineToReserve(string Sku, int Quantity);

public sealed class StockReservations(NpgsqlDataSource dataSource)
{
    public async Task<bool> ReserveAsync(IReadOnlyList<LineToReserve> lines, CancellationToken ct)
    {
        await using var connection = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await connection.BeginTransactionAsync(ct);
        await using var batch = new NpgsqlBatch(connection, tx);
        // Lock rows in one global order (by SKU) so two concurrent orders cannot deadlock.
        foreach (var line in lines.OrderBy(l => l.Sku, StringComparer.Ordinal))
        {
            batch.BatchCommands.Add(new NpgsqlBatchCommand(
                "UPDATE inventory SET reserved = reserved + $1 WHERE sku = $2 AND on_hand - reserved >= $1")
            {
                Parameters =
                {
                    new NpgsqlParameter<int> { TypedValue = line.Quantity },
                    new NpgsqlParameter<string> { TypedValue = line.Sku },
                },
            });
        }

        await batch.ExecuteNonQueryAsync(ct);
        // A line that matched no row means not enough stock: undo the whole reservation.
        if (batch.BatchCommands.Any<NpgsqlBatchCommand>(c => c.RecordsAffected != 1))
        {
            await tx.RollbackAsync(ct);
            return false;
        }
        await tx.CommitAsync(ct);
        return true;
    }
}

using Npgsql;
using NpgsqlTypes;

public sealed record PriceUpdate(string Sku, decimal UnitPrice, DateTime EffectiveAtUtc);

public sealed class PriceImporter(NpgsqlDataSource dataSource)
{
    public async Task<int> ImportAsync(IAsyncEnumerable<PriceUpdate> rows, CancellationToken ct)
    {
        await using var connection = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await connection.BeginTransactionAsync(ct);

        await using (var create = new NpgsqlCommand(
            "CREATE TEMP TABLE prices_in (sku text, unit_price numeric(12,2), effective_at timestamptz) ON COMMIT DROP",
            connection, tx))
        {
            await create.ExecuteNonQueryAsync(ct);
        }

        // The importer must be completed and disposed before the connection can run another command.
        await using (var importer = await connection.BeginBinaryImportAsync(
            "COPY prices_in (sku, unit_price, effective_at) FROM STDIN (FORMAT BINARY)", ct))
        {
            await foreach (var row in rows.WithCancellation(ct))
            {
                await importer.StartRowAsync(ct);
                await importer.WriteAsync(row.Sku, NpgsqlDbType.Text, ct);
                await importer.WriteAsync(row.UnitPrice, NpgsqlDbType.Numeric, ct);
                await importer.WriteAsync(row.EffectiveAtUtc, NpgsqlDbType.TimestampTz, ct);
            }
            await importer.CompleteAsync(ct);
        }

        // DISTINCT ON keeps the newest row per SKU: ON CONFLICT cannot touch the same row twice.
        await using var upsert = new NpgsqlCommand("""
            INSERT INTO product_prices (sku, unit_price, effective_at)
            SELECT DISTINCT ON (sku) sku, unit_price, effective_at
            FROM prices_in
            ORDER BY sku, effective_at DESC
            ON CONFLICT (sku) DO UPDATE
               SET unit_price = EXCLUDED.unit_price, effective_at = EXCLUDED.effective_at
             WHERE product_prices.effective_at < EXCLUDED.effective_at
            """, connection, tx);
        var changed = await upsert.ExecuteNonQueryAsync(ct);
        await tx.CommitAsync(ct);
        return changed;
    }
}

using Npgsql;
using NpgsqlTypes;

public sealed class PaymentLog(NpgsqlDataSource dataSource, TimeProvider clock)
{
    // Silently wrong: an untyped DateTime with Kind=Local is sent as timestamp (no zone),
    // and PostgreSQL casts it to timestamptz using the session TimeZone, not the app's.
    public async Task RecordBrokenAsync(long paymentId, CancellationToken ct)
    {
        await using var cmd = dataSource.CreateCommand("UPDATE payments SET captured_at = $1 WHERE id = $2");
        cmd.Parameters.Add(new NpgsqlParameter<DateTime> { TypedValue = DateTime.Now });
        cmd.Parameters.Add(new NpgsqlParameter<long> { TypedValue = paymentId });
        await cmd.ExecuteNonQueryAsync(ct);
    }

    // Works: a UTC instant from the injected clock, which tests can also control. The explicit
    // TimestampTz type makes a Local value throw ("Cannot write DateTime with Kind=Local ...").
    public async Task RecordAsync(long paymentId, CancellationToken ct)
    {
        await using var cmd = dataSource.CreateCommand("UPDATE payments SET captured_at = $1 WHERE id = $2");
        cmd.Parameters.Add(new NpgsqlParameter<DateTime>
        {
            TypedValue = clock.GetUtcNow().UtcDateTime,
            NpgsqlDbType = NpgsqlDbType.TimestampTz,
        });
        cmd.Parameters.Add(new NpgsqlParameter<long> { TypedValue = paymentId });
        await cmd.ExecuteNonQueryAsync(ct);
    }
}