feat(import): Data Import and Export

JobFlow.API/Controllers/DataExportController.cs

@@ -0,0 +1,198 @@
+using Hangfire;
+using JobFlow.API.Extensions;
+using JobFlow.API.Models;
+using JobFlow.API.Services;
+using JobFlow.Business.Services.ServiceInterfaces;
+using JobFlow.Domain.Models;
+using JobFlow.Infrastructure.Persistence;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.EntityFrameworkCore;
+
+namespace JobFlow.API.Controllers;
+
+[ApiController]
+[Route("api/data-export")]
+public class DataExportController : ControllerBase
+{
+    private readonly IDbContextFactory<JobFlowDbContext> _dbContextFactory;
+    private readonly DataExportBuilderService _builder;
+    private readonly IOrganizationService _organizations;
+
+    public DataExportController(
+        IDbContextFactory<JobFlowDbContext> dbContextFactory,
+        DataExportBuilderService builder,
+        IOrganizationService organizations)
+    {
+        _dbContextFactory = dbContextFactory;
+        _builder = builder;
+        _organizations = organizations;
+    }
+
+    [HttpGet("json")]
+    public async Task<IResult> ExportOrganizationDataJson(CancellationToken cancellationToken)
+    {
+        var organizationId = HttpContext.GetOrganizationId();
+        var (bytes, fileName) = await _builder.BuildJsonBundleAsync(organizationId, cancellationToken);
+
+        return Results.File(bytes, "application/json", fileName);
+    }
+
+    [HttpGet("clients.csv")]
+    public async Task<IResult> ExportClientsCsv(CancellationToken cancellationToken)
+    {
+        var organizationId = HttpContext.GetOrganizationId();
+        var (bytes, fileName) = await _builder.BuildClientsCsvAsync(organizationId, cancellationToken);
+
+        return Results.File(bytes, "text/csv", fileName);
+    }
+
+    [HttpPost("jobs")]
+    public async Task<IResult> StartDataExportJob(CancellationToken cancellationToken)
+    {
+        var organizationId = HttpContext.GetOrganizationId();
+        var userId = HttpContext.GetUserId();
+
+        var orgResult = await _organizations.GetOrganizationDtoById(organizationId);
+        if (orgResult.IsFailure)
+        {
+            return Results.Problem(statusCode: 404, title: "Organization not found", detail: "Organization context is invalid.");
+        }
+
+        if (!HasMinPlan(orgResult.Value.SubscriptionPlanName, "Flow"))
+        {
+            return Results.Problem(
+                statusCode: StatusCodes.Status403Forbidden,
+                title: "Subscription Required",
+                detail: "A Flow plan is required for async ZIP data exports.");
+        }
+
+        var jobId = Guid.NewGuid();
+        await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken);
+
+        var activeJob = await dbContext.Set<DataExportJob>()
+            .AsNoTracking()
+            .FirstOrDefaultAsync(x => x.OrganizationId == organizationId && (x.Status == "queued" || x.Status == "running"), cancellationToken);
+
+        if (activeJob is not null)
+        {
+            return Results.Ok(new StartDataExportJobResponse { JobId = activeJob.Id.ToString("N") });
+        }
+
+        dbContext.Set<DataExportJob>().Add(new DataExportJob
+        {
+            Id = jobId,
+            OrganizationId = organizationId,
+            RequestedByUserId = userId,
+            Status = "queued",
+            CreatedAt = DateTime.UtcNow,
+            IsActive = true
+        });
+
+        await dbContext.SaveChangesAsync(cancellationToken);
+
+        BackgroundJob.Enqueue<DataExportJobProcessor>(x => x.ProcessAsync(jobId, organizationId));
+
+        return Results.Ok(new StartDataExportJobResponse { JobId = jobId.ToString("N") });
+    }
+
+    [HttpGet("jobs")]
+    public async Task<IResult> GetDataExportJobs(CancellationToken cancellationToken)
+    {
+        var organizationId = HttpContext.GetOrganizationId();
+        await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken);
+
+        var jobs = await dbContext.Set<DataExportJob>()
+            .AsNoTracking()
+            .Where(x => x.OrganizationId == organizationId)
+            .OrderByDescending(x => x.CreatedAt)
+            .Take(20)
+            .Select(x => new DataExportJobStatusResponse
+            {
+                JobId = x.Id.ToString("N"),
+                Status = x.Status,
+                ErrorMessage = x.ErrorMessage,
+                FileName = x.FileName,
+                ContentType = x.ContentType,
+                StartedAtUtc = x.StartedAtUtc,
+                CompletedAtUtc = x.CompletedAtUtc,
+                ExpiresAtUtc = x.ExpiresAtUtc,
+                DownloadCount = x.DownloadCount
+            })
+            .ToListAsync(cancellationToken);
+
+        return Results.Ok(jobs);
+    }
+
+    [HttpGet("jobs/{jobId}")]
+    public async Task<IResult> GetDataExportJobStatus(string jobId, CancellationToken cancellationToken)
+    {
+        if (!Guid.TryParse(jobId, out var parsedJobId))
+            return Results.BadRequest("Invalid export job id.");
+
+        var organizationId = HttpContext.GetOrganizationId();
+        await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken);
+
+        var job = await dbContext.Set<DataExportJob>()
+            .AsNoTracking()
+            .Where(x => x.Id == parsedJobId && x.OrganizationId == organizationId)
+            .Select(x => new DataExportJobStatusResponse
+            {
+                JobId = x.Id.ToString("N"),
+                Status = x.Status,
+                ErrorMessage = x.ErrorMessage,
+                FileName = x.FileName,
+                ContentType = x.ContentType,
+                StartedAtUtc = x.StartedAtUtc,
+                CompletedAtUtc = x.CompletedAtUtc,
+                ExpiresAtUtc = x.ExpiresAtUtc,
+                DownloadCount = x.DownloadCount
+            })
+            .FirstOrDefaultAsync(cancellationToken);
+
+        return job is null ? Results.NotFound() : Results.Ok(job);
+    }
+
+    [HttpGet("jobs/{jobId}/download")]
+    public async Task<IResult> DownloadDataExportJobFile(string jobId, CancellationToken cancellationToken)
+    {
+        if (!Guid.TryParse(jobId, out var parsedJobId))
+            return Results.BadRequest("Invalid export job id.");
+
+        var organizationId = HttpContext.GetOrganizationId();
+        await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken);
+
+        var job = await dbContext.Set<DataExportJob>()
+            .FirstOrDefaultAsync(x => x.Id == parsedJobId && x.OrganizationId == organizationId, cancellationToken);
+
+        if (job is null)
+            return Results.NotFound();
+
+        if (job.Status != "completed" || job.FileContent is null || string.IsNullOrWhiteSpace(job.FileName))
+            return Results.Conflict(new { message = "Export file is not ready yet." });
+
+        if (job.ExpiresAtUtc.HasValue && job.ExpiresAtUtc.Value < DateTime.UtcNow)
+            return Results.StatusCode(StatusCodes.Status410Gone);
+
+        job.DownloadCount += 1;
+        await dbContext.SaveChangesAsync(cancellationToken);
+
+        return Results.File(job.FileContent, job.ContentType ?? "application/octet-stream", job.FileName);
+    }
+
+    private static bool HasMinPlan(string? planName, string required)
+    {
+        static int Rank(string? plan)
+        {
+            var value = (plan ?? string.Empty).Trim().ToLowerInvariant();
+            return value switch
+            {
+                "go" => 0,
+                "flow" => 1,
+                "max" => 2,
+                _ => -1
+            };
+        }
+
+        return Rank(planName) >= Rank(required);
+    }
+}

JobFlow.API/Controllers/OrganizationClientController.cs

@@ -1,13 +1,17 @@
 using JobFlow.API.Extensions;
 using JobFlow.API.Mappings;
+using JobFlow.API.Services;
 using JobFlow.Business;
 using JobFlow.API.Models;
 using JobFlow.Business.Extensions;
 using JobFlow.Business.Models.DTOs;
 using JobFlow.Business.Services.ServiceInterfaces;
 using JobFlow.Domain.Models;
+using Hangfire;
 using MapsterMapper;
 using Microsoft.AspNetCore.Mvc;
+using Microsoft.EntityFrameworkCore;
+using JobFlow.Infrastructure.Persistence;
 
 namespace JobFlow.API.Controllers;
 
@@ -18,15 +22,24 @@ public class OrganizationClientController : ControllerBase
     private readonly IOrganizationClientService organizationClientService;
     private readonly IOrganizationClientPortalService _clientPortal;
     private readonly IMapper _mapper;
+    private readonly ClientImportCsvService _csvImportService;
+    private readonly ClientImportUploadSessionService _uploadSessionService;
+    private readonly IDbContextFactory<JobFlowDbContext> _dbContextFactory;
 
     public OrganizationClientController(
         IOrganizationClientService organizationClientService,
         IOrganizationClientPortalService clientPortal,
-        IMapper mapper)
+        IMapper mapper,
+        ClientImportCsvService csvImportService,
+        ClientImportUploadSessionService uploadSessionService,
+        IDbContextFactory<JobFlowDbContext> dbContextFactory)
     {
         this.organizationClientService = organizationClientService;
         _clientPortal = clientPortal;
         _mapper = mapper;
+        _csvImportService = csvImportService;
+        _uploadSessionService = uploadSessionService;
+        _dbContextFactory = dbContextFactory;
     }
 
     [HttpGet]
@@ -121,4 +134,134 @@ public async Task<IResult> RestoreClient(Guid clientId)
         var result = await organizationClientService.RestoreClient(clientId, organizationId);
         return result.IsSuccess ? Results.Ok(result) : result.ToProblemDetails();
     }
+
+    [HttpPost("import/preview")]
+    [RequestSizeLimit(10 * 1024 * 1024)]
+    [Consumes("multipart/form-data")]
+    public async Task<IResult> PreviewClientImport([FromForm] PreviewClientImportRequest request, CancellationToken cancellationToken)
+    {
+        var file = request.File;
+        if (file is null)
+            return Results.BadRequest("A CSV file is required.");
+
+        if (!file.FileName.EndsWith(".csv", StringComparison.OrdinalIgnoreCase))
+            return Results.BadRequest("Only CSV files are supported in this version.");
+
+        try
+        {
+            var organizationId = HttpContext.GetOrganizationId();
+            var parsed = await _csvImportService.ParseAsync(file, cancellationToken);
+            var source = string.IsNullOrWhiteSpace(request.SourceSystem) ? "csv" : request.SourceSystem.Trim();
+            var uploadSessionId = await _uploadSessionService.SaveAsync(organizationId, source, parsed.Rows, cancellationToken);
+
+            var previewRows = parsed.Rows
+                .Take(25)
+                .Select(r => r.ToDictionary(x => x.Key, x => x.Value, StringComparer.OrdinalIgnoreCase))
+                .ToList();
+
+            var response = new ClientImportPreviewResponse
+            {
+                UploadToken = uploadSessionId.ToString("N"),
+                SourceSystem = source,
+                SourceColumns = parsed.Headers,
+                SuggestedMappings = parsed.SuggestedMappings,
+                PreviewRows = previewRows,
+                TotalRows = parsed.Rows.Count
+            };
+
+            return Results.Ok(response);
+        }
+        catch (Exception ex)
+        {
+            return Results.BadRequest(ex.Message);
+        }
+    }
+
+    [HttpPost("import/start")]
+    public async Task<IResult> StartClientImport([FromBody] StartClientImportRequest request)
+    {
+        if (request is null || string.IsNullOrWhiteSpace(request.UploadToken))
+            return Results.BadRequest("Upload token is required.");
+
+        if (!Guid.TryParse(request.UploadToken, out var uploadSessionId))
+            return Results.BadRequest("Invalid upload token format.");
+
+        var organizationId = HttpContext.GetOrganizationId();
+        var uploadSession = await _uploadSessionService.GetActiveSessionAsync(uploadSessionId, organizationId, CancellationToken.None);
+        if (uploadSession is null)
+            return Results.BadRequest("Import session expired or invalid. Please upload your CSV again.");
+
+        if (request.ColumnMappings.Count == 0)
+            return Results.BadRequest("At least one column mapping is required.");
+
+        var jobId = Guid.NewGuid();
+        var sourceSystem = string.IsNullOrWhiteSpace(request.SourceSystem) ? "csv" : request.SourceSystem.Trim();
+
+        await using var dbContext = await _dbContextFactory.CreateDbContextAsync();
+        var importJob = new ClientImportJob
+        {
+            Id = jobId,
+            OrganizationId = organizationId,
+            SourceSystem = sourceSystem,
+            Status = "queued",
+            TotalRows = uploadSession.TotalRows,
+            ProcessedRows = 0,
+            SucceededRows = 0,
+            FailedRows = 0,
+            CreatedAt = DateTime.UtcNow,
+            IsActive = true
+        };
+
+        dbContext.Set<ClientImportJob>().Add(importJob);
+        await dbContext.SaveChangesAsync();
+
+        BackgroundJob.Enqueue<ClientImportProcessor>(
+            processor => processor.ProcessAsync(jobId, organizationId, uploadSessionId, request.ColumnMappings));
+
+        return Results.Ok(new StartClientImportResponse { JobId = jobId.ToString("N") });
+    }
+
+    [HttpGet("import/jobs/{jobId}")]
+    public async Task<IResult> GetClientImportStatus(string jobId)
+    {
+        if (!Guid.TryParse(jobId, out var parsedJobId))
+            return Results.BadRequest("Invalid import job id.");
+
+        var organizationId = HttpContext.GetOrganizationId();
+
+        await using var dbContext = await _dbContextFactory.CreateDbContextAsync();
+        var job = await dbContext.Set<ClientImportJob>()
+            .AsNoTracking()
+            .FirstOrDefaultAsync(x => x.Id == parsedJobId && x.OrganizationId == organizationId);
+
+        if (job is null)
+            return Results.NotFound();
+
+        var errors = await dbContext.Set<ClientImportJobError>()
+            .AsNoTracking()
+            .Where(x => x.ClientImportJobId == parsedJobId)
+            .OrderBy(x => x.RowNumber)
+            .Take(100)
+            .Select(x => new ClientImportErrorItem
+            {
+                RowNumber = x.RowNumber,
+                Message = x.Message
+            })
+            .ToListAsync();
+
+        var status = new ClientImportJobStatusResponse
+        {
+            JobId = job.Id.ToString("N"),
+            SourceSystem = job.SourceSystem,
+            Status = job.Status,
+            TotalRows = job.TotalRows,
+            ProcessedRows = job.ProcessedRows,
+            SucceededRows = job.SucceededRows,
+            FailedRows = job.FailedRows,
+            ErrorMessage = job.ErrorMessage,
+            Errors = errors
+        };
+
+        return Results.Ok(status);
+    }
 }

JobFlow.API/JobFlow.API.csproj

@@ -14,6 +14,7 @@
     <ItemGroup>
         <PackageReference Include="Azure.Extensions.AspNetCore.Configuration.Secrets" Version="1.4.0" />
         <PackageReference Include="Azure.Identity" Version="1.17.1" />
+        <PackageReference Include="CsvHelper" Version="33.1.0" />
         <PackageReference Include="FirebaseAdmin" Version="3.4.0" />
         <PackageReference Include="FluentValidation.AspNetCore" Version="11.3.0" />
         <PackageReference Include="Google.Apis.Auth" Version="1.73.0" />