master to dev

Api/Abstractions/IFirestoreService.cs

@@ -1,5 +1,6 @@
 using System.Text.Json.Nodes;
 using Api.Models;
+using Google.Cloud.Firestore;
 
 namespace Api.Abstractions;
 
@@ -8,4 +9,5 @@ public interface IFirestoreService
     Task CreateInitAsync(string jobId, string userId, string rawPath, JsonObject? targetSchema, string? prompt, JsonObject? options);
     Task<Dictionary<string, object>?> GetAsync(string jobId);
     Task UpdateStateAsync(string jobId, JobState state, Dictionary<string, object?>? extra = null);
+    Task<IReadOnlyList<DocumentSnapshot>> GetDocumentsByField(string fieldName, string fieldValue);
 }

Api/Abstractions/IRateLimiter.cs

@@ -0,0 +1,8 @@
+namespace Api.Abstractions;
+
+public interface IRateLimiter
+{
+    Task<bool> GetUserLimitsAsync(string userId);
+    Task<bool> CheckUserLimit(string userId);
+    Task UpdateOrCreateUserLimitAsync(string userId);
+}

Api/Api/JobLists.cs

@@ -0,0 +1,44 @@
+using Api.Abstractions;
+using Api.Services;
+using Google.Cloud.Firestore;
+using Microsoft.AspNetCore.Mvc;
+
+namespace Api.Api;
+
+[ApiController]
+[Route("api/jobs")]
+public class JobLists: ControllerBase
+{
+    private readonly ILogger<JobLists> _logger;
+    private readonly IFirestoreService _firestoreService;
+
+    public JobLists(
+        ILogger<JobLists> logger,
+        IFirestoreService firestoreService
+    )
+    {
+        _logger = logger;
+        _firestoreService = firestoreService;
+    }
+
+    [HttpGet]
+    public async Task <IActionResult> GetJobByUserId([FromQuery] string userId)
+    {
+        if (string.IsNullOrWhiteSpace(userId))
+            return Ok(Array.Empty<object>());
+
+        List<JobsDocument> jobList = new();
+        var documents = await _firestoreService.GetDocumentsByField("userId", userId);
+
+        if (documents is null || documents.Count == 0) 
+            return Ok(jobList);
+
+        foreach (var doc in documents)
+        {
+            var job = doc.ConvertTo<JobsDocument>();
+            jobList.Add(job);
+        }
+
+        return Ok(jobList);
+    }
+}

Api/Api/Pipelines/Initialize.cs

@@ -14,21 +14,24 @@ public class Initialize:
     private readonly IFirestoreService _firestoreService;
     private readonly IGcsSigner _gcsSigner;
     private readonly ApiConfig _apiConfig;
+    private readonly IRateLimiter _rateLimiter;
 
     public Initialize(
         IFirestoreService firestoreService,
         IGcsSigner gcsSigner,
-        IOptions<ApiConfig> apiConfig
+        IOptions<ApiConfig> apiConfig,
+        IRateLimiter rateLimiter
         )
     {
         _firestoreService = firestoreService;
         _gcsSigner = gcsSigner;
         _apiConfig = apiConfig.Value;
+        _rateLimiter = rateLimiter;
     }
 
     [HttpPost("init")]
     public virtual async Task<IActionResult> Init([FromBody] InitRequest request)
-        => await Initialize(request, _firestoreService, _gcsSigner, _apiConfig);
+        => await Initialize(request, _firestoreService, _gcsSigner, _rateLimiter, _apiConfig);
 
     [HttpGet("init")]
     public virtual IActionResult GetNotAllowed()

Api/Api/Pipelines/InitializeBase.cs

@@ -10,8 +10,13 @@ protected async Task<IActionResult> Initialize(
         [FromBody] InitRequest request, 
         IFirestoreService firestoreService,
         IGcsSigner gcsSigner,
+        IRateLimiter rateLimiter,
         ApiConfig apiConfig)
     {
+        var rateLimitResult = await rateLimiter.CheckUserLimit(request.UserId);
+        if (rateLimitResult)
+            return StatusCode(StatusCodes.Status429TooManyRequests, new {error = "To prevent abuse, we have rate limited your requests. Please try again later."});
+
         if (string.IsNullOrWhiteSpace(request.FileName))
             return BadRequest(new {error = "File Name is required"});
 

Api/DependencyInjection/AddApiService.cs

@@ -16,6 +16,7 @@ public static IServiceCollection AddApiServices( this IServiceCollection service
         services.AddSingleton<IPipelines<InitRequest>, Initialize>();
         services.AddSingleton<IPipelines<JobStatusRequest>, JobStatus>();
         services.AddSingleton<IHealth, HealthChecks>();
+        services.AddSingleton<IRateLimiter, RateLimiter>();
 
         return services;
     }

Api/Models/ApiConfig.cs

@@ -5,5 +5,6 @@ public class ApiConfig
     public string ProjectId { get; set; }
     public string BucketName { get; set; }
     public string FirestoreCollection { get; set; }
+    public string FirestoreRateCollection { get; set; }
     public string ServiceAccountKeyPath { get; set; }
 }

Api/Models/JobsDocument.cs

@@ -0,0 +1,26 @@
+using Google.Cloud.Firestore;
+
+[FirestoreData]
+public class JobsDocument
+{
+    [FirestoreProperty("jobId")]
+    public string JobId { get; set; }
+
+    [FirestoreProperty("userId")]
+    public string UserId { get; set; }
+
+    [FirestoreProperty("state")]
+    public string State { get; set; }
+
+    [FirestoreProperty("createdAt")]
+    public DateTime CreatedAt { get; set; }
+
+    [FirestoreProperty("updatedAt")]
+    public DateTime UpdatedAt { get; set; }
+
+    [FirestoreProperty("result")]
+    public object Result { get; set; }
+
+    [FirestoreProperty("error")]
+	public string? Error { get; set; }
+}

Api/Services/FirestoreService.cs

@@ -49,6 +49,15 @@ public async Task CreateInitAsync( string jobId,
         return snap.Exists ? snap.ToDictionary() : null;
     }
 
+    public async Task<IReadOnlyList<DocumentSnapshot>> GetDocumentsByField(string fieldName, string fieldValue)
+    {
+        var querySnap = await _collection.WhereEqualTo(fieldName, fieldValue).GetSnapshotAsync();
+        if(querySnap == null || querySnap.Documents.Count == 0)
+            return null;
+        return querySnap.Documents;
+
+    }
+
     public async Task UpdateStateAsync(string jobId, JobState state, Dictionary<string, object?>? extra = null)
     {
         var updates = new Dictionary<string, object?>

Api/Services/RateLimiter.cs

@@ -0,0 +1,72 @@
+using Api.Abstractions;
+using Api.Models;
+using Google.Cloud.Firestore;
+using Microsoft.Extensions.Options;
+
+namespace Api.Services;
+
+public class RateLimiter: IRateLimiter
+{
+    private readonly CollectionReference _collection;
+    private readonly ApiConfig _apiConfig;
+
+    public RateLimiter(IOptions<ApiConfig> apiConfig)
+    {
+        _apiConfig = apiConfig.Value;
+        var db = FirestoreDb.Create(_apiConfig.ProjectId);
+        _collection = db.Collection(_apiConfig.FirestoreRateCollection);
+    }
+
+    public async Task<bool> CheckUserLimit(string userId)
+    {
+        var userLimits = await GetUserLimitsAsync(userId);
+        if (userLimits)
+        {
+            return true;
+        }
+
+        return false;
+    }
+
+    public async Task<bool> GetUserLimitsAsync(string userId)
+    {
+        var snap = await _collection.Document(userId).GetSnapshotAsync();
+        if (!snap.Exists)
+        {
+            await UpdateOrCreateUserLimitAsync(userId);
+            return true;
+        }
+
+
+        if (snap.TryGetValue("lastUpdate", out Timestamp lastUpdateTimestamp))
+        {
+            var lastUpdated = lastUpdateTimestamp.ToDateTime().ToUniversalTime();
+            var now = DateTime.UtcNow;
+
+            if (now - lastUpdated > TimeSpan.FromSeconds(60))
+            {
+                await UpdateOrCreateUserLimitAsync(userId);
+                return false;
+            }
+
+            return true;
+        }
+
+        await UpdateOrCreateUserLimitAsync(userId);
+        return true;
+    }
+
+    public async Task UpdateOrCreateUserLimitAsync(string userId)
+    {
+        var doc = _collection.Document(userId);
+        var now = DateTime.UtcNow;
+        var data = new Dictionary<string, object>
+        {
+            ["userId"] = userId,
+            ["lastUpdate"] = now
+        };
+
+        await doc.SetAsync(data);
+    }
+
+}

Parser/Parser/Abstractions/IOutputSink.cs

@@ -0,0 +1,10 @@
+namespace Parser.Abstractions;
+
+public interface IOutputSink: IAsyncDisposable
+{
+    Task<int> WriteChunkAsync(StreamWriter writer, string transformedChunk);
+    int TotalRows { get; }
+    Stream GetFinalStream();
+    string FileExtension { get; }
+    string ContentType { get; } 
+}

Parser/Parser/Abstractions/IOutputSinkFactory.cs

@@ -0,0 +1,6 @@
+namespace Parser.Abstractions;
+
+public interface IOutputSinkFactory
+{
+    IOutputSink Create(string formatHint);
+}

Parser/Parser/Abstractions/ITransformationService.cs

@@ -4,4 +4,5 @@ public interface ITransformationService
 {
     Task<string> CallGeminiForChunkAsync(List<string> jsonlChunk, string userPrompt);
     Task<int> WriteJsonlAsync(StreamWriter writer, string modelOutput);
+    Task<string> GetValueFromGeminiAsync(string systemInstruction, string userPrompt);
 }