feat: using dapper operate db

GeoPulse Pipeline/GeoPulse Pipeline.csproj

@@ -12,6 +12,7 @@
         <PackageReference Include="Dapper" Version="2.1.72" />
         <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.5" />
         <PackageReference Include="Npgsql" Version="10.0.2" />
+        <PackageReference Include="Npgsql.DependencyInjection" Version="10.0.2" />
         <PackageReference Include="Swashbuckle.AspNetCore" Version="6.4.0" />
     </ItemGroup>
 

GeoPulse Pipeline/KafkaConsumer.cs

@@ -1,3 +1,4 @@
+using System.Data;
 using System.Text.Json;
 using Confluent.Kafka;
 using Npgsql;
@@ -7,18 +8,17 @@ namespace GeoPulse_Pipeline;
 public class KafkaConsumer : BackgroundService
 {
 	private readonly ILogger<KafkaConsumer> _logger;
-	private readonly string _connectionString;
+	private readonly NpgsqlDataSource _dataSource;
 	private readonly string _kafkaHost;
 
 	private const int BatchSize = 1000;
 	private readonly TimeSpan MaxWaitTime = TimeSpan.FromSeconds(2);
 
-	public KafkaConsumer(ILogger<KafkaConsumer> logger, IConfiguration configuration)
+	public KafkaConsumer(ILogger<KafkaConsumer> logger, IConfiguration configuration, NpgsqlDataSource dataSource)
 	{
 		_logger = logger;
-		_connectionString = configuration.GetConnectionString("DefaultConnection")
-		                    ?? throw new ArgumentNullException("資料庫連線字串未設定");
 		_kafkaHost = configuration["KafkaHost"];
+		_dataSource = dataSource;
 	}
 
 	protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -28,20 +28,25 @@ public class KafkaConsumer : BackgroundService
 		var consumerConfig = new ConsumerConfig
 		{
 			BootstrapServers = _kafkaHost,
-			GroupId = "telemetry-db-writer-group",
+			GroupId = "telemetry-db-writer-group", // 💡 建議換個名字，強迫 Kafka 重置 Offset
 			AutoOffsetReset = AutoOffsetReset.Earliest,
 			EnableAutoCommit = false
 		};
 
-		using var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build();
+		using var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig)
+			.SetErrorHandler((_, e) => _logger.LogError($"Kafka 底層錯誤: {e.Reason}"))
+			.Build();
+
 		consumer.Subscribe("telemetry-events");
 
 		var buffer = new List<ConsumeResult<Ignore, string>>();
 		var lastBatchTime = DateTime.UtcNow;
 
-		try
-		{
+		_logger.LogInformation("🚀 Consumer 已啟動，開始監控 Kafka 訊息...");
+
 		while (!stoppingToken.IsCancellationRequested)
+		{
+			try // 💡 關鍵修正：try 放在 while 裡面
 			{
 				var consumeResult = consumer.Consume(TimeSpan.FromMilliseconds(500));
 
@@ -55,27 +60,31 @@ public class KafkaConsumer : BackgroundService
 
 				if (isBatchFull || isTimeUp)
 				{
+					_logger.LogInformation($"[處理中] 準備批次寫入 {buffer.Count} 筆資料...");
+
 					await BulkInsertToDatabaseAsync(buffer, stoppingToken);
 
 					var latestOffset = buffer.Last();
 					consumer.Commit(latestOffset);
 
-					_logger.LogInformation($"成功批次寫入 {buffer.Count} 筆資料至 PostgreSQL。");
+					_logger.LogInformation($"✅ [成功] 已寫入 {buffer.Count} 筆資料至 PostgreSQL。");
 
 					buffer.Clear();
 					lastBatchTime = DateTime.UtcNow;
 				}
 			}
-		}
 			catch (Exception ex)
 			{
-			_logger.LogError($"發生錯誤: {ex.Message} \n {ex.StackTrace}");
+				// 💡 就算這裡噴錯，程式也只是印出 Log，然後回到 while 頂部繼續跑下一輪
+				_logger.LogError(ex, "❌ [失敗] 批次處理發生錯誤！原因: {Message}", ex.Message);
 
+				// 發生錯誤後，為了避免壞資料卡死 buffer 導致無窮重試，這裡建議清空
 				buffer.Clear();
+				lastBatchTime = DateTime.UtcNow;
+
+				// 稍微停 2 秒，避免在連續錯誤時瘋狂刷屏
+				await Task.Delay(2000, stoppingToken);
 			}
-		finally
-		{
-			consumer.Close();
 		}
 	}
 
@@ -83,8 +92,7 @@ public class KafkaConsumer : BackgroundService
 		CancellationToken stoppingToken)
 	{
 		var jsonOptions = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
-		await using var dataSource = NpgsqlDataSource.Create(_connectionString);
-		await using var batch = dataSource.CreateBatch();
+		await using var batch = _dataSource.CreateBatch();
 
 		foreach (var msg in buffer)
 		{

GeoPulse Pipeline/Program.cs

@@ -1,5 +1,7 @@
+using System.Data;
 using System.Text.Json;
 using Confluent.Kafka;
+using Dapper;
 using GeoPulse_Pipeline;
 using Microsoft.AspNetCore.Mvc;
 using Npgsql;
@@ -12,12 +14,14 @@ var producerConfig = new ProducerConfig()
 	Acks = Acks.Leader
 };
 
-
+var connString = builder.Configuration.GetConnectionString("DefaultConnection");
 var kafkaProducer = new ProducerBuilder<Null, string>(producerConfig).Build();
 builder.Services.AddSingleton(kafkaProducer);
 builder.Services.AddEndpointsApiExplorer();
 builder.Services.AddSwaggerGen();
 builder.Services.AddHostedService<KafkaConsumer>();
+builder.Services.AddNpgsqlDataSource(connString);
+builder.Services.AddScoped<IDbConnection>(sp => sp.GetRequiredService<NpgsqlDataSource>().CreateConnection());
 
 var app = builder.Build();
 
@@ -33,7 +37,7 @@ app.MapPost("/api/telemetry", async (
 	[FromBody] TelemetryRequest request,
 	[FromQuery] bool useKafka,
 	IProducer<Null, string> producer,
-    IConfiguration config) =>
+	IDbConnection dbConnection) =>
 {
 	if (string.IsNullOrWhiteSpace(request.DeviceId))
 		return Results.BadRequest("Device ID is required.");
@@ -56,19 +60,9 @@ app.MapPost("/api/telemetry", async (
 	{
 		try
 		{
-            var connString = config.GetConnectionString("DefaultConnection");
-            
-            await using var dataSource = NpgsqlDataSource.Create(connString!);
-            await using var cmd = dataSource.CreateCommand(@"
-                INSERT INTO telemetry_table (id, device_id, geom, timestamp) 
-                VALUES (gen_random_uuid(), $1, ST_SetSRID(ST_MakePoint($2, $3), 4326), $4)");
-
-            cmd.Parameters.AddWithValue(request.DeviceId);
-            cmd.Parameters.AddWithValue(request.Lng);
-            cmd.Parameters.AddWithValue(request.Lat);
-            cmd.Parameters.AddWithValue(request.Timestamp);
-
-            await cmd.ExecuteNonQueryAsync();
+			string sql =
+				@"INSERT INTO telemetry_history (id, device_id, geom, timestamp) VALUES (gen_random_uuid(), @DeviceId, @Lng, @Lat, @Timestamp";
+			await dbConnection.ExecuteAsync(sql, request);
 
 			return Results.Ok(new { Status = "Success", Route = "Direct-DB", Message = "已直接寫入資料庫" });
 		}