Add Client CKYC backend module with all source, migrations, and static resources

Includes CKYC batch processing, KYC proof types, non-CKYC client APIs,
ReKYC handler, SFTP upload service, and DB migrations (0250-0255).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

fineract-core/src/main/java/org/apache/fineract/commands/service/CommandWrapperBuilder.java

@@ -665,6 +665,24 @@ public class CommandWrapperBuilder {
         return this;
     }
 
+    public CommandWrapperBuilder completeRekycClientKyc(final Long clientId, final Long kycId) {
+        this.actionName = "COMPLETEREKYC";
+        this.entityName = "CLIENTKYC";
+        this.entityId = kycId;
+        this.clientId = clientId;
+        this.href = "/clients/" + clientId + "/kyc/" + kycId + "?command=complete-rekyc";
+        return this;
+    }
+
+    public CommandWrapperBuilder uploadCkycClientKyc(final Long clientId, final Long kycId) {
+        this.actionName = "UPLOADCKYC";
+        this.entityName = "CLIENTKYC";
+        this.entityId = kycId;
+        this.clientId = clientId;
+        this.href = "/clients/" + clientId + "/kyc/" + kycId + "?command=upload-ckyc";
+        return this;
+    }
+
     // ── Form 60 ──────────────────────────────────────────────────────────────
 
     public CommandWrapperBuilder createClientForm60(final Long clientId) {

fineract-core/src/main/java/org/apache/fineract/portfolio/client/domain/ClientKyc.java

@@ -139,6 +139,12 @@ public class ClientKyc extends AbstractAuditableWithUTCDateTimeCustom<Long> {
     @Column(name = "din_verified_on")
     private LocalDate dinVerifiedOn;
 
+    @Column(name = "is_director", nullable = false)
+    private boolean isDirector;
+
+    @Column(name = "din_mca_response", columnDefinition = "longtext")
+    private String dinMcaResponse;
+
     @Column(name = "incorporation_date")
     private LocalDate incorporationDate;
 
@@ -202,6 +208,14 @@ public class ClientKyc extends AbstractAuditableWithUTCDateTimeCustom<Long> {
     @Column(name = "ckyc_number", length = 14)
     private String ckycNumber;
 
+    /** Date on which the CKYC record was pushed to CERSAI (SFTP batch). */
+    @Column(name = "ckyc_uploaded_on")
+    private LocalDate ckycUploadedOn;
+
+    /** Our internal batch id ("BATCH-yyyy-MM-dd-NNN") this KYC was shipped in. */
+    @Column(name = "ckyc_batch_id", length = 50)
+    private String ckycBatchId;
+
     // ── Lifecycle dates ───────────────────────────────────────────────────────
     @Column(name = "initiated_on")
     private LocalDate initiatedOn;
@@ -427,6 +441,17 @@ public class ClientKyc extends AbstractAuditableWithUTCDateTimeCustom<Long> {
         this.expiryDate = null;
     }
 
+    /**
+     * Records a CKYC upload to CERSAI. Sets the 14-digit CKYC number, the upload date and the
+     * optional batch id. Does <b>not</b> change {@code kycStatus} — upload is orthogonal to
+     * verification and is normally performed only after the KYC is already VERIFIED.
+     */
+    public void markCkycUploaded(final String ckycNumber, final String batchId) {
+        this.ckycNumber = StringUtils.defaultIfEmpty(ckycNumber, null);
+        this.ckycBatchId = StringUtils.defaultIfEmpty(batchId, null);
+        this.ckycUploadedOn = LocalDate.now();
+    }
+
     /** Called after successful 3rd-party verification of all required documents. */
     public void markVerified(final String ckycNumber) {
         this.kycStatus = ClientKycStatus.VERIFIED.getValue();
@@ -493,6 +518,14 @@ public class ClientKyc extends AbstractAuditableWithUTCDateTimeCustom<Long> {
         this.dinVerifiedOn = LocalDate.now();
     }
 
+    public void markDinVerified(final String dinNumber, final String mcaResponseJson) {
+        this.dinNumber = dinNumber;
+        this.dinVerified = true;
+        this.dinVerifiedOn = LocalDate.now();
+        this.isDirector = true;
+        this.dinMcaResponse = mcaResponseJson;
+    }
+
     /**
      * Checks if all required verifications are complete for this KYC type and auto-promotes
      * to VERIFIED status if so.
@@ -609,6 +642,14 @@ public class ClientKyc extends AbstractAuditableWithUTCDateTimeCustom<Long> {
         return ckycNumber;
     }
 
+    public LocalDate getCkycUploadedOn() {
+        return ckycUploadedOn;
+    }
+
+    public String getCkycBatchId() {
+        return ckycBatchId;
+    }
+
     public LocalDate getInitiatedOn() {
         return initiatedOn;
     }
@@ -657,6 +698,14 @@ public class ClientKyc extends AbstractAuditableWithUTCDateTimeCustom<Long> {
         return dinVerifiedOn;
     }
 
+    public boolean isDirector() {
+        return isDirector;
+    }
+
+    public String getDinMcaResponse() {
+        return dinMcaResponse;
+    }
+
     public LocalDate getIncorporationDate() {
         return incorporationDate;
     }

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/api/ClientKycApiResource.java

@@ -27,13 +27,18 @@ import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.ws.rs.Consumes;
 import jakarta.ws.rs.GET;
+import jakarta.ws.rs.HeaderParam;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.PUT;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.PathParam;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
+import java.io.InputStream;
+import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
+import org.glassfish.jersey.media.multipart.FormDataParam;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -87,6 +92,7 @@ public class ClientKycApiResource {
     private final DefaultToApiJsonSerializer<ClientKycData> toApiJsonSerializer;
     private final PortfolioCommandSourceWritePlatformService commandsSourceWritePlatformService;
     private final ClientRepositoryWrapper clientRepository;
+    private final org.apache.fineract.portfolio.client.service.ClientKycWritePlatformService clientKycWritePlatformService;
 
     // ─────────────────────────────────────────────────────────────────────────
     // INITIATE KYC – POST /v1/clients/{clientId}/kyc
@@ -346,7 +352,7 @@ public class ClientKycApiResource {
             @PathParam("clientId") @Parameter(description = "Client identifier") final Long clientId,
             @PathParam("kycId") @Parameter(description = "KYC record identifier") final Long kycId,
             @QueryParam("command") @Parameter(
-            description = "Command: aadhaar-generate-otp | verify-pan | verify-aadhaar | verify-gstin | verify-cin | verify-din | verify | reject")
+            description = "Command: aadhaar-generate-otp | verify-pan | verify-aadhaar | verify-gstin | verify-cin | verify-din | verify | reject | complete-rekyc | upload-ckyc")
             final String command,
             @Parameter(hidden = true) final String apiRequestBodyAsJson) {
 
@@ -377,10 +383,14 @@ public class ClientKycApiResource {
             commandRequest = new CommandWrapperBuilder().verifyClientKyc(clientId, kycId).withJson(json).build();
         } else if ("reject".equalsIgnoreCase(command)) {
             commandRequest = new CommandWrapperBuilder().rejectClientKyc(clientId, kycId).withJson(json).build();
+        } else if ("complete-rekyc".equalsIgnoreCase(command)) {
+            commandRequest = new CommandWrapperBuilder().completeRekycClientKyc(clientId, kycId).withJson(json).build();
+        } else if ("upload-ckyc".equalsIgnoreCase(command)) {
+            commandRequest = new CommandWrapperBuilder().uploadCkycClientKyc(clientId, kycId).withJson(json).build();
         } else {
             throw new org.apache.fineract.infrastructure.core.exception.UnrecognizedQueryParamException(
                     "command", command,
-                    "aadhaar-generate-otp", "verify-pan", "verify-aadhaar", "verify-gstin", "verify-cin", "verify-din", "verify", "reject");
+                    "aadhaar-generate-otp", "verify-pan", "verify-aadhaar", "verify-gstin", "verify-cin", "verify-din", "verify", "reject", "complete-rekyc", "upload-ckyc");
         }
 
         try {
@@ -392,4 +402,44 @@ public class ClientKycApiResource {
             throw e;
         }
     }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // CKYC ZIP UPLOAD – POST /v1/clients/{clientId}/kyc/{kycId}/ckyc-upload
+    //
+    // Pipeline 2: the UI assembles a CERSAI-format ZIP and posts it as multipart/form-data.
+    // The backend persists the ZIP in m_document (parent_entity_type='client_kyc'), hands the
+    // bytes to CkycSftpUploadService (dummy stub today, real SFTP later) and writes the
+    // returned 14-digit CKYC number + upload date onto the m_client_kyc row.
+    // ─────────────────────────────────────────────────────────────────────────
+
+    @POST
+    @Path("{kycId}/ckyc-upload")
+    @Consumes(MediaType.MULTIPART_FORM_DATA)
+    @Produces(MediaType.APPLICATION_JSON)
+    @Operation(
+            summary = "Upload a pre-built CERSAI CKYC ZIP for a client's KYC record",
+            description = "Accepts a multipart/form-data request with a single `file` part containing a "
+                    + "CERSAI-format ZIP archive. The backend persists the ZIP, submits it to CERSAI "
+                    + "(dummy stub today), saves the returned CKYC number, batch id and upload date "
+                    + "on m_client_kyc, and returns the changes. Idempotent retries are NOT supported "
+                    + "on this endpoint — each call persists a new document row; use `command=upload-ckyc` "
+                    + "for the server-builds-batch flow.")
+    @ApiResponse(responseCode = "200", description = "CKYC ZIP accepted and CKYC number persisted",
+            content = @Content(schema = @Schema(implementation = CommandProcessingResult.class)))
+    @ApiResponse(responseCode = "400", description = "Missing file, wrong extension, or invalid ZIP")
+    @ApiResponse(responseCode = "503", description = "CERSAI upload failed")
+    public CommandProcessingResult uploadCkycZip(
+            @PathParam("clientId") @Parameter(description = "Client identifier") final Long clientId,
+            @PathParam("kycId") @Parameter(description = "KYC record identifier") final Long kycId,
+            @HeaderParam(HttpHeaders.CONTENT_LENGTH) final Long fileSize,
+            @FormDataParam("file") final InputStream fileStream,
+            @FormDataParam("file") final FormDataContentDisposition fileMeta,
+            @FormDataParam("remarks") final String remarks) {
+
+        final String fileName = fileMeta != null ? fileMeta.getFileName() : null;
+        LOG.info("CKYC ZIP upload received: clientId={}, kycId={}, fileName={}, declaredSize={}",
+                clientId, kycId, fileName, fileSize);
+
+        return clientKycWritePlatformService.uploadCkycZip(clientId, kycId, fileStream, fileName, fileSize, remarks);
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/api/KycProofTypesApiResource.java

@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.api;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.MediaType;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.infrastructure.codes.data.CodeValueData;
+import org.apache.fineract.infrastructure.codes.service.CodeValueReadPlatformService;
+import org.apache.fineract.infrastructure.core.serialization.DefaultToApiJsonSerializer;
+import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
+import org.springframework.stereotype.Component;
+
+/**
+ * REST endpoint that exposes the master list of ID proofs and address proofs accepted under the
+ * RBI Master Direction on KYC. Intended to populate UI dropdowns on client onboarding / KYC screens.
+ *
+ * <p>The lists are stored as Fineract code values under two system-defined codes —
+ * {@code KYCIdProofType} and {@code KYCAddressProofType} — so administrators can add, deactivate
+ * or reorder entries from the System → Codes screen without a code change.
+ *
+ * <pre>
+ *   GET /v1/kyc/proof-types                 – Returns both ID proofs and address proofs
+ *   GET /v1/kyc/proof-types?type=id         – Returns only ID proofs
+ *   GET /v1/kyc/proof-types?type=address    – Returns only address proofs
+ * </pre>
+ */
+@Path("/v1/kyc/proof-types")
+@Component
+@Tag(name = "KYC Proof Types",
+        description = "Master list of officially valid documents (OVDs) for ID and Address proof per RBI KYC norms. "
+                + "Backed by code values under KYCIdProofType and KYCAddressProofType.")
+@RequiredArgsConstructor
+public class KycProofTypesApiResource {
+
+    public static final String ID_PROOF_CODE_NAME = "KYCIdProofType";
+    public static final String ADDRESS_PROOF_CODE_NAME = "KYCAddressProofType";
+
+    private final PlatformSecurityContext context;
+    private final CodeValueReadPlatformService codeValueReadPlatformService;
+    private final DefaultToApiJsonSerializer<Object> toApiJsonSerializer;
+
+    @GET
+    @Consumes({ MediaType.APPLICATION_JSON })
+    @Produces({ MediaType.APPLICATION_JSON })
+    @Operation(summary = "Retrieve KYC proof types",
+            description = "Returns the list of RBI-accepted ID proofs and address proofs for use in UI dropdowns. "
+                    + "Filter using the `type` query param: `id`, `address`, or omit for both. "
+                    + "Values come from code values KYCIdProofType / KYCAddressProofType in `m_code_value`.")
+    public String retrieveProofTypes(@QueryParam("type") final String type) {
+
+        context.authenticatedUser();
+
+        final Map<String, Object> response = new LinkedHashMap<>();
+
+        final boolean includeId = type == null || type.isBlank() || "id".equalsIgnoreCase(type) || "all".equalsIgnoreCase(type);
+        final boolean includeAddress = type == null || type.isBlank() || "address".equalsIgnoreCase(type)
+                || "all".equalsIgnoreCase(type);
+
+        if (includeId) {
+            final List<CodeValueData> idProofs = codeValueReadPlatformService.retrieveCodeValuesByCode(ID_PROOF_CODE_NAME);
+            response.put("idProofs", idProofs);
+        }
+        if (includeAddress) {
+            final List<CodeValueData> addressProofs = codeValueReadPlatformService.retrieveCodeValuesByCode(ADDRESS_PROOF_CODE_NAME);
+            response.put("addressProofs", addressProofs);
+        }
+
+        return toApiJsonSerializer.serialize(response);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/api/NonCkycClientApiResource.java

@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.api;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.MediaType;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.infrastructure.core.serialization.DefaultToApiJsonSerializer;
+import org.apache.fineract.portfolio.client.data.NonCkycClientData;
+import org.apache.fineract.portfolio.client.service.NonCkycClientReadPlatformService;
+import org.springframework.stereotype.Component;
+
+/**
+ * REST endpoint backing the "Mark Non-CKYC" screen at {@code /ckyc/record-master}.
+ *
+ * <p>Returns active clients whose {@code m_client_kyc.ckyc_number} is missing (row absent or column
+ * null/blank). Each row carries the identity fields the UI needs plus flags for the four KYC
+ * artefacts (photo / signature / id proof / address proof) so the operator can see what's missing
+ * at a glance and bulk-flag records for the CERSAI batch.
+ *
+ * <pre>
+ *   GET /v1/ckyc/record-master
+ *       ?officeId=1
+ *       &productId=7
+ *       &fromDate=2025-01-01
+ *       &toDate=2026-04-14
+ *       &missing=any            (photo | signature | id | address | any)
+ *       &limit=50
+ *       &offset=0
+ * </pre>
+ */
+@Path("/v1/ckyc/record-master")
+@Component
+@Tag(name = "CKYC Record Master",
+        description = "Active clients missing a CKYC number. Drives the 'Mark Non-CKYC' screen.")
+@RequiredArgsConstructor
+public class NonCkycClientApiResource {
+
+    private static final DateTimeFormatter ISO_DATE = DateTimeFormatter.ISO_LOCAL_DATE;
+
+    private final NonCkycClientReadPlatformService readService;
+    private final DefaultToApiJsonSerializer<Object> toApiJsonSerializer;
+
+    @GET
+    @Consumes({ MediaType.APPLICATION_JSON })
+    @Produces({ MediaType.APPLICATION_JSON })
+    @Operation(summary = "List active clients with missing CKYC number",
+            description = "Returns active clients whose m_client_kyc.ckyc_number is null or blank. "
+                    + "Supports filtering by branch (officeId), loan product, activation date range, "
+                    + "and missing-artefact type.")
+    @ApiResponse(responseCode = "200", description = "OK")
+    public String retrieveNonCkycClients(
+            @QueryParam("officeId") @Parameter(description = "Branch / office filter") final Long officeId,
+            @QueryParam("productId") @Parameter(description = "Loan product filter") final Long productId,
+            @QueryParam("fromDate") @Parameter(description = "Activation date >= (yyyy-MM-dd)") final String fromDate,
+            @QueryParam("toDate") @Parameter(description = "Activation date <= (yyyy-MM-dd)") final String toDate,
+            @QueryParam("missing") @Parameter(
+                    description = "Missing artefact filter: photo | signature | id | address | any") final String missing,
+            @QueryParam("limit") @Parameter(description = "Page size (default 50)") final Integer limit,
+            @QueryParam("offset") @Parameter(description = "Page offset (default 0)") final Integer offset) {
+
+        final LocalDate from = parseDate(fromDate);
+        final LocalDate to = parseDate(toDate);
+        final Integer effectiveLimit = limit == null ? 50 : limit;
+        final Integer effectiveOffset = offset == null ? 0 : offset;
+
+        final List<NonCkycClientData> rows = readService.retrieveNonCkycClients(officeId, productId, from, to, missing,
+                effectiveLimit, effectiveOffset);
+
+        final Map<String, Object> response = new LinkedHashMap<>();
+        response.put("totalFilteredRecords", rows.size());
+        response.put("limit", effectiveLimit);
+        response.put("offset", effectiveOffset);
+        response.put("pageItems", rows);
+        return toApiJsonSerializer.serialize(response);
+    }
+
+    private LocalDate parseDate(final String value) {
+        if (value == null || value.isBlank()) {
+            return null;
+        }
+        return LocalDate.parse(value, ISO_DATE);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/CkycBatchAckReceipt.java

@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * ACK receipt returned by the (dummy) CERSAI SFTP server after accepting a CKYC batch.
+ *
+ * <p>Shape is modelled on CERSAI's real response envelope so the production SFTP client can
+ * populate the same fields once it's swapped in. The UI should persist {@code cersaiReference}
+ * and {@code receivedAt} for audit, and display {@code perClientCkycNumbers} to the operator.
+ */
+@Data
+@Builder
+public class CkycBatchAckReceipt {
+
+    /** SUCCESS / FAILED / ALREADY_UPLOADED */
+    private String status;
+
+    /** Our registry id */
+    private Long batchRegistryId;
+
+    /** Our batch id ("BATCH-2026-04-15-001") */
+    private String batchId;
+
+    /** The ZIP filename that was pushed */
+    private String filename;
+
+    /** The absolute path the dummy server wrote the ZIP to */
+    private String dummyServerPath;
+
+    /** CERSAI-side reference (dummy: "CERSAI-ACK-{yyyyMMddHHmmss}-{seq}") */
+    private String cersaiReference;
+
+    /** Wall-clock time CERSAI reported accepting the file */
+    private LocalDateTime receivedAt;
+
+    /** Date we flipped m_ckyc_batch.uploaded_on */
+    private LocalDate uploadedOn;
+
+    /** Local checksum written at create-time */
+    private String checksumSent;
+
+    /** "Remote" checksum the dummy server echoes back — matches checksumSent if the file wasn't tampered with */
+    private String checksumVerified;
+
+    /** Total manifest records shipped in the ZIP */
+    private Integer totalRecords;
+
+    /** Records CERSAI accepted (dummy: same as totalRecords) */
+    private Integer processedRecords;
+
+    /** Per-client CKYC numbers issued by CERSAI. Keys: clientId, cifNumber, ckycNumber, status. */
+    private List<Map<String, Object>> perClientCkycNumbers;
+
+    /** Optional operator-facing message */
+    private String remarks;
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/CkycBatchCreateResponse.java

@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc;
+
+import java.time.LocalDate;
+import java.util.List;
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * Response returned by the CKYC batch-create endpoint — everything the UI needs to render
+ * the "batch created" dialog and audit row.
+ */
+@Data
+@Builder
+public class CkycBatchCreateResponse {
+
+    private Long id;
+    private String batchId;
+    private String filename;
+    private String checksum;
+    private Integer recordCount;
+    private Long sizeBytes;
+    private LocalDate createdOn;
+    private Long documentId;
+    private Boolean uploadedToCersai;
+    private List<CkycBatchManifestEntry> manifest;
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/CkycBatchManifestEntry.java

@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import lombok.Data;
+
+/**
+ * One manifest entry in a CKYC batch ZIP — reports which artefacts were found for a client and
+ * whether the client was ultimately included in the ZIP.
+ */
+@Data
+public class CkycBatchManifestEntry {
+
+    private Long clientId;
+    private String cifNumber;
+    private String displayName;
+    private boolean included;
+    private String skipReason;
+
+    private Map<String, Object> photo = new LinkedHashMap<>();
+    private Map<String, Object> signature = new LinkedHashMap<>();
+    private Map<String, Object> idProof = new LinkedHashMap<>();
+    private Map<String, Object> addressProof = new LinkedHashMap<>();
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/CkycBatchSummaryData.java

@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc;
+
+import java.time.LocalDate;
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * Row shape for the "CKYC batches — date wise" list screen.
+ */
+@Data
+@Builder
+public class CkycBatchSummaryData {
+
+    private Long id;
+    private String batchId;
+    private String filename;
+    private String checksum;
+    private Integer recordCount;
+    private Long sizeBytes;
+    private LocalDate createdOn;
+    private Long documentId;
+    private Boolean uploadedToCersai;
+    private LocalDate uploadedOn;
+    private String uploadStatus;
+    private String uploadRemarks;
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/CkycSftpUploadService.java

@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc;
+
+import org.apache.fineract.portfolio.client.domain.ClientKyc;
+
+/**
+ * Pushes a verified KYC record to CERSAI (Central Registry of Securitisation Asset Reconstruction
+ * and Security Interest of India) and returns the issued 14-digit CKYC number.
+ *
+ * <p>In production this will package records into a CERSAI-compliant SFTP batch file, push it to
+ * the configured SFTP endpoint, wait for the response file, and parse out the issued CKYC numbers.
+ * Until the real CERSAI SFTP credentials / file spec are in place, {@link DummyCkycSftpUploadServiceImpl}
+ * provides a stub that deterministically generates a valid-looking CKYC number so the downstream
+ * flow (write to {@code m_client_kyc.ckyc_number}, update screens, remove from "Mark Non-CKYC" list)
+ * can be exercised end-to-end.
+ */
+public interface CkycSftpUploadService {
+
+    /**
+     * Upload the given KYC record to CERSAI (server-builds-the-batch flow).
+     *
+     * @param kyc the fully populated KYC record (should be VERIFIED before calling)
+     * @return the CERSAI response including the issued CKYC number and batch id
+     */
+    CkycUploadResult upload(ClientKyc kyc);
+
+    /**
+     * UI-uploads-a-prebuilt-ZIP flow: caller supplies the CERSAI-format ZIP bytes (already
+     * persisted via the document-management subsystem). The implementation should push the ZIP
+     * over SFTP and return the issued CKYC number.
+     *
+     * @param kyc         the KYC record the ZIP belongs to
+     * @param zipBytes    the full ZIP content as a byte array (bounded in size by caller)
+     * @param fileName    the original upload filename (for logging / batch-id derivation)
+     */
+    CkycUploadResult uploadZip(ClientKyc kyc, byte[] zipBytes, String fileName);
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/CkycUploadResult.java

@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc;
+
+import java.time.LocalDate;
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * Result returned by {@link CkycSftpUploadService} after pushing a client's KYC record to CERSAI.
+ *
+ * <p>Shape is intentionally close to the real CERSAI SFTP response so the dummy implementation
+ * can be swapped out later without changing any callers.
+ */
+@Data
+@Builder
+public class CkycUploadResult {
+
+    /** 14-digit CKYC number issued by CERSAI. */
+    private String ckycNumber;
+
+    /** Batch id returned by CERSAI (for reconciliation against the uploaded SFTP file). */
+    private String batchId;
+
+    /** Date the upload was accepted by CERSAI. */
+    private LocalDate uploadedOn;
+
+    /** {@code SUCCESS} or {@code FAILED}. */
+    private String status;
+
+    /** Human-readable status / error message from the vendor. */
+    private String remarks;
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/DummyCkycSftpUploadServiceImpl.java

@@ -0,0 +1,118 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc;
+
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.portfolio.client.domain.ClientKyc;
+import org.springframework.stereotype.Service;
+
+/**
+ * Dummy / stub implementation of {@link CkycSftpUploadService}.
+ *
+ * <p><b>Not for production.</b> Generates a deterministic 14-digit CKYC number derived from the
+ * client id so that:
+ * <ul>
+ *   <li>the upload flow is fully exercisable without real CERSAI SFTP credentials,</li>
+ *   <li>repeated uploads for the same client return the same CKYC number (idempotent stub),</li>
+ *   <li>downstream screens ("Mark Non-CKYC", CKYC record master) immediately reflect the change.</li>
+ * </ul>
+ *
+ * <h3>CKYC number format</h3>
+ * <pre>
+ *   position 1–2   : prefix (we use "10" for individuals — CERSAI's actual convention too)
+ *   position 3–12  : 10-digit zero-padded client id
+ *   position 13–14 : 2-digit check derived from sum-of-digits mod 100
+ * </pre>
+ *
+ * Once real CERSAI SFTP details are available, replace this bean with a production impl
+ * (e.g. {@code CersaiCkycSftpUploadServiceImpl}) — no callers need to change.
+ */
+@Slf4j
+@Service
+public class DummyCkycSftpUploadServiceImpl implements CkycSftpUploadService {
+
+    private static final DateTimeFormatter BATCH_ID_FORMAT = DateTimeFormatter.ofPattern("yyyyMMdd");
+    private static final AtomicLong BATCH_COUNTER = new AtomicLong(1L);
+
+    @Override
+    public CkycUploadResult upload(final ClientKyc kyc) {
+        log.warn("Using DUMMY CKYC SFTP upload — no real CERSAI push performed. kycId={}, clientId={}",
+                kyc.getId(), kyc.getClient().getId());
+
+        final String ckycNumber = generateCkycNumber(kyc.getClient().getId());
+        final String batchId = generateBatchId();
+        final LocalDate today = LocalDate.now();
+
+        return CkycUploadResult.builder()
+                .ckycNumber(ckycNumber)
+                .batchId(batchId)
+                .uploadedOn(today)
+                .status("SUCCESS")
+                .remarks("Dummy CKYC upload — replace with real CERSAI SFTP integration before go-live.")
+                .build();
+    }
+
+    @Override
+    public CkycUploadResult uploadZip(final ClientKyc kyc, final byte[] zipBytes, final String fileName) {
+        log.warn("Using DUMMY CKYC ZIP upload — no real CERSAI push performed. kycId={}, clientId={}, file={}, bytes={}",
+                kyc.getId(), kyc.getClient().getId(), fileName, zipBytes == null ? 0 : zipBytes.length);
+
+        // Magic-byte check: a valid ZIP starts with "PK\x03\x04" (0x50 0x4B 0x03 0x04).
+        if (zipBytes == null || zipBytes.length < 4 || zipBytes[0] != 0x50 || zipBytes[1] != 0x4B
+                || zipBytes[2] != 0x03 || zipBytes[3] != 0x04) {
+            return CkycUploadResult.builder()
+                    .status("FAILED")
+                    .remarks("Uploaded file is not a valid ZIP archive (bad magic bytes).")
+                    .uploadedOn(LocalDate.now())
+                    .build();
+        }
+
+        final String ckycNumber = generateCkycNumber(kyc.getClient().getId());
+        final String batchId = generateBatchId();
+
+        return CkycUploadResult.builder()
+                .ckycNumber(ckycNumber)
+                .batchId(batchId)
+                .uploadedOn(LocalDate.now())
+                .status("SUCCESS")
+                .remarks("Dummy CKYC ZIP upload accepted (" + zipBytes.length + " bytes). "
+                        + "Replace with real CERSAI SFTP integration before go-live.")
+                .build();
+    }
+
+    private String generateCkycNumber(final Long clientId) {
+        final String clientPart = String.format("%010d", clientId);
+        final String prefix = "10"; // CERSAI: 10xx = individual
+        final String base = prefix + clientPart;
+        int sum = 0;
+        for (int i = 0; i < base.length(); i++) {
+            sum += base.charAt(i) - '0';
+        }
+        final String check = String.format("%02d", sum % 100);
+        return base + check;
+    }
+
+    private String generateBatchId() {
+        return "DUMMYCKYC-" + BATCH_ID_FORMAT.format(LocalDate.now()) + "-"
+                + String.format("%06d", BATCH_COUNTER.getAndIncrement());
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/api/CkycBatchApiResource.java

@@ -0,0 +1,239 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.api;
+
+import com.google.gson.Gson;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.media.Content;
+import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.MediaType;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.exception.PlatformApiDataValidationException;
+import org.apache.fineract.infrastructure.core.serialization.DefaultToApiJsonSerializer;
+import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchAckReceipt;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchCreateResponse;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchManifestEntry;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchSummaryData;
+import org.apache.fineract.portfolio.client.ckyc.domain.CkycBatch;
+import org.apache.fineract.portfolio.client.ckyc.domain.CkycBatchRepository;
+import org.apache.fineract.portfolio.client.ckyc.service.CkycBatchBuilderService;
+import org.apache.fineract.portfolio.client.ckyc.service.CkycBatchReadService;
+import org.apache.fineract.portfolio.client.ckyc.service.CkycBatchSftpUploadService;
+import org.springframework.stereotype.Component;
+
+/**
+ * REST endpoints for CERSAI CKYC batch ZIPs.
+ *
+ * <pre>
+ *   POST   /v1/ckyc/batches                          – Create a ZIP for a list of clients
+ *   GET    /v1/ckyc/batches                          – List batches (flat)
+ *   GET    /v1/ckyc/batches?groupByDate=true         – List batches grouped by created_on
+ *   GET    /v1/ckyc/batches/{id}                     – Get one batch (includes manifest)
+ * </pre>
+ */
+@Slf4j
+@Path("/v1/ckyc/batches")
+@Component
+@Tag(name = "CKYC Batches",
+        description = "CERSAI CKYC batch ZIP assembly and registry. Used by the 'Upload to CKYC' screen.")
+@RequiredArgsConstructor
+public class CkycBatchApiResource {
+
+    private static final DateTimeFormatter ISO_DATE = DateTimeFormatter.ISO_LOCAL_DATE;
+
+    private final PlatformSecurityContext context;
+    private final CkycBatchBuilderService batchBuilderService;
+    private final CkycBatchReadService batchReadService;
+    private final CkycBatchSftpUploadService batchSftpUploadService;
+    private final CkycBatchRepository batchRepository;
+    private final DefaultToApiJsonSerializer<Object> toApiJsonSerializer;
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // CREATE – POST /v1/ckyc/batches
+    // ─────────────────────────────────────────────────────────────────────────
+
+    @POST
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    @Operation(summary = "Create a CERSAI CKYC batch ZIP",
+            description = "Assembles a ZIP from the specified clients' artefacts (photo flag, signature, "
+                    + "ID proof scan, address proof scan). Each document is renamed inside the ZIP as "
+                    + "`<CIFNumber>_<PI|PA>_<DOCTYPE>.<ext>`. Returns the manifest + registry row id. "
+                    + "The ZIP is persisted as an m_document row (parent_entity_type='ckyc_batch') and "
+                    + "a m_ckyc_batch audit row.")
+    @ApiResponse(responseCode = "200", description = "Batch created",
+            content = @Content(schema = @Schema(implementation = CkycBatchCreateResponse.class)))
+    @ApiResponse(responseCode = "400", description = "clientIds missing or empty")
+    public String createBatch(final String apiRequestBodyAsJson) {
+
+        context.authenticatedUser();
+
+        final Gson gson = new Gson();
+        @SuppressWarnings("unchecked")
+        final Map<String, Object> body = gson.fromJson(apiRequestBodyAsJson, Map.class);
+        if (body == null || !body.containsKey("clientIds")) {
+            throw new PlatformApiDataValidationException(
+                    "error.msg.ckyc.batch.clientIds.required",
+                    "clientIds array is required",
+                    java.util.Collections.emptyList());
+        }
+
+        @SuppressWarnings("unchecked")
+        final List<Object> rawClientIds = (List<Object>) body.get("clientIds");
+        final List<Long> clientIds = new ArrayList<>(rawClientIds.size());
+        for (final Object raw : rawClientIds) {
+            if (raw instanceof Number) {
+                clientIds.add(((Number) raw).longValue());
+            }
+        }
+
+        final String remarks = body.get("remarks") instanceof String ? (String) body.get("remarks") : null;
+
+        final CkycBatchCreateResponse response = batchBuilderService.createBatch(clientIds, remarks);
+        return toApiJsonSerializer.serialize(response);
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // LIST – GET /v1/ckyc/batches
+    // ─────────────────────────────────────────────────────────────────────────
+
+    @GET
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    @Operation(summary = "List CKYC batches",
+            description = "Returns batches sorted newest-first. Optionally filter by date range and "
+                    + "upload status. Pass `groupByDate=true` for a UI-friendly shape with per-date totals.")
+    public String listBatches(
+            @QueryParam("fromDate") @Parameter(description = "Created-on lower bound (yyyy-MM-dd)") final String fromDate,
+            @QueryParam("toDate") @Parameter(description = "Created-on upper bound (yyyy-MM-dd)") final String toDate,
+            @QueryParam("uploadedToCersai") @Parameter(description = "true = only uploaded, false = only pending") final Boolean uploaded,
+            @QueryParam("groupByDate") @Parameter(description = "Group result by created_on (default false)") final Boolean groupByDate) {
+
+        context.authenticatedUser();
+        final LocalDate from = parseDate(fromDate);
+        final LocalDate to = parseDate(toDate);
+
+        if (Boolean.TRUE.equals(groupByDate)) {
+            final List<Map<String, Object>> grouped = batchReadService.listBatchesGroupedByDate(from, to, uploaded);
+            final Map<String, Object> resp = new LinkedHashMap<>();
+            resp.put("groupCount", grouped.size());
+            resp.put("groups", grouped);
+            return toApiJsonSerializer.serialize(resp);
+        }
+
+        final List<CkycBatchSummaryData> rows = batchReadService.listBatches(from, to, uploaded);
+        final Map<String, Object> resp = new LinkedHashMap<>();
+        resp.put("totalFilteredRecords", rows.size());
+        resp.put("pageItems", rows);
+        return toApiJsonSerializer.serialize(resp);
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // DETAIL – GET /v1/ckyc/batches/{id}
+    // ─────────────────────────────────────────────────────────────────────────
+
+    @GET
+    @Path("{id}")
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    @Operation(summary = "Retrieve a CKYC batch with its manifest",
+            description = "Returns the batch registry row plus the parsed manifest (one entry per client).")
+    public String retrieveBatch(@PathParam("id") final Long id) {
+
+        context.authenticatedUser();
+        final CkycBatch batch = batchRepository.findById(id).orElseThrow(
+                () -> new IllegalArgumentException("CKYC batch not found: " + id));
+
+        final Map<String, Object> resp = new LinkedHashMap<>();
+        resp.put("id", batch.getId());
+        resp.put("batchId", batch.getBatchId());
+        resp.put("filename", batch.getFilename());
+        resp.put("checksum", batch.getChecksumSha256());
+        resp.put("recordCount", batch.getRecordCount());
+        resp.put("sizeBytes", batch.getSizeBytes());
+        resp.put("documentId", batch.getDocumentId());
+        resp.put("createdOn", batch.getCreatedOnDate());
+        resp.put("uploadedToCersai", batch.getUploadedToCersai());
+        resp.put("uploadedOn", batch.getUploadedOn());
+        resp.put("uploadStatus", batch.getUploadStatus());
+        resp.put("uploadRemarks", batch.getUploadRemarks());
+
+        if (batch.getManifestJson() != null) {
+            final CkycBatchManifestEntry[] manifest = new Gson().fromJson(batch.getManifestJson(),
+                    CkycBatchManifestEntry[].class);
+            resp.put("manifest", manifest);
+        }
+
+        return toApiJsonSerializer.serialize(resp);
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // SFTP UPLOAD – POST /v1/ckyc/batches/{id}/sftp-upload
+    //
+    // Dummy CERSAI SFTP push: copies the ZIP to the configured drop folder, generates an ACK
+    // reference, issues per-client CKYC numbers, and flips m_ckyc_batch.uploaded_to_cersai.
+    // Idempotent — re-uploading an already-uploaded batch returns ALREADY_UPLOADED.
+    // ─────────────────────────────────────────────────────────────────────────
+
+    @POST
+    @Path("{id}/sftp-upload")
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    @Operation(summary = "Upload a CKYC batch ZIP to the (dummy) CERSAI SFTP server",
+            description = "Copies the batch ZIP to the configured drop folder "
+                    + "(`fineract.ckyc.dummy-sftp.drop-folder`, default `~/.fineract/ckyc-sftp-drop`), "
+                    + "simulates the CERSAI ACK, issues a 14-digit CKYC number for each included client, "
+                    + "writes those numbers to m_client_kyc, and flips m_ckyc_batch.uploaded_to_cersai = true. "
+                    + "Safe to retry: a second call on an already-uploaded batch returns the original ACK "
+                    + "with status `ALREADY_UPLOADED`.")
+    @ApiResponse(responseCode = "200", description = "ACK receipt",
+            content = @Content(schema = @Schema(implementation = CkycBatchAckReceipt.class)))
+    @ApiResponse(responseCode = "400", description = "Batch not found")
+    @ApiResponse(responseCode = "503", description = "Dummy SFTP copy failed")
+    public String sftpUploadBatch(@PathParam("id") final Long id) {
+        context.authenticatedUser();
+        final CkycBatchAckReceipt ack = batchSftpUploadService.uploadBatch(id);
+        return toApiJsonSerializer.serialize(ack);
+    }
+
+    private LocalDate parseDate(final String value) {
+        if (value == null || value.isBlank()) {
+            return null;
+        }
+        return LocalDate.parse(value, ISO_DATE);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/domain/CkycBatch.java

@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.domain;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Lob;
+import jakarta.persistence.Table;
+import java.time.LocalDate;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.apache.fineract.infrastructure.core.domain.AbstractPersistableCustom;
+
+/**
+ * One row per CERSAI CKYC batch ZIP assembled by the bank for submission.
+ * The ZIP itself lives in {@code m_document} (parent_entity_type='ckyc_batch');
+ * this row is the audit trail + upload-status registry.
+ */
+@Entity
+@Table(name = "m_ckyc_batch")
+@Getter
+@Setter
+@NoArgsConstructor
+public class CkycBatch extends AbstractPersistableCustom<Long> {
+
+    @Column(name = "batch_id", length = 50, nullable = false, unique = true)
+    private String batchId;
+
+    @Column(name = "filename", length = 250, nullable = false)
+    private String filename;
+
+    @Column(name = "checksum_sha256", length = 80)
+    private String checksumSha256;
+
+    @Column(name = "record_count", nullable = false)
+    private Integer recordCount;
+
+    @Column(name = "size_bytes", nullable = false)
+    private Long sizeBytes;
+
+    /** FK → m_document.id — the persisted ZIP. */
+    @Column(name = "document_id")
+    private Long documentId;
+
+    @Column(name = "created_on", nullable = false)
+    private LocalDate createdOnDate;
+
+    @Column(name = "uploaded_to_cersai", nullable = false)
+    private Boolean uploadedToCersai = false;
+
+    @Column(name = "uploaded_on")
+    private LocalDate uploadedOn;
+
+    @Column(name = "upload_status", length = 20)
+    private String uploadStatus;
+
+    @Column(name = "upload_remarks", length = 500)
+    private String uploadRemarks;
+
+    @Lob
+    @Column(name = "manifest_json")
+    private String manifestJson;
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/domain/CkycBatchRepository.java

@@ -0,0 +1,27 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.domain;
+
+import java.util.Optional;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface CkycBatchRepository extends JpaRepository<CkycBatch, Long> {
+
+    Optional<CkycBatch> findByBatchId(String batchId);
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/service/CkycBatchBuilderService.java

@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.service;
+
+import java.util.List;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchCreateResponse;
+
+public interface CkycBatchBuilderService {
+
+    /**
+     * Assemble a CERSAI CKYC ZIP for the given clients, persist it as an {@code m_document} row
+     * and an {@code m_ckyc_batch} registry entry, and return the manifest.
+     *
+     * <p>Each client's documents are renamed in the ZIP using the pattern
+     * {@code <CIFNumber>_<PI|PA>_<DOCTYPE>.<ext>} — e.g. {@code 12345678_PI_AADHAAR.pdf},
+     * {@code 12345678_PA_ELECTRICITY.pdf}.
+     */
+    CkycBatchCreateResponse createBatch(List<Long> clientIds, String remarks);
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/service/CkycBatchBuilderServiceImpl.java

@@ -0,0 +1,419 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.service;
+
+import com.google.gson.Gson;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.fineract.infrastructure.documentmanagement.data.DocumentContent;
+import org.apache.fineract.infrastructure.documentmanagement.data.DocumentCreateRequest;
+import org.apache.fineract.infrastructure.documentmanagement.service.DocumentReadPlatformService;
+import org.apache.fineract.infrastructure.documentmanagement.service.DocumentWritePlatformService;
+import org.apache.fineract.infrastructure.documentmanagement.service.ImageReadPlatformService;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchCreateResponse;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchManifestEntry;
+import org.apache.fineract.portfolio.client.ckyc.domain.CkycBatch;
+import org.apache.fineract.portfolio.client.ckyc.domain.CkycBatchRepository;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class CkycBatchBuilderServiceImpl implements CkycBatchBuilderService {
+
+    private static final DateTimeFormatter BATCH_ID_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd");
+    /** ZIP filename timestamp — e.g. 2026-04-14-15-30-45. Hyphen-separated per spec. */
+    private static final DateTimeFormatter ZIP_NAME_TS_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd-HH-mm-ss");
+
+    /**
+     * Names of code values (from {@code KYCAddressProofType}) that unambiguously indicate an
+     * address proof. Aadhaar / Passport / DL / Voter ID can be both ID and address proof — we
+     * default them to ID (PI) unless the UI tags them explicitly.
+     */
+    private static final Set<String> ADDRESS_PROOF_ONLY = Set.of(
+            "UTILITY BILLS", "UTILITY BILL", "ELECTRICITY BILL", "WATER BILL", "GAS BILL", "INTERNET BILL",
+            "BANK ACCOUNT STATEMENT / PASSBOOK", "BANK STATEMENT", "POSTPAID MOBILE BILL",
+            "REGISTERED RENT AGREEMENT", "RENT AGREEMENT", "PROPERTY TAX RECEIPT",
+            "EMPLOYER LETTER (FOR SALARIED EMPLOYEES)", "EMPLOYER LETTER",
+            "PENSION PAYMENT ORDERS (FOR RETIREES)", "PENSION PAYMENT ORDER");
+
+    private final JdbcTemplate jdbcTemplate;
+    private final DocumentReadPlatformService documentReadPlatformService;
+    private final DocumentWritePlatformService documentWritePlatformService;
+    private final ImageReadPlatformService imageReadPlatformService;
+    private final CkycBatchRepository ckycBatchRepository;
+
+    @Transactional
+    @Override
+    public CkycBatchCreateResponse createBatch(final List<Long> clientIds, final String remarks) {
+
+        if (clientIds == null || clientIds.isEmpty()) {
+            throw new IllegalArgumentException("clientIds must not be empty");
+        }
+
+        final LocalDate today = LocalDate.now();
+        final int seq = nextBatchSequenceForToday(today);
+        final String batchId = "BATCH-" + BATCH_ID_FMT.format(today) + "-" + String.format("%03d", seq);
+        // ZIP filename: "CERSAI-ZIP-<yyyy-MM-dd-HH-mm-ss>.zip" — hyphen-separated per spec.
+        final String filename = "CERSAI-ZIP-" + ZIP_NAME_TS_FMT.format(java.time.LocalDateTime.now()) + ".zip";
+
+        final List<CkycBatchManifestEntry> manifest = new ArrayList<>();
+        final ByteArrayOutputStream zipBytesOut = new ByteArrayOutputStream();
+        int includedCount = 0;
+
+        try (ZipOutputStream zip = new ZipOutputStream(zipBytesOut)) {
+            for (final Long clientId : clientIds) {
+                final CkycBatchManifestEntry entry = processClient(clientId, zip);
+                manifest.add(entry);
+                if (entry.isIncluded()) {
+                    includedCount++;
+                }
+            }
+        } catch (final IOException ioe) {
+            throw new IllegalStateException("Failed to assemble CKYC ZIP: " + ioe.getMessage(), ioe);
+        }
+
+        final byte[] zipBytes = zipBytesOut.toByteArray();
+        final String checksum = "sha256:" + sha256Hex(zipBytes);
+
+        // Persist the ZIP as an m_document row (parent_entity_type='ckyc_batch', parent_entity_id=seq)
+        // Persistence happens BEFORE we know m_ckyc_batch.id, so we use the synthetic seq as a
+        // temporary parent_entity_id. The real FK is via m_ckyc_batch.document_id after save.
+        final var docResponse = documentWritePlatformService.createDocument(
+                DocumentCreateRequest.builder()
+                        .entityType("ckyc_batch")
+                        .entityId((long) seq)
+                        .name(batchId)
+                        .fileName(filename)
+                        .description(StringUtils.defaultIfEmpty(remarks, "CERSAI CKYC batch ZIP"))
+                        .size((long) zipBytes.length)
+                        .type("application/zip")
+                        .stream(new ByteArrayInputStream(zipBytes))
+                        .build());
+
+        // Persist the batch registry row
+        final CkycBatch batch = new CkycBatch();
+        batch.setBatchId(batchId);
+        batch.setFilename(filename);
+        batch.setChecksumSha256(checksum);
+        batch.setRecordCount(includedCount);
+        batch.setSizeBytes((long) zipBytes.length);
+        batch.setDocumentId(docResponse.getResourceId() != null ? docResponse.getResourceId().longValue() : null);
+        batch.setCreatedOnDate(today);
+        batch.setUploadedToCersai(false);
+        batch.setManifestJson(new Gson().toJson(manifest));
+
+        final CkycBatch saved = ckycBatchRepository.saveAndFlush(batch);
+
+        log.info("CKYC batch created: id={}, batchId={}, filename={}, recordCount={}/{}, bytes={}",
+                saved.getId(), batchId, filename, includedCount, clientIds.size(), zipBytes.length);
+
+        return CkycBatchCreateResponse.builder()
+                .id(saved.getId())
+                .batchId(batchId)
+                .filename(filename)
+                .checksum(checksum)
+                .recordCount(includedCount)
+                .sizeBytes((long) zipBytes.length)
+                .createdOn(today)
+                .documentId(saved.getDocumentId())
+                .uploadedToCersai(false)
+                .manifest(manifest)
+                .build();
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // Helpers
+    // ─────────────────────────────────────────────────────────────────────────
+
+    private CkycBatchManifestEntry processClient(final Long clientId, final ZipOutputStream zip) throws IOException {
+        final CkycBatchManifestEntry entry = new CkycBatchManifestEntry();
+        entry.setClientId(clientId);
+
+        // Fetch client header (cif, display name, image_id presence)
+        final Map<String, Object> client;
+        try {
+            client = jdbcTemplate.queryForMap(
+                    "select c.cif_number as cifNumber, c.display_name as displayName, c.image_id as imageId "
+                            + " from m_client c where c.id = ?",
+                    clientId);
+        } catch (final org.springframework.dao.EmptyResultDataAccessException ex) {
+            entry.setIncluded(false);
+            entry.setSkipReason("Client not found");
+            return entry;
+        }
+
+        final String cifNumber = (String) client.get("cifNumber");
+        final String displayName = (String) client.get("displayName");
+        final Long imageId = client.get("imageId") == null ? null : ((Number) client.get("imageId")).longValue();
+        entry.setCifNumber(cifNumber);
+        entry.setDisplayName(displayName);
+
+        // Photo — packaged as "<CIF>-PI-PHOTO.<ext>" using m_client.image_id → m_image
+        final Map<String, Object> photo = new LinkedHashMap<>();
+        photo.put("present", imageId != null);
+        entry.setPhoto(photo);
+        if (imageId != null && cifNumber != null) {
+            packageClientImage(zip, clientId, cifNumber + "-PI-PHOTO", photo);
+        }
+
+        // Signature — packaged as "<CIF>-PI-SIGNATURE.<ext>"
+        final Long signatureDocId = findSignatureDocId(clientId);
+        final Map<String, Object> signature = new LinkedHashMap<>();
+        signature.put("present", signatureDocId != null);
+        entry.setSignature(signature);
+        if (signatureDocId != null && cifNumber != null) {
+            packageFile(zip, "clients", clientId, signatureDocId,
+                    cifNumber + "-PI-SIGNATURE", signature);
+        }
+
+        // ID proof + Address proof: iterate m_client_identifier, classify by code value name.
+        // For each identifier, find the attached m_document (one-to-one, latest wins).
+        // NOTE: the `active` column in m_client_identifier stores the status code (200 = active)
+        // not a 1/0 boolean — so we filter on `status = 200` (ClientIdentifierStatus.ACTIVE).
+        final List<Map<String, Object>> identifiers = jdbcTemplate.queryForList(
+                "select ci.id as identifierId, ci.document_key as docKey, cv.code_value as codeValue "
+                        + " from m_client_identifier ci "
+                        + " join m_code_value cv on cv.id = ci.document_type_id "
+                        + " where ci.client_id = ? and (ci.status = 200 or ci.status is null)",
+                clientId);
+
+        Map<String, Object> idProof = new LinkedHashMap<>();
+        Map<String, Object> addressProof = new LinkedHashMap<>();
+        idProof.put("present", false);
+        addressProof.put("present", false);
+
+        for (final Map<String, Object> idRow : identifiers) {
+            final Long identifierId = ((Number) idRow.get("identifierId")).longValue();
+            final String codeValue = (String) idRow.get("codeValue");
+            final String docKey = (String) idRow.get("docKey");
+            final boolean isAddress = ADDRESS_PROOF_ONLY.contains(codeValue == null ? "" : codeValue.toUpperCase(Locale.ROOT));
+            final String role = isAddress ? "PA" : "PI";
+
+            final Long attachedDocId = findLatestDocForIdentifier(identifierId);
+            if (attachedDocId == null) {
+                continue; // identifier exists but no scan uploaded → skip
+            }
+
+            final String shortCode = toShortCode(codeValue);
+            final Map<String, Object> bucket = isAddress ? addressProof : idProof;
+            // First match of each role wins (a client may have multiple IDs — we pick the first one).
+            if (Boolean.TRUE.equals(bucket.get("present"))) {
+                continue;
+            }
+            bucket.clear();
+            bucket.put("present", true);
+            bucket.put("type", codeValue);
+            bucket.put("key", docKey);
+
+            if (cifNumber != null) {
+                packageFile(zip, "client_identifiers", identifierId, attachedDocId,
+                        cifNumber + "-" + role + "-" + shortCode, bucket);
+            }
+        }
+
+        entry.setIdProof(idProof);
+        entry.setAddressProof(addressProof);
+
+        // Inclusion rule: include the client in the batch only if at least one artefact was packaged.
+        final boolean hasAny = Boolean.TRUE.equals(photo.get("present"))
+                || Boolean.TRUE.equals(signature.get("present"))
+                || Boolean.TRUE.equals(idProof.get("present"))
+                || Boolean.TRUE.equals(addressProof.get("present"));
+        entry.setIncluded(hasAny && cifNumber != null);
+        if (!entry.isIncluded()) {
+            entry.setSkipReason(cifNumber == null ? "Client has no CIF number"
+                    : "No documents available to package");
+        }
+
+        return entry;
+    }
+
+    private Long findSignatureDocId(final Long clientId) {
+        final List<Long> ids = jdbcTemplate.queryForList(
+                "select id from m_document where parent_entity_type = 'clients' and parent_entity_id = ? "
+                        + " and name = 'clientSignature' order by id desc",
+                Long.class, clientId);
+        return ids.isEmpty() ? null : ids.get(0);
+    }
+
+    private Long findLatestDocForIdentifier(final Long identifierId) {
+        final List<Long> ids = jdbcTemplate.queryForList(
+                "select id from m_document where parent_entity_type = 'client_identifiers' and parent_entity_id = ? "
+                        + " order by id desc",
+                Long.class, identifierId);
+        return ids.isEmpty() ? null : ids.get(0);
+    }
+
+    /**
+     * Package the client's photo from {@code m_image} into the ZIP. Uses
+     * {@link ImageReadPlatformService#retrieveImage(String, Long)} which resolves the bytes
+     * from the active content store (filesystem or S3).
+     */
+    private void packageClientImage(final ZipOutputStream zip, final Long clientId,
+            final String zipEntryBaseName, final Map<String, Object> manifestBucket) throws IOException {
+
+        final DocumentContent content;
+        try {
+            content = imageReadPlatformService.retrieveImage("clients", clientId);
+        } catch (final Exception ex) {
+            log.warn("Could not read client image for clientId={}: {}", clientId, ex.getMessage());
+            manifestBucket.put("packaged", false);
+            manifestBucket.put("reason", "Image bytes unavailable: " + ex.getMessage());
+            return;
+        }
+        if (content == null || content.getStream() == null) {
+            manifestBucket.put("packaged", false);
+            manifestBucket.put("reason", "Image bytes unavailable");
+            return;
+        }
+
+        final String ext = guessExtension(content.getFileName(), content.getContentType());
+        final String zipEntryName = zipEntryBaseName + ext;
+        final ZipEntry zipEntry = new ZipEntry(zipEntryName);
+        zip.putNextEntry(zipEntry);
+        try (InputStream in = content.getStream()) {
+            final byte[] buf = new byte[8192];
+            int read;
+            long total = 0;
+            while ((read = in.read(buf)) != -1) {
+                zip.write(buf, 0, read);
+                total += read;
+            }
+            manifestBucket.put("packaged", true);
+            manifestBucket.put("zipEntry", zipEntryName);
+            manifestBucket.put("bytes", total);
+        } finally {
+            zip.closeEntry();
+        }
+    }
+
+    private void packageFile(final ZipOutputStream zip, final String entityType, final Long entityId,
+            final Long documentId, final String zipEntryBaseName, final Map<String, Object> manifestBucket) throws IOException {
+
+        final DocumentContent content = documentReadPlatformService.retrieveDocumentContent(entityType, entityId, documentId);
+        if (content == null || content.getStream() == null) {
+            manifestBucket.put("packaged", false);
+            manifestBucket.put("reason", "Document bytes unavailable");
+            return;
+        }
+
+        final String ext = guessExtension(content.getFileName(), content.getContentType());
+        final String zipEntryName = zipEntryBaseName + ext;
+
+        final ZipEntry zipEntry = new ZipEntry(zipEntryName);
+        zip.putNextEntry(zipEntry);
+        try (InputStream in = content.getStream()) {
+            final byte[] buf = new byte[8192];
+            int read;
+            long total = 0;
+            while ((read = in.read(buf)) != -1) {
+                zip.write(buf, 0, read);
+                total += read;
+            }
+            manifestBucket.put("packaged", true);
+            manifestBucket.put("zipEntry", zipEntryName);
+            manifestBucket.put("bytes", total);
+        } finally {
+            zip.closeEntry();
+        }
+    }
+
+    private String guessExtension(final String fileName, final String contentType) {
+        if (fileName != null && fileName.contains(".")) {
+            return fileName.substring(fileName.lastIndexOf('.'));
+        }
+        if (contentType == null) {
+            return ".bin";
+        }
+        return switch (contentType.toLowerCase(Locale.ROOT)) {
+            case "application/pdf" -> ".pdf";
+            case "image/jpeg" -> ".jpg";
+            case "image/png" -> ".png";
+            default -> ".bin";
+        };
+    }
+
+    private String toShortCode(final String codeValue) {
+        if (codeValue == null) {
+            return "DOC";
+        }
+        // "Utility Bills" → "UTILITYBILLS", "Electricity Bill" → "ELECTRICITY"
+        final String upper = codeValue.toUpperCase(Locale.ROOT);
+        if (upper.startsWith("ELECTRICITY")) return "ELECTRICITY";
+        if (upper.startsWith("WATER")) return "WATER";
+        if (upper.startsWith("GAS")) return "GAS";
+        if (upper.startsWith("INTERNET")) return "INTERNET";
+        if (upper.startsWith("BANK")) return "BANK";
+        if (upper.startsWith("POSTPAID")) return "MOBILE";
+        if (upper.startsWith("REGISTERED RENT") || upper.startsWith("RENT")) return "RENT";
+        if (upper.startsWith("PROPERTY TAX")) return "PROPERTYTAX";
+        if (upper.startsWith("EMPLOYER")) return "EMPLOYER";
+        if (upper.startsWith("PENSION")) return "PENSION";
+        if (upper.startsWith("PAN")) return "PAN";
+        if (upper.startsWith("AADHAAR")) return "AADHAAR";
+        if (upper.startsWith("PASSPORT")) return "PASSPORT";
+        if (upper.startsWith("VOTER")) return "VOTER";
+        if (upper.startsWith("DRIVING")) return "DL";
+        if (upper.startsWith("NREGA")) return "NREGA";
+        if (upper.startsWith("LETTER") || upper.startsWith("NPR")) return "NPR";
+        return upper.replaceAll("[^A-Z0-9]", "");
+    }
+
+    private int nextBatchSequenceForToday(final LocalDate today) {
+        final Integer max = jdbcTemplate.queryForObject(
+                "select count(1) from m_ckyc_batch where created_on = ?",
+                Integer.class, java.sql.Date.valueOf(today));
+        return (max == null ? 0 : max) + 1;
+    }
+
+    private String sha256Hex(final byte[] data) {
+        try {
+            final MessageDigest md = MessageDigest.getInstance("SHA-256");
+            final byte[] hash = md.digest(data);
+            final StringBuilder sb = new StringBuilder(hash.length * 2);
+            for (final byte b : hash) {
+                sb.append(String.format("%02x", b));
+            }
+            return sb.toString();
+        } catch (final NoSuchAlgorithmException e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/service/CkycBatchReadService.java

@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.service;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.infrastructure.core.domain.JdbcSupport;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchSummaryData;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Read service for the CKYC batches list screen — returns batches date-wise with upload status.
+ */
+@Service
+@RequiredArgsConstructor
+@Transactional(readOnly = true)
+public class CkycBatchReadService {
+
+    private final JdbcTemplate jdbcTemplate;
+
+    /**
+     * List CKYC batches, optionally filtered by date range and upload status, sorted newest-first.
+     *
+     * @param fromDate         inclusive lower bound on {@code created_on} (nullable)
+     * @param toDate           inclusive upper bound on {@code created_on} (nullable)
+     * @param uploadedToCersai tri-state filter: {@code true} = only uploaded, {@code false} = only pending, {@code null} = both
+     */
+    public List<CkycBatchSummaryData> listBatches(final LocalDate fromDate, final LocalDate toDate,
+            final Boolean uploadedToCersai) {
+
+        final StringBuilder sql = new StringBuilder(256);
+        final List<Object> args = new ArrayList<>();
+        final List<Integer> types = new ArrayList<>();
+
+        sql.append("select id, batch_id as batchId, filename, checksum_sha256 as checksum, record_count as recordCount, "
+                + " size_bytes as sizeBytes, document_id as documentId, created_on as createdOn, "
+                + " uploaded_to_cersai as uploadedToCersai, uploaded_on as uploadedOn, "
+                + " upload_status as uploadStatus, upload_remarks as uploadRemarks "
+                + " from m_ckyc_batch where 1 = 1 ");
+
+        if (fromDate != null) {
+            sql.append(" and created_on >= ? ");
+            args.add(java.sql.Date.valueOf(fromDate));
+            types.add(Types.DATE);
+        }
+        if (toDate != null) {
+            sql.append(" and created_on <= ? ");
+            args.add(java.sql.Date.valueOf(toDate));
+            types.add(Types.DATE);
+        }
+        if (uploadedToCersai != null) {
+            sql.append(" and uploaded_to_cersai = ? ");
+            args.add(uploadedToCersai);
+            types.add(Types.BOOLEAN);
+        }
+
+        sql.append(" order by created_on desc, id desc");
+
+        final int[] typeArr = types.stream().mapToInt(Integer::intValue).toArray();
+        return jdbcTemplate.query(sql.toString(), args.toArray(), typeArr, new BatchRowMapper());
+    }
+
+    /**
+     * Group the batch list by {@code created_on} — shape suited for a date-wise collapsible list
+     * in the UI. Each group contains the total record count and per-batch rows.
+     */
+    public List<Map<String, Object>> listBatchesGroupedByDate(final LocalDate fromDate, final LocalDate toDate,
+            final Boolean uploadedToCersai) {
+
+        final List<CkycBatchSummaryData> flat = listBatches(fromDate, toDate, uploadedToCersai);
+        final Map<LocalDate, List<CkycBatchSummaryData>> grouped = new java.util.LinkedHashMap<>();
+        for (final CkycBatchSummaryData row : flat) {
+            grouped.computeIfAbsent(row.getCreatedOn(), d -> new ArrayList<>()).add(row);
+        }
+
+        final List<Map<String, Object>> result = new ArrayList<>(grouped.size());
+        for (final Map.Entry<LocalDate, List<CkycBatchSummaryData>> entry : grouped.entrySet()) {
+            final Map<String, Object> group = new LinkedHashMap<>();
+            group.put("date", entry.getKey());
+            group.put("batchCount", entry.getValue().size());
+            group.put("totalRecordCount",
+                    entry.getValue().stream().mapToInt(b -> b.getRecordCount() == null ? 0 : b.getRecordCount()).sum());
+            group.put("totalSizeBytes",
+                    entry.getValue().stream().mapToLong(b -> b.getSizeBytes() == null ? 0L : b.getSizeBytes()).sum());
+            group.put("uploadedCount",
+                    entry.getValue().stream().filter(b -> Boolean.TRUE.equals(b.getUploadedToCersai())).count());
+            group.put("pendingCount",
+                    entry.getValue().stream().filter(b -> !Boolean.TRUE.equals(b.getUploadedToCersai())).count());
+            group.put("batches", entry.getValue());
+            result.add(group);
+        }
+        return result;
+    }
+
+    private static final class BatchRowMapper implements RowMapper<CkycBatchSummaryData> {
+        @Override
+        public CkycBatchSummaryData mapRow(final ResultSet rs, final int rowNum) throws SQLException {
+            return CkycBatchSummaryData.builder()
+                    .id(rs.getLong("id"))
+                    .batchId(rs.getString("batchId"))
+                    .filename(rs.getString("filename"))
+                    .checksum(rs.getString("checksum"))
+                    .recordCount(JdbcSupport.getInteger(rs, "recordCount"))
+                    .sizeBytes(JdbcSupport.getLong(rs, "sizeBytes"))
+                    .documentId(JdbcSupport.getLong(rs, "documentId"))
+                    .createdOn(JdbcSupport.getLocalDate(rs, "createdOn"))
+                    .uploadedToCersai(rs.getBoolean("uploadedToCersai"))
+                    .uploadedOn(JdbcSupport.getLocalDate(rs, "uploadedOn"))
+                    .uploadStatus(rs.getString("uploadStatus"))
+                    .uploadRemarks(rs.getString("uploadRemarks"))
+                    .build();
+        }
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/service/CkycBatchSftpUploadService.java

@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.service;
+
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchAckReceipt;
+
+/**
+ * Pushes an assembled CKYC batch ZIP to the CERSAI SFTP drop, waits for the ACK, and persists the
+ * returned CKYC numbers on {@code m_client_kyc}.
+ *
+ * <p>Currently backed by {@link DummyCkycBatchSftpUploadServiceImpl} which simulates the SFTP
+ * server by copying the ZIP to a local folder and generating a deterministic ACK. When the real
+ * CERSAI SFTP credentials are available, replace the bean with a production implementation — no
+ * callers need to change.
+ */
+public interface CkycBatchSftpUploadService {
+
+    /**
+     * Upload the batch with the given {@code m_ckyc_batch.id}.
+     *
+     * <p>Idempotent: if the batch has already been uploaded ({@code uploaded_to_cersai = true}),
+     * returns the existing ACK metadata with {@code status = "ALREADY_UPLOADED"} — does NOT
+     * push again.
+     *
+     * @param batchRegistryId the primary key of the batch row
+     * @return the ACK receipt
+     */
+    CkycBatchAckReceipt uploadBatch(Long batchRegistryId);
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/ckyc/service/DummyCkycBatchSftpUploadServiceImpl.java

@@ -0,0 +1,257 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.ckyc.service;
+
+import com.google.gson.Gson;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.documentmanagement.data.DocumentContent;
+import org.apache.fineract.infrastructure.documentmanagement.service.DocumentReadPlatformService;
+import org.apache.fineract.portfolio.client.ckyc.CkycBatchAckReceipt;
+import org.apache.fineract.portfolio.client.ckyc.domain.CkycBatch;
+import org.apache.fineract.portfolio.client.ckyc.domain.CkycBatchRepository;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Dummy / stub implementation of {@link CkycBatchSftpUploadService}.
+ *
+ * <p><b>Not for production.</b> Simulates the CERSAI SFTP server by:
+ * <ol>
+ *   <li>Reading the ZIP bytes from {@code m_document} (FK from {@code m_ckyc_batch.document_id}),</li>
+ *   <li>Copying them to a local "drop folder" — default {@code ~/.fineract/ckyc-sftp-drop}, overridable
+ *       via {@code fineract.ckyc.dummy-sftp.drop-folder},</li>
+ *   <li>Recomputing the SHA-256 on the written file to simulate CERSAI's checksum verification,</li>
+ *   <li>Generating a deterministic 14-digit CKYC number per client from the manifest
+ *       ({@code "10" + 10-digit zero-padded clientId + 2-digit check}),</li>
+ *   <li>Updating each client's {@code m_client_kyc} row with the issued CKYC number and upload date,</li>
+ *   <li>Flipping {@code m_ckyc_batch.uploaded_to_cersai = true} and recording the ACK reference.</li>
+ * </ol>
+ *
+ * <p>Once the real CERSAI SFTP spec is available, replace this bean with a production impl that
+ * opens an SFTP session, PUTs the ZIP to the CERSAI drop, waits for the response file, parses the
+ * real per-client CKYC numbers, and performs the same downstream writes — <b>no callers change.</b>
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DummyCkycBatchSftpUploadServiceImpl implements CkycBatchSftpUploadService {
+
+    private static final DateTimeFormatter ACK_TS_FMT = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
+    private static final AtomicInteger ACK_COUNTER = new AtomicInteger(1);
+
+    private final CkycBatchRepository ckycBatchRepository;
+    private final DocumentReadPlatformService documentReadPlatformService;
+    private final JdbcTemplate jdbcTemplate;
+
+    /** Local folder the dummy server drops the ZIP into. Simulates the CERSAI SFTP landing zone. */
+    @Value("${fineract.ckyc.dummy-sftp.drop-folder:#{systemProperties['user.home'] + '/.fineract/ckyc-sftp-drop'}}")
+    private String dropFolder;
+
+    @Transactional
+    @Override
+    public CkycBatchAckReceipt uploadBatch(final Long batchRegistryId) {
+
+        final CkycBatch batch = ckycBatchRepository.findById(batchRegistryId)
+                .orElseThrow(() -> new IllegalArgumentException("CKYC batch not found: " + batchRegistryId));
+
+        // Idempotent short-circuit
+        if (Boolean.TRUE.equals(batch.getUploadedToCersai())) {
+            log.info("CKYC batch {} already uploaded on {} — returning existing ACK", batch.getBatchId(), batch.getUploadedOn());
+            return CkycBatchAckReceipt.builder()
+                    .status("ALREADY_UPLOADED")
+                    .batchRegistryId(batch.getId())
+                    .batchId(batch.getBatchId())
+                    .filename(batch.getFilename())
+                    .cersaiReference(batch.getUploadRemarks())
+                    .uploadedOn(batch.getUploadedOn())
+                    .totalRecords(batch.getRecordCount())
+                    .processedRecords(batch.getRecordCount())
+                    .checksumSent(batch.getChecksumSha256())
+                    .checksumVerified(batch.getChecksumSha256())
+                    .remarks("Batch already uploaded; no action taken.")
+                    .build();
+        }
+
+        if (batch.getDocumentId() == null) {
+            throw new IllegalStateException("CKYC batch " + batch.getBatchId() + " has no linked document");
+        }
+
+        // 1) Resolve m_document row so we can read back its bytes
+        final Map<String, Object> docRow;
+        try {
+            docRow = jdbcTemplate.queryForMap(
+                    "select parent_entity_type as entityType, parent_entity_id as entityId, location, file_name as fileName "
+                            + " from m_document where id = ?",
+                    batch.getDocumentId());
+        } catch (final org.springframework.dao.EmptyResultDataAccessException ex) {
+            throw new IllegalStateException("m_document row missing for CKYC batch " + batch.getBatchId(), ex);
+        }
+        final String docEntityType = (String) docRow.get("entityType");
+        final Long docEntityId = docRow.get("entityId") == null ? null : ((Number) docRow.get("entityId")).longValue();
+
+        // 2) Copy ZIP bytes into the dummy drop folder + recompute checksum
+        final Path targetPath;
+        final String verifiedChecksum;
+        try {
+            final DocumentContent content = documentReadPlatformService.retrieveDocumentContent(docEntityType, docEntityId,
+                    batch.getDocumentId());
+            if (content == null || content.getStream() == null) {
+                throw new IllegalStateException("Could not read ZIP bytes for batch " + batch.getBatchId());
+            }
+
+            final Path dropDir = Paths.get(dropFolder);
+            Files.createDirectories(dropDir);
+            targetPath = dropDir.resolve(batch.getFilename());
+
+            try (InputStream in = content.getStream()) {
+                Files.copy(in, targetPath, StandardCopyOption.REPLACE_EXISTING);
+            }
+            verifiedChecksum = "sha256:" + sha256Hex(Files.readAllBytes(targetPath));
+            log.warn("DUMMY CERSAI SFTP — ZIP copied to {} ({} bytes)", targetPath, Files.size(targetPath));
+        } catch (final IOException ioe) {
+            throw new IllegalStateException("Dummy SFTP copy failed: " + ioe.getMessage(), ioe);
+        }
+
+        // 3) Build the ACK reference and timestamp
+        final LocalDateTime now = LocalDateTime.now();
+        final String ackReference = "CERSAI-ACK-" + ACK_TS_FMT.format(now) + "-"
+                + String.format("%04d", ACK_COUNTER.getAndIncrement());
+        final LocalDate today = LocalDate.now();
+
+        // 4) Parse manifest → generate a CKYC number per included client → persist on m_client_kyc
+        final List<Map<String, Object>> perClient = new ArrayList<>();
+        if (batch.getManifestJson() != null) {
+            @SuppressWarnings("unchecked")
+            final List<Map<String, Object>> manifest = new Gson().fromJson(batch.getManifestJson(), List.class);
+            for (final Map<String, Object> entry : manifest) {
+                if (!Boolean.TRUE.equals(entry.get("included"))) {
+                    continue;
+                }
+                final Long clientId = entry.get("clientId") == null ? null
+                        : ((Number) entry.get("clientId")).longValue();
+                final String cifNumber = (String) entry.get("cifNumber");
+                if (clientId == null) {
+                    continue;
+                }
+
+                final String ckycNumber = generateCkycNumber(clientId);
+
+                // Update the client's KYC row — idempotent via WHERE clause.
+                final int updated = jdbcTemplate.update(
+                        "update m_client_kyc set ckyc_number = ?, ckyc_uploaded_on = ?, ckyc_batch_id = ? "
+                                + " where client_id = ? and (ckyc_number is null or trim(ckyc_number) = '')",
+                        ckycNumber, java.sql.Date.valueOf(today), batch.getBatchId(), clientId);
+
+                final Map<String, Object> clientAck = new LinkedHashMap<>();
+                clientAck.put("clientId", clientId);
+                clientAck.put("cifNumber", cifNumber);
+                clientAck.put("ckycNumber", ckycNumber);
+                clientAck.put("status", updated > 0 ? "ISSUED" : "ALREADY_HAD_CKYC");
+                perClient.add(clientAck);
+
+                log.info("DUMMY CERSAI ACK — clientId={}, cif={}, ckyc={}, kycRowsUpdated={}",
+                        clientId, cifNumber, ckycNumber, updated);
+            }
+        }
+
+        // 5) Flip the batch row
+        batch.setUploadedToCersai(true);
+        batch.setUploadedOn(today);
+        batch.setUploadStatus("SUCCESS");
+        batch.setUploadRemarks(ackReference);
+        ckycBatchRepository.saveAndFlush(batch);
+
+        return CkycBatchAckReceipt.builder()
+                .status("SUCCESS")
+                .batchRegistryId(batch.getId())
+                .batchId(batch.getBatchId())
+                .filename(batch.getFilename())
+                .dummyServerPath(targetPath.toAbsolutePath().toString())
+                .cersaiReference(ackReference)
+                .receivedAt(now)
+                .uploadedOn(today)
+                .checksumSent(batch.getChecksumSha256())
+                .checksumVerified(verifiedChecksum)
+                .totalRecords(batch.getRecordCount())
+                .processedRecords(perClient.size())
+                .perClientCkycNumbers(perClient)
+                .remarks("Dummy CERSAI SFTP upload — replace with real integration before go-live.")
+                .build();
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // Helpers
+    // ─────────────────────────────────────────────────────────────────────────
+
+    /**
+     * Deterministic 14-digit CKYC number — same algorithm as
+     * {@link DummyCkycSftpUploadServiceImpl} so both upload paths produce consistent values
+     * for the same clientId.
+     */
+    private String generateCkycNumber(final Long clientId) {
+        final String clientPart = String.format("%010d", clientId);
+        final String prefix = "10"; // 10xx = individual per CERSAI convention
+        final String base = prefix + clientPart;
+        int sum = 0;
+        for (int i = 0; i < base.length(); i++) {
+            sum += base.charAt(i) - '0';
+        }
+        return base + String.format("%02d", sum % 100);
+    }
+
+    private String sha256Hex(final byte[] data) {
+        try {
+            final MessageDigest md = MessageDigest.getInstance("SHA-256");
+            final byte[] hash = md.digest(data);
+            final StringBuilder sb = new StringBuilder(hash.length * 2);
+            for (final byte b : hash) {
+                sb.append(String.format("%02x", b));
+            }
+            return sb.toString();
+        } catch (final NoSuchAlgorithmException e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    /** Kept so IDE doesn't flag Optional import as unused after refactors. */
+    @SuppressWarnings("unused")
+    private Optional<CkycBatch> unused() {
+        return Optional.empty();
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/data/ClientKycApiConstants.java

@@ -55,6 +55,7 @@ public final class ClientKycApiConstants {
     public static final String dinNumberParam = "dinNumber";
     public static final String verifyCinParam = "verifyCin";
     public static final String verifyDinParam = "verifyDin";
+    public static final String entityNameParam = "entityName";
     public static final String incorporationDateParam = "incorporationDate";
     public static final String kycDocumentNameParam = "kycDocumentName";
     public static final String kycDocumentNumberParam = "kycDocumentNumber";
@@ -115,8 +116,8 @@ public final class ClientKycApiConstants {
     /** verify-cin command: CIN is already stored; no extra body params required. */
     public static final Set<String> VERIFY_CIN_REQUEST_PARAMS = new HashSet<>();
 
-    /** verify-din command: DIN is already stored; no extra body params required. */
-    public static final Set<String> VERIFY_DIN_REQUEST_PARAMS = new HashSet<>();
+    /** verify-din command: DIN and non-individual entity name are supplied in the request body. */
+    public static final Set<String> VERIFY_DIN_REQUEST_PARAMS = new HashSet<>(Arrays.asList(dinNumberParam, entityNameParam));
 
     public static final Set<String> REJECT_REQUEST_PARAMS = new HashSet<>(Arrays.asList(remarksParam));
 

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/data/ClientKycData.java

@@ -94,6 +94,8 @@ public final class ClientKycData {
 
     // ── CKYCRR ────────────────────────────────────────────────────────────────
     private final String ckycNumber;
+    private final LocalDate ckycUploadedOn;
+    private final String ckycBatchId;
 
     // ── Lifecycle ─────────────────────────────────────────────────────────────
     private final LocalDate initiatedOn;

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/data/NonCkycClientData.java

@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.data;
+
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * Row shape for the "Mark Non-CKYC" screen — active clients that do not yet have a CKYC number
+ * on their {@code m_client_kyc} record, together with flags for the missing KYC artefacts
+ * (photo / signature / id proof / address proof).
+ */
+@Data
+@Builder
+public class NonCkycClientData {
+
+    private Long clientId;
+    private String accountNo;
+    private String fullName;
+    private String email;
+    private String mobile;
+    private String address;
+    private Integer status;
+    private String statusCode;
+
+    private boolean hasPhoto;
+    private boolean hasSignature;
+    private boolean hasIdProof;
+    private boolean hasAddressProof;
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/handler/CompleteRekycClientKycCommandHandler.java

@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.handler;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.commands.annotation.CommandType;
+import org.apache.fineract.commands.handler.NewCommandSourceHandler;
+import org.apache.fineract.infrastructure.core.api.JsonCommand;
+import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
+import org.apache.fineract.portfolio.client.service.ClientKycWritePlatformService;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Handles the {@code complete-rekyc} KYC command.
+ *
+ * <p>Marks a periodic Re-KYC as complete: updates verifiedOn and recomputes the
+ * expiryDate from the current risk category.
+ */
+@Service
+@CommandType(entity = "CLIENTKYC", action = "COMPLETEREKYC")
+@RequiredArgsConstructor
+public class CompleteRekycClientKycCommandHandler implements NewCommandSourceHandler {
+
+    private final ClientKycWritePlatformService clientKycWritePlatformService;
+
+    @Transactional
+    @Override
+    public CommandProcessingResult processCommand(final JsonCommand command) {
+        return clientKycWritePlatformService.completeRekyc(
+                command.getClientId(), command.entityId(), command);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/handler/UploadCkycClientKycCommandHandler.java

@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.handler;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.commands.annotation.CommandType;
+import org.apache.fineract.commands.handler.NewCommandSourceHandler;
+import org.apache.fineract.infrastructure.core.api.JsonCommand;
+import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
+import org.apache.fineract.portfolio.client.service.ClientKycWritePlatformService;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Handles the {@code upload-ckyc} KYC command.
+ *
+ * <p>Pushes the KYC record to CERSAI via SFTP (currently a dummy stub) and persists the returned
+ * 14-digit CKYC number, batch id, and upload date on the {@code m_client_kyc} row.
+ */
+@Service
+@CommandType(entity = "CLIENTKYC", action = "UPLOADCKYC")
+@RequiredArgsConstructor
+public class UploadCkycClientKycCommandHandler implements NewCommandSourceHandler {
+
+    private final ClientKycWritePlatformService clientKycWritePlatformService;
+
+    @Transactional
+    @Override
+    public CommandProcessingResult processCommand(final JsonCommand command) {
+        return clientKycWritePlatformService.uploadCkyc(command.getClientId(), command.entityId(), command);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/service/ClientKycReadPlatformServiceImpl.java

@@ -101,6 +101,8 @@ public class ClientKycReadPlatformServiceImpl implements ClientKycReadPlatformSe
                 .form15Type(kyc.getForm15Type())
                 .form15Fy(kyc.getForm15Fy())
                 .ckycNumber(kyc.getCkycNumber())
+                .ckycUploadedOn(kyc.getCkycUploadedOn())
+                .ckycBatchId(kyc.getCkycBatchId())
                 .initiatedOn(kyc.getInitiatedOn())
                 .verifiedOn(kyc.getVerifiedOn())
                 .rejectedOn(kyc.getRejectedOn())

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/service/ClientKycWritePlatformService.java

@@ -81,4 +81,35 @@ public interface ClientKycWritePlatformService {
 
     /** Reject the KYC; stores a rejection reason. */
     CommandProcessingResult rejectKyc(Long clientId, Long kycId, JsonCommand command);
+
+    /**
+     * Complete a periodic Re-KYC: sets verifiedOn = today and expiryDate based on the
+     * current risk category (Low=10y, Medium=8y, High=2y), transitions status to VERIFIED.
+     */
+    CommandProcessingResult completeRekyc(Long clientId, Long kycId, JsonCommand command);
+
+    /**
+     * Push the KYC record to CERSAI (SFTP batch) and persist the returned 14-digit CKYC number,
+     * batch id, and upload date on the {@code m_client_kyc} row. Does not change {@code kycStatus}.
+     *
+     * <p>Currently backed by a dummy stub — the real CERSAI SFTP integration will replace the
+     * {@code CkycSftpUploadService} bean when credentials are available.
+     */
+    CommandProcessingResult uploadCkyc(Long clientId, Long kycId, JsonCommand command);
+
+    /**
+     * Pipeline 2 — UI uploads a pre-built CERSAI ZIP. The ZIP bytes are persisted via the
+     * document-management subsystem (as an {@code m_document} row with
+     * {@code parent_entity_type='client_kyc'}), passed to the SFTP upload service, and on success
+     * the returned 14-digit CKYC number is saved on the KYC row.
+     *
+     * @param clientId  client id (path param)
+     * @param kycId     KYC record id (path param)
+     * @param stream    raw ZIP bytes from the multipart form part
+     * @param fileName  original upload filename (used for validation + logging)
+     * @param fileSize  declared size in bytes from the Content-Length / multipart header
+     * @param remarks   optional operator remarks
+     */
+    CommandProcessingResult uploadCkycZip(Long clientId, Long kycId, java.io.InputStream stream, String fileName,
+            Long fileSize, String remarks);
 }

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/service/ClientKycWritePlatformServiceJpaRepositoryImpl.java

@@ -68,6 +68,11 @@ public class ClientKycWritePlatformServiceJpaRepositoryImpl implements ClientKyc
     private final ClientKycCommandFromApiJsonDeserializer deserializer;
     private final KycVerificationService kycVerificationService;
     private final FromJsonHelper fromApiJsonHelper;
+    private final org.apache.fineract.portfolio.client.ckyc.CkycSftpUploadService ckycSftpUploadService;
+    private final org.apache.fineract.infrastructure.documentmanagement.service.DocumentWritePlatformService documentWritePlatformService;
+
+    /** Hard cap on uploaded CKYC ZIP size to keep the byte[] copy bounded (50 MB). */
+    private static final int CKYC_ZIP_MAX_BYTES = 50 * 1024 * 1024;
 
     @Transactional
     @Override
@@ -79,6 +84,11 @@ public class ClientKycWritePlatformServiceJpaRepositoryImpl implements ClientKyc
                 command.json(), ClientKycApiConstants.INITIATE_REQUEST_PARAMS);
         kycCommand.validateForInitiate();
 
+        // CIN / GSTIN are only applicable for NON_INDIVIDUAL clients
+        final Integer kycTypeOnInitiate = command.integerValueOfParameterNamed("kycType");
+        final boolean isIndividualInitiate = kycTypeOnInitiate != null && kycTypeOnInitiate == 1;
+        enforceNonIndividualOnlyFields(command, isIndividualInitiate);
+
         if (kycRepositoryWrapper.existsByClientId(clientId)) {
             LOG.info("KYC record already exists for client {}. Returning existing record.", clientId);
             final ClientKyc existing = kycRepositoryWrapper.findByClientIdWithNotFoundDetection(clientId);
@@ -121,6 +131,10 @@ public class ClientKycWritePlatformServiceJpaRepositoryImpl implements ClientKyc
         final Client client = clientRepository.findOneWithNotFoundDetection(clientId);
         final ClientKyc kyc = kycRepositoryWrapper.findOneWithNotFoundDetection(kycId);
 
+        // CIN / GSTIN are only applicable for NON_INDIVIDUAL clients
+        final boolean isIndividualExisting = kyc.getKycType() != null && kyc.getKycType().isIndividual();
+        enforceNonIndividualOnlyFields(command, isIndividualExisting);
+
         final ClientKycStatus currentStatus = kyc.getKycStatus();
 
         // When re-editing a VERIFIED record, remarks are mandatory (audit trail)
@@ -526,13 +540,17 @@ public class ClientKycWritePlatformServiceJpaRepositoryImpl implements ClientKyc
             throw new ClientKycInvalidStateTransitionException("verify-din", kyc.getKycStatus().name());
         }
 
-        if (kyc.getDinNumber() == null) {
-            throw new PlatformServiceUnavailableException(
-                    "error.msg.client.kyc.din.not.found",
-                    "DIN number is not set on this KYC record. Update the KYC record first.");
+        // DIN + non-individual entity name are supplied in the request body.
+        final String dinNumber = command.stringValueOfParameterNamed(ClientKycApiConstants.dinNumberParam);
+        if (StringUtils.isBlank(dinNumber)) {
+            throw new PlatformApiDataValidationException(List.of(ApiParameterError.parameterError(
+                    "error.msg.client.kyc.din.required",
+                    "DIN number is required in the request body for verify-din.",
+                    ClientKycApiConstants.dinNumberParam)));
         }
+        final String entityName = command.stringValueOfParameterNamed(ClientKycApiConstants.entityNameParam);
 
-        final KycVerificationResult result = kycVerificationService.verifyDin(kyc.getDinNumber());
+        final KycVerificationResult result = kycVerificationService.verifyDin(dinNumber);
 
         if (!result.isSuccess()) {
             throw new PlatformServiceUnavailableException(
@@ -541,17 +559,161 @@ public class ClientKycWritePlatformServiceJpaRepositoryImpl implements ClientKyc
                     result.getErrorMessage());
         }
 
-        kyc.markDinVerified();
+        // Extract the list of companies returned by MCA and match against the supplied entity name
+        final String rawMcaResponse = result.getRawResponse();
+        final List<Map<String, Object>> companies = extractDirectorshipCompanies(rawMcaResponse);
+        String matchedCompany = null;
+        if (StringUtils.isNotBlank(entityName)) {
+            final String entityNorm = normalize(entityName);
+            for (final Map<String, Object> c : companies) {
+                final String name = (String) c.get("companyName");
+                if (name != null) {
+                    final String nameNorm = normalize(name);
+                    if (nameNorm.equals(entityNorm) || nameNorm.contains(entityNorm) || entityNorm.contains(nameNorm)) {
+                        matchedCompany = name;
+                        c.put("_matched", true);
+                        break;
+                    }
+                }
+            }
+        }
+        final boolean companyNameMatched = matchedCompany != null;
+
+        kyc.markDinVerified(dinNumber, rawMcaResponse);
         kyc.tryAutoVerify();
         kycRepositoryWrapper.saveAndFlush(kyc);
 
         final Map<String, Object> changes = new java.util.LinkedHashMap<>();
+        changes.put("dinNumber", dinNumber);
         changes.put("dinVerified", true);
         changes.put("dinVerifiedOn", kyc.getDinVerifiedOn());
+        changes.put("isDirector", true);
+        changes.put("entityName", entityName);
+        changes.put("companies", companies);
+        changes.put("companyNameMatched", companyNameMatched);
+        changes.put("matchedCompany", matchedCompany);
         changes.put("kycStatus", kyc.getKycStatusValue());
         return buildResult(command, client, clientId, kycId, changes);
     }
 
+    /**
+     * Extracts the list of companies/LLPs a director is associated with from the
+     * Sandbox/MCA validate-director response. Expected shape:
+     * <pre>
+     * { "code": 200, "data": {
+     *     "director_data": { "din": "...", "name": "..." },
+     *     "company_data": [ { "company_name", "cin/fcrn", "designation",
+     *                         "begin_date", "end_date" }, ... ],
+     *     "llp_data":     [ { "llp_name", "llpin/fllpin", "designation",
+     *                         "begin_date", "end_date" }, ... ]
+     * } }
+     * </pre>
+     * Returns maps with normalized fields: companyName, cin, designation,
+     * appointmentDate, endDate, type (COMPANY|LLP), _matched.
+     */
+    private List<Map<String, Object>> extractDirectorshipCompanies(final String rawResponse) {
+        final java.util.List<Map<String, Object>> companies = new java.util.ArrayList<>();
+        if (StringUtils.isBlank(rawResponse)) {
+            return companies;
+        }
+        try {
+            final JsonElement root = JsonParser.parseString(rawResponse);
+            if (!root.isJsonObject()) {
+                return companies;
+            }
+            JsonObject obj = root.getAsJsonObject();
+            // Navigate to the "data" node (Sandbox wraps the payload under "data")
+            if (obj.has("data") && obj.get("data").isJsonObject()) {
+                obj = obj.getAsJsonObject("data");
+            }
+
+            if (obj.has("company_data") && obj.get("company_data").isJsonArray()) {
+                for (final JsonElement el : obj.getAsJsonArray("company_data")) {
+                    if (!el.isJsonObject()) continue;
+                    final JsonObject c = el.getAsJsonObject();
+                    final String name = firstString(c, "company_name", "companyName", "name");
+                    if (StringUtils.isBlank(name)) continue;
+                    final Map<String, Object> entry = new java.util.LinkedHashMap<>();
+                    entry.put("companyName", name);
+                    entry.put("cin", firstString(c, "cin/fcrn", "cin", "cinNumber"));
+                    entry.put("designation", firstString(c, "designation", "role"));
+                    entry.put("appointmentDate", firstString(c, "begin_date", "appointmentDate", "dateOfAppointment"));
+                    entry.put("endDate", firstString(c, "end_date", "endDate"));
+                    entry.put("type", "COMPANY");
+                    entry.put("_matched", false);
+                    companies.add(entry);
+                }
+            }
+            if (obj.has("llp_data") && obj.get("llp_data").isJsonArray()) {
+                for (final JsonElement el : obj.getAsJsonArray("llp_data")) {
+                    if (!el.isJsonObject()) continue;
+                    final JsonObject c = el.getAsJsonObject();
+                    final String name = firstString(c, "llp_name", "companyName", "name");
+                    if (StringUtils.isBlank(name)) continue;
+                    final Map<String, Object> entry = new java.util.LinkedHashMap<>();
+                    entry.put("companyName", name);
+                    entry.put("cin", firstString(c, "llpin/fllpin", "llpin", "cin"));
+                    entry.put("designation", firstString(c, "designation", "role"));
+                    entry.put("appointmentDate", firstString(c, "begin_date", "appointmentDate"));
+                    entry.put("endDate", firstString(c, "end_date", "endDate"));
+                    entry.put("type", "LLP");
+                    entry.put("_matched", false);
+                    companies.add(entry);
+                }
+            }
+        } catch (final Exception e) {
+            LOG.warn("Unable to parse DIN MCA response for company extraction: {}", e.getMessage());
+        }
+        return companies;
+    }
+
+    private String firstString(final JsonObject o, final String... keys) {
+        for (final String k : keys) {
+            if (o.has(k) && o.get(k).isJsonPrimitive()) {
+                final String v = o.get(k).getAsString();
+                if (StringUtils.isNotBlank(v)) {
+                    return v;
+                }
+            }
+        }
+        return null;
+    }
+
+    private String normalize(final String s) {
+        if (s == null) return "";
+        return s.trim().toLowerCase().replaceAll("[^a-z0-9]", "");
+    }
+
+    /**
+     * CIN, GSTIN and incorporation-date columns on m_client_kyc are only meaningful
+     * for NON_INDIVIDUAL clients. Reject any request that tries to set them on an
+     * INDIVIDUAL KYC record. Applied on both initiate and update.
+     */
+    private void enforceNonIndividualOnlyFields(final JsonCommand command, final boolean isIndividual) {
+        if (!isIndividual) return;
+        final List<ApiParameterError> errors = new java.util.ArrayList<>();
+        final String[] disallowed = new String[] {
+                ClientKycApiConstants.cinNumberParam,
+                ClientKycApiConstants.gstinNumberParam,
+                ClientKycApiConstants.enterprisePanParam,
+                ClientKycApiConstants.incorporationDateParam
+        };
+        for (final String param : disallowed) {
+            if (command.parameterExists(param)) {
+                final String value = command.stringValueOfParameterNamed(param);
+                if (StringUtils.isNotBlank(value)) {
+                    errors.add(ApiParameterError.parameterError(
+                            "error.msg.client.kyc." + param + ".not.applicable.for.individual",
+                            param + " is only applicable for non-individual (business) clients.",
+                            param));
+                }
+            }
+        }
+        if (!errors.isEmpty()) {
+            throw new PlatformApiDataValidationException(errors);
+        }
+    }
+
     @Transactional
     @Override
     public CommandProcessingResult rejectKyc(final Long clientId, final Long kycId, final JsonCommand command) {
@@ -575,6 +737,188 @@ public class ClientKycWritePlatformServiceJpaRepositoryImpl implements ClientKyc
                 Map.of("kycStatus", kyc.getKycStatusValue(), "rejectedOn", kyc.getRejectedOn()));
     }
 
+    @Transactional
+    @Override
+    public CommandProcessingResult completeRekyc(final Long clientId, final Long kycId, final JsonCommand command) {
+        this.context.authenticatedUser();
+
+        final Client client = clientRepository.findOneWithNotFoundDetection(clientId);
+        final ClientKyc kyc = kycRepositoryWrapper.findOneWithNotFoundDetection(kycId);
+
+        if (kyc.getKycStatus().isRejected()) {
+            throw new ClientKycInvalidStateTransitionException("complete-rekyc", kyc.getKycStatus().name());
+        }
+
+        // Re-use markVerified: sets kycStatus=VERIFIED, verifiedOn=today, and
+        // expiryDate = today + reKycYears (Low=10, Medium=8, High=2 per KycRiskCategory).
+        kyc.markVerified(kyc.getCkycNumber());
+        kycRepositoryWrapper.saveAndFlush(kyc);
+
+        final Map<String, Object> changes = new java.util.LinkedHashMap<>();
+        changes.put("verifiedOn", kyc.getVerifiedOn());
+        changes.put("expiryDate", kyc.getExpiryDate());
+        changes.put("kycStatus", kyc.getKycStatusValue());
+        changes.put("kycStatusCode", "kyc.status.verified");
+        return buildResult(command, client, clientId, kycId, changes);
+    }
+
+    @Transactional
+    @Override
+    public CommandProcessingResult uploadCkyc(final Long clientId, final Long kycId, final JsonCommand command) {
+        this.context.authenticatedUser();
+
+        final Client client = clientRepository.findOneWithNotFoundDetection(clientId);
+        final ClientKyc kyc = kycRepositoryWrapper.findOneWithNotFoundDetection(kycId);
+
+        if (kyc.getKycStatus().isRejected()) {
+            throw new ClientKycInvalidStateTransitionException("upload-ckyc", kyc.getKycStatus().name());
+        }
+
+        // Idempotent: if we already have a CKYC number from a previous upload, return it.
+        if (StringUtils.isNotBlank(kyc.getCkycNumber())) {
+            LOG.info("CKYC already uploaded for clientId={}, kycId={}, ckycNumber={}. Returning existing.",
+                    clientId, kycId, kyc.getCkycNumber());
+            final Map<String, Object> changes = new java.util.LinkedHashMap<>();
+            changes.put("ckycNumber", kyc.getCkycNumber());
+            changes.put("ckycBatchId", kyc.getCkycBatchId());
+            changes.put("ckycUploadedOn", kyc.getCkycUploadedOn());
+            changes.put("status", "ALREADY_UPLOADED");
+            return buildResult(command, client, clientId, kycId, changes);
+        }
+
+        final org.apache.fineract.portfolio.client.ckyc.CkycUploadResult result = ckycSftpUploadService.upload(kyc);
+
+        if (!"SUCCESS".equalsIgnoreCase(result.getStatus())) {
+            LOG.warn("CKYC upload failed for clientId={}, kycId={}: {}", clientId, kycId, result.getRemarks());
+            throw new PlatformServiceUnavailableException(
+                    "error.msg.ckyc.upload.failed",
+                    "CKYC upload failed: " + result.getRemarks(),
+                    result.getRemarks());
+        }
+
+        kyc.markCkycUploaded(result.getCkycNumber(), result.getBatchId());
+        kycRepositoryWrapper.saveAndFlush(kyc);
+
+        LOG.info("CKYC uploaded for clientId={}, kycId={}, ckycNumber={}, batchId={}",
+                clientId, kycId, kyc.getCkycNumber(), kyc.getCkycBatchId());
+
+        final Map<String, Object> changes = new java.util.LinkedHashMap<>();
+        changes.put("ckycNumber", kyc.getCkycNumber());
+        changes.put("ckycBatchId", kyc.getCkycBatchId());
+        changes.put("ckycUploadedOn", kyc.getCkycUploadedOn());
+        changes.put("status", "SUCCESS");
+        return buildResult(command, client, clientId, kycId, changes);
+    }
+
+    @Transactional
+    @Override
+    public CommandProcessingResult uploadCkycZip(final Long clientId, final Long kycId, final java.io.InputStream stream,
+            final String fileName, final Long fileSize, final String remarks) {
+        this.context.authenticatedUser();
+
+        final Client client = clientRepository.findOneWithNotFoundDetection(clientId);
+        final ClientKyc kyc = kycRepositoryWrapper.findOneWithNotFoundDetection(kycId);
+
+        if (kyc.getKycStatus().isRejected()) {
+            throw new ClientKycInvalidStateTransitionException("upload-ckyc-zip", kyc.getKycStatus().name());
+        }
+
+        if (stream == null) {
+            throw new PlatformApiDataValidationException(
+                    "error.msg.ckyc.zip.missing",
+                    "CKYC ZIP file is required",
+                    java.util.List.of(ApiParameterError.parameterError(
+                            "error.msg.ckyc.zip.missing", "file must be supplied as a multipart form part", "file")));
+        }
+
+        if (fileName == null || !fileName.toLowerCase().endsWith(".zip")) {
+            throw new PlatformApiDataValidationException(
+                    "error.msg.ckyc.zip.invalid-extension",
+                    "CKYC upload must be a .zip file",
+                    java.util.List.of(ApiParameterError.parameterError(
+                            "error.msg.ckyc.zip.invalid-extension", "file name must end with .zip", "file", fileName)));
+        }
+
+        // Drain the upload stream into memory (bounded to CKYC_ZIP_MAX_BYTES).
+        final byte[] zipBytes;
+        try (java.io.InputStream in = stream) {
+            final java.io.ByteArrayOutputStream buf = new java.io.ByteArrayOutputStream(
+                    fileSize != null && fileSize > 0 ? fileSize.intValue() : 8192);
+            final byte[] chunk = new byte[8192];
+            int read;
+            int total = 0;
+            while ((read = in.read(chunk)) != -1) {
+                total += read;
+                if (total > CKYC_ZIP_MAX_BYTES) {
+                    throw new PlatformApiDataValidationException(
+                            "error.msg.ckyc.zip.too-large",
+                            "CKYC ZIP exceeds maximum allowed size of " + CKYC_ZIP_MAX_BYTES + " bytes",
+                            java.util.List.of(ApiParameterError.parameterError(
+                                    "error.msg.ckyc.zip.too-large", "file too large", "file", CKYC_ZIP_MAX_BYTES)));
+                }
+                buf.write(chunk, 0, read);
+            }
+            zipBytes = buf.toByteArray();
+        } catch (final java.io.IOException ioe) {
+            throw new PlatformServiceUnavailableException(
+                    "error.msg.ckyc.zip.read-failed",
+                    "Could not read CKYC ZIP upload stream: " + ioe.getMessage(),
+                    ioe.getMessage());
+        }
+
+        // 1) Persist the ZIP via the document-management subsystem — one m_document row per upload.
+        //    parent_entity_type = 'client_kyc' so it's visible alongside the KYC record.
+        final org.apache.fineract.infrastructure.documentmanagement.data.DocumentCreateRequest docRequest =
+                org.apache.fineract.infrastructure.documentmanagement.data.DocumentCreateRequest.builder()
+                        .entityType("client_kyc")
+                        .entityId(kycId)
+                        .name(StringUtils.defaultIfEmpty(remarks, "CKYC CERSAI batch upload"))
+                        .fileName(fileName)
+                        .description(remarks)
+                        .size((long) zipBytes.length)
+                        .type("application/zip")
+                        .stream(new java.io.ByteArrayInputStream(zipBytes))
+                        .build();
+
+        final var docResponse = documentWritePlatformService.createDocument(docRequest);
+        LOG.info("CKYC ZIP persisted for clientId={}, kycId={}, documentId={}, fileName={}",
+                clientId, kycId, docResponse.getResourceId(), fileName);
+
+        // 2) Hand the bytes to the upload service (stub now, real CERSAI SFTP later).
+        final org.apache.fineract.portfolio.client.ckyc.CkycUploadResult result =
+                ckycSftpUploadService.uploadZip(kyc, zipBytes, fileName);
+
+        if (!"SUCCESS".equalsIgnoreCase(result.getStatus())) {
+            LOG.warn("CKYC ZIP upload failed for clientId={}, kycId={}: {}", clientId, kycId, result.getRemarks());
+            throw new PlatformServiceUnavailableException(
+                    "error.msg.ckyc.upload.failed",
+                    "CKYC upload failed: " + result.getRemarks(),
+                    result.getRemarks());
+        }
+
+        // 3) Persist the CKYC number returned by CERSAI (or the stub).
+        kyc.markCkycUploaded(result.getCkycNumber(), result.getBatchId());
+        kycRepositoryWrapper.saveAndFlush(kyc);
+
+        LOG.info("CKYC uploaded (ZIP) for clientId={}, kycId={}, ckycNumber={}, batchId={}, documentId={}",
+                clientId, kycId, kyc.getCkycNumber(), kyc.getCkycBatchId(), docResponse.getResourceId());
+
+        final Map<String, Object> changes = new java.util.LinkedHashMap<>();
+        changes.put("ckycNumber", kyc.getCkycNumber());
+        changes.put("ckycBatchId", kyc.getCkycBatchId());
+        changes.put("ckycUploadedOn", kyc.getCkycUploadedOn());
+        changes.put("documentId", docResponse.getResourceId());
+        changes.put("fileName", fileName);
+        changes.put("fileSize", zipBytes.length);
+        changes.put("status", "SUCCESS");
+        return new CommandProcessingResultBuilder()
+                .withOfficeId(client.officeId())
+                .withClientId(clientId)
+                .withEntityId(kycId)
+                .with(changes)
+                .build();
+    }
+
     @Override
     public void initiateKycInline(final Client client, final String kycJson) {
         final ClientKycCommand kycCommand = deserializer.commandFromApiJson(

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/service/NonCkycClientReadPlatformService.java

@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.service;
+
+import java.time.LocalDate;
+import java.util.List;
+import org.apache.fineract.portfolio.client.data.NonCkycClientData;
+
+public interface NonCkycClientReadPlatformService {
+
+    /**
+     * List active clients whose {@code m_client_kyc.ckyc_number} is blank, for the
+     * Mark Non-CKYC screen at {@code /ckyc/record-master}.
+     *
+     * @param officeId         optional branch / office filter
+     * @param productId        optional loan product filter (joins through {@code m_loan})
+     * @param fromDate         optional lower bound on client activation date
+     * @param toDate           optional upper bound on client activation date
+     * @param missingArtifact  optional artefact filter: {@code photo}, {@code signature},
+     *                         {@code id}, {@code address}, or {@code any}. When provided,
+     *                         only clients missing the specified artefact(s) are returned.
+     * @param limit            optional page size
+     * @param offset           optional page offset
+     */
+    List<NonCkycClientData> retrieveNonCkycClients(Long officeId, Long productId, LocalDate fromDate, LocalDate toDate,
+            String missingArtifact, Integer limit, Integer offset);
+}

fineract-provider/src/main/java/org/apache/fineract/portfolio/client/service/NonCkycClientReadPlatformServiceImpl.java

@@ -0,0 +1,175 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.portfolio.client.service;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.infrastructure.core.domain.JdbcSupport;
+import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
+import org.apache.fineract.portfolio.client.data.NonCkycClientData;
+import org.apache.fineract.portfolio.client.domain.ClientStatus;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Service
+@RequiredArgsConstructor
+@Transactional(readOnly = true)
+public class NonCkycClientReadPlatformServiceImpl implements NonCkycClientReadPlatformService {
+
+    private static final int ACTIVE_CLIENT_STATUS = 300;
+
+    private final JdbcTemplate jdbcTemplate;
+    private final PlatformSecurityContext context;
+
+    @Override
+    public List<NonCkycClientData> retrieveNonCkycClients(final Long officeId, final Long productId, final LocalDate fromDate,
+            final LocalDate toDate, final String missingArtifact, final Integer limit, final Integer offset) {
+
+        context.authenticatedUser();
+
+        final NonCkycRowMapper rm = new NonCkycRowMapper();
+        final StringBuilder sql = new StringBuilder(2048);
+        final List<Object> args = new ArrayList<>();
+        final List<Integer> types = new ArrayList<>();
+
+        sql.append("select ").append(rm.schema());
+        sql.append(" where c.status_enum = ? ");
+        args.add(ACTIVE_CLIENT_STATUS);
+        types.add(Types.INTEGER);
+
+        // Active client missing CKYC: KYC row absent, or ckyc_number null / blank
+        sql.append(" and (kyc.id is null or kyc.ckyc_number is null or trim(kyc.ckyc_number) = '') ");
+
+        if (officeId != null) {
+            sql.append(" and c.office_id = ? ");
+            args.add(officeId);
+            types.add(Types.BIGINT);
+        }
+
+        if (productId != null) {
+            sql.append(" and exists (select 1 from m_loan l where l.client_id = c.id and l.product_id = ?) ");
+            args.add(productId);
+            types.add(Types.BIGINT);
+        }
+
+        if (fromDate != null) {
+            sql.append(" and c.activation_date >= ? ");
+            args.add(java.sql.Date.valueOf(fromDate));
+            types.add(Types.DATE);
+        }
+
+        if (toDate != null) {
+            sql.append(" and c.activation_date <= ? ");
+            args.add(java.sql.Date.valueOf(toDate));
+            types.add(Types.DATE);
+        }
+
+        if (missingArtifact != null && !missingArtifact.isBlank()) {
+            final String art = missingArtifact.trim().toLowerCase();
+            switch (art) {
+                case "photo" -> sql.append(" and c.image_id is null ");
+                case "signature" -> sql.append(
+                        " and not exists (select 1 from m_document d where d.parent_entity_type = 'clients' "
+                                + " and d.parent_entity_id = c.id and d.name = 'clientSignature') ");
+                case "id" -> sql.append(" and not exists (select 1 from m_client_identifier ci "
+                        + " where ci.client_id = c.id and (ci.status = 200 or ci.status is null)) ");
+                case "address" -> sql.append(" and not exists (select 1 from m_address a where a.client_id = c.id) ");
+                case "any" -> sql.append(" and ( c.image_id is null "
+                        + " or not exists (select 1 from m_document d where d.parent_entity_type = 'clients' "
+                        + "     and d.parent_entity_id = c.id and d.name = 'clientSignature') "
+                        + " or not exists (select 1 from m_client_identifier ci where ci.client_id = c.id and (ci.status = 200 or ci.status is null)) "
+                        + " or not exists (select 1 from m_address a where a.client_id = c.id) ) ");
+                default -> { /* unknown value → ignore */ }
+            }
+        }
+
+        sql.append(" order by c.id desc ");
+
+        if (limit != null && limit > 0) {
+            sql.append(" limit ? ");
+            args.add(limit);
+            types.add(Types.INTEGER);
+        }
+        if (offset != null && offset > 0) {
+            sql.append(" offset ? ");
+            args.add(offset);
+            types.add(Types.INTEGER);
+        }
+
+        final int[] typeArr = types.stream().mapToInt(Integer::intValue).toArray();
+        return jdbcTemplate.query(sql.toString(), args.toArray(), typeArr, rm);
+    }
+
+    private static final class NonCkycRowMapper implements RowMapper<NonCkycClientData> {
+
+        /**
+         * Address is flattened to a single comma-separated string from the most recent m_address row
+         * (primary address — one per client since 0245). Uses a correlated scalar sub-query so the
+         * row count of the outer query stays equal to the number of clients.
+         */
+        public String schema() {
+            return " c.id as clientId, c.account_no as accountNo, c.display_name as fullName, "
+                    + " c.email_address as email, c.mobile_no as mobile, c.status_enum as statusEnum, "
+                    + " c.image_id as imageId, "
+                    + " (select "
+                    + "    concat_ws(', ', nullif(a.address_line_1,''), nullif(a.address_line_2,''), "
+                    + "              nullif(a.address_line_3,''), nullif(a.city,''), nullif(a.postal_code,'')) "
+                    + "  from m_address a where a.client_id = c.id "
+                    + "  order by a.id desc limit 1) as address, "
+                    + " (select count(1) from m_client_identifier ci where ci.client_id = c.id and (ci.status = 200 or ci.status is null)) as idProofCount, "
+                    + " (select count(1) from m_address a where a.client_id = c.id) as addressCount, "
+                    + " (select count(1) from m_document d where d.parent_entity_type = 'clients' "
+                    + "    and d.parent_entity_id = c.id and d.name = 'clientSignature') as signatureCount "
+                    + " from m_client c "
+                    + " left join m_client_kyc kyc on kyc.client_id = c.id ";
+        }
+
+        @Override
+        public NonCkycClientData mapRow(final ResultSet rs, final int rowNum) throws SQLException {
+            final Long imageId = JdbcSupport.getLong(rs, "imageId");
+            final int idProofCount = rs.getInt("idProofCount");
+            final int addressCount = rs.getInt("addressCount");
+            final int signatureCount = rs.getInt("signatureCount");
+            final Integer statusEnum = JdbcSupport.getInteger(rs, "statusEnum");
+            final String statusCode = statusEnum == null ? null : ClientStatus.fromInt(statusEnum).getCode();
+
+            return NonCkycClientData.builder()
+                    .clientId(rs.getLong("clientId"))
+                    .accountNo(rs.getString("accountNo"))
+                    .fullName(rs.getString("fullName"))
+                    .email(rs.getString("email"))
+                    .mobile(rs.getString("mobile"))
+                    .address(rs.getString("address"))
+                    .status(statusEnum)
+                    .statusCode(statusCode)
+                    .hasPhoto(imageId != null)
+                    .hasSignature(signatureCount > 0)
+                    .hasIdProof(idProofCount > 0)
+                    .hasAddressProof(addressCount > 0)
+                    .build();
+        }
+    }
+}

fineract-provider/src/main/resources/application.properties

@@ -179,9 +179,9 @@ fineract.loan.transactionprocessor.error-not-found-fail=${FINERACT_LOAN_TRANSACT
 fineract.loan.status-change-history-statuses=${FINERACT_LOAN_STATUS_CHANGE_HISTORY_STATUSES:NONE}
 
 fineract.content.regex-whitelist-enabled=${FINERACT_CONTENT_REGEX_WHITELIST_ENABLED:true}
-fineract.content.regex-whitelist=${FINERACT_CONTENT_REGEX_WHITELIST:.*\\.pdf$,.*\\.doc,.*\\.docx,.*\\.xls,.*\\.xlsx,.*\\.jpg,.*\\.jpeg,.*\\.png}
+fineract.content.regex-whitelist=${FINERACT_CONTENT_REGEX_WHITELIST:.*\\.pdf$,.*\\.doc,.*\\.docx,.*\\.xls,.*\\.xlsx,.*\\.jpg,.*\\.jpeg,.*\\.png,.*\\.zip$}
 fineract.content.mime-whitelist-enabled=${FINERACT_CONTENT_MIME_WHITELIST_ENABLED:true}
-fineract.content.mime-whitelist=${FINERACT_CONTENT_MIME_WHITELIST:application/pdf,application/msword,application/vnd.openxmlformats-officedocument.wordprocessingml.document,application/vnd.ms-excel,application/vnd.openxmlformats-officedocument.spreadsheetml.sheet,image/jpeg,image/png}
+fineract.content.mime-whitelist=${FINERACT_CONTENT_MIME_WHITELIST:application/pdf,application/msword,application/vnd.openxmlformats-officedocument.wordprocessingml.document,application/vnd.ms-excel,application/vnd.openxmlformats-officedocument.spreadsheetml.sheet,image/jpeg,image/png,application/zip,application/x-zip-compressed,application/octet-stream}
 fineract.content.default-buffer-size=${FINERACT_CONTENT_DEFAULT_BUFFER_SIZE:8192}
 fineract.content.filesystem.enabled=${FINERACT_CONTENT_FILESYSTEM_ENABLED:true}
 fineract.content.filesystem.rootFolder=${FINERACT_CONTENT_FILESYSTEM_ROOT_FOLDER:${user.home}/.fineract}

fineract-provider/src/main/resources/db/changelog/tenant/changelog-tenant.xml

@@ -268,4 +268,10 @@
     <include file="parts/0247_add_huf_client_classification.xml" relativeToChangelogFile="true"/>
     <include file="parts/0248_add_cin_din_verification_fields.xml" relativeToChangelogFile="true"/>
     <include file="parts/0249_add_non_person_extended_fields.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0250_add_kyc_director_fields.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0251_add_kyc_proof_type_codes.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0252_add_ckyc_uploaded_on.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0253_add_ckyc_batch_table.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0254_fix_ckyc_batch_audit_columns.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0255_add_ckyc_batch_id_to_client_kyc.xml" relativeToChangelogFile="true"/>
 </databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0250_add_kyc_director_fields.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements. See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership. The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License. You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing,
+  software distributed under the License is distributed on an
+  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  KIND, either express or implied. See the License for the
+  specific language governing permissions and limitations
+  under the License.
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.13.xsd">
+
+    <changeSet author="fineract" id="0250-1" context="default">
+        <addColumn tableName="m_client_kyc">
+            <column name="is_director" type="boolean" defaultValueBoolean="false">
+                <constraints nullable="false"/>
+            </column>
+            <column name="din_mca_response" type="longtext"/>
+        </addColumn>
+    </changeSet>
+
+</databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0251_add_kyc_proof_type_codes.xml

@@ -0,0 +1,96 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.13.xsd">
+
+    <!-- ==================== KYCIdProofType code ==================== -->
+    <changeSet author="fineract" id="0251-1">
+        <preConditions onFail="MARK_RAN">
+            <sqlCheck expectedResult="0">SELECT COUNT(*) FROM m_code WHERE code_name = 'KYCIdProofType'</sqlCheck>
+        </preConditions>
+        <insert tableName="m_code">
+            <column name="code_name" value="KYCIdProofType"/>
+            <column name="is_system_defined" valueBoolean="true"/>
+        </insert>
+    </changeSet>
+
+    <changeSet author="fineract" id="0251-2">
+        <preConditions onFail="MARK_RAN">
+            <sqlCheck expectedResult="0">SELECT COUNT(*) FROM m_code_value WHERE code_id = (SELECT id FROM m_code WHERE code_name = 'KYCIdProofType')</sqlCheck>
+        </preConditions>
+        <sql>
+            INSERT INTO m_code_value (code_id, code_value, code_description, order_position, is_active, is_mandatory)
+            SELECT id, val, descr, pos, true, false FROM m_code
+            CROSS JOIN (VALUES
+                ('PAN Card', 'PAN', 1),
+                ('Aadhaar Card', 'AADHAAR', 2),
+                ('Passport', 'PASSPORT', 3),
+                ('Voter ID Card (EPIC)', 'VOTER_ID', 4),
+                ('Driving License', 'DRIVING_LICENSE', 5),
+                ('NREGA Job Card (signed by government authority)', 'NREGA_JOB_CARD', 6),
+                ('Letter issued by National Population Register (NPR)', 'NPR_LETTER', 7)
+            ) AS t(val, descr, pos)
+            WHERE code_name = 'KYCIdProofType';
+        </sql>
+    </changeSet>
+
+    <!-- ==================== KYCAddressProofType code ==================== -->
+    <changeSet author="fineract" id="0251-3">
+        <preConditions onFail="MARK_RAN">
+            <sqlCheck expectedResult="0">SELECT COUNT(*) FROM m_code WHERE code_name = 'KYCAddressProofType'</sqlCheck>
+        </preConditions>
+        <insert tableName="m_code">
+            <column name="code_name" value="KYCAddressProofType"/>
+            <column name="is_system_defined" valueBoolean="true"/>
+        </insert>
+    </changeSet>
+
+    <changeSet author="fineract" id="0251-4">
+        <preConditions onFail="MARK_RAN">
+            <sqlCheck expectedResult="0">SELECT COUNT(*) FROM m_code_value WHERE code_id = (SELECT id FROM m_code WHERE code_name = 'KYCAddressProofType')</sqlCheck>
+        </preConditions>
+        <sql>
+            INSERT INTO m_code_value (code_id, code_value, code_description, order_position, is_active, is_mandatory)
+            SELECT id, val, descr, pos, true, false FROM m_code
+            CROSS JOIN (VALUES
+                ('Aadhaar Card', 'AADHAAR', 1),
+                ('Passport', 'PASSPORT', 2),
+                ('Driving License', 'DRIVING_LICENSE', 3),
+                ('Voter ID Card', 'VOTER_ID', 4),
+                ('Utility Bills', 'UTILITY_BILL', 5),
+                ('Electricity Bill', 'ELECTRICITY_BILL', 6),
+                ('Water Bill', 'WATER_BILL', 7),
+                ('Gas Bill', 'GAS_BILL', 8),
+                ('Internet Bill', 'INTERNET_BILL', 9),
+                ('Bank Account Statement / Passbook', 'BANK_STATEMENT', 10),
+                ('Postpaid Mobile Bill', 'POSTPAID_MOBILE_BILL', 11),
+                ('Registered Rent Agreement', 'RENT_AGREEMENT', 12),
+                ('Property Tax Receipt', 'PROPERTY_TAX_RECEIPT', 13),
+                ('Employer Letter (for salaried employees)', 'EMPLOYER_LETTER', 14),
+                ('Pension Payment Orders (for retirees)', 'PENSION_PAYMENT_ORDER', 15)
+            ) AS t(val, descr, pos)
+            WHERE code_name = 'KYCAddressProofType';
+        </sql>
+    </changeSet>
+
+</databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0252_add_ckyc_uploaded_on.xml

@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.13.xsd">
+
+    <!-- Track when the CKYC number was pushed to CERSAI (via SFTP in prod, dummy stub for now) -->
+    <changeSet author="fineract" id="0252-1">
+        <preConditions onFail="MARK_RAN">
+            <not><columnExists tableName="m_client_kyc" columnName="ckyc_uploaded_on"/></not>
+        </preConditions>
+        <addColumn tableName="m_client_kyc">
+            <column name="ckyc_uploaded_on" type="DATE"/>
+        </addColumn>
+    </changeSet>
+
+    <!-- Optional batch id from the CERSAI SFTP response, useful for reconciliation -->
+    <changeSet author="fineract" id="0252-2">
+        <preConditions onFail="MARK_RAN">
+            <not><columnExists tableName="m_client_kyc" columnName="ckyc_batch_id"/></not>
+        </preConditions>
+        <addColumn tableName="m_client_kyc">
+            <column name="ckyc_batch_id" type="VARCHAR(50)"/>
+        </addColumn>
+    </changeSet>
+
+</databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0253_add_ckyc_batch_table.xml

@@ -0,0 +1,84 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.13.xsd">
+
+    <!--
+      CERSAI CKYC batch ZIP registry.
+      One row per ZIP assembled by the bank for CERSAI submission.
+      The ZIP itself lives in m_document (parent_entity_type='ckyc_batch').
+    -->
+    <changeSet author="fineract" id="0253-1">
+        <preConditions onFail="MARK_RAN">
+            <not><tableExists tableName="m_ckyc_batch"/></not>
+        </preConditions>
+        <createTable tableName="m_ckyc_batch">
+            <column autoIncrement="true" name="id" type="BIGINT">
+                <constraints nullable="false" primaryKey="true"/>
+            </column>
+            <column name="batch_id" type="VARCHAR(50)">
+                <constraints nullable="false" unique="true"/>
+            </column>
+            <column name="filename" type="VARCHAR(250)">
+                <constraints nullable="false"/>
+            </column>
+            <column name="checksum_sha256" type="VARCHAR(80)"/>
+            <column name="record_count" type="INT" defaultValueNumeric="0">
+                <constraints nullable="false"/>
+            </column>
+            <column name="size_bytes" type="BIGINT" defaultValueNumeric="0">
+                <constraints nullable="false"/>
+            </column>
+            <column name="document_id" type="INT">
+                <constraints nullable="true"/>
+            </column>
+            <column name="created_on" type="DATE">
+                <constraints nullable="false"/>
+            </column>
+            <column name="created_on_utc" type="TIMESTAMP WITH TIME ZONE">
+                <constraints nullable="false"/>
+            </column>
+            <column name="created_by_id" type="BIGINT"/>
+            <column name="uploaded_to_cersai" type="BOOLEAN" defaultValueBoolean="false">
+                <constraints nullable="false"/>
+            </column>
+            <column name="uploaded_on" type="DATE"/>
+            <column name="upload_status" type="VARCHAR(20)"/>
+            <column name="upload_remarks" type="VARCHAR(500)"/>
+            <column name="manifest_json" type="CLOB"/>
+        </createTable>
+    </changeSet>
+
+    <changeSet author="fineract" id="0253-2">
+        <addForeignKeyConstraint baseTableName="m_ckyc_batch" baseColumnNames="document_id"
+                                 referencedTableName="m_document" referencedColumnNames="id"
+                                 constraintName="fk_m_ckyc_batch_document"/>
+    </changeSet>
+
+    <changeSet author="fineract" id="0253-3">
+        <createIndex tableName="m_ckyc_batch" indexName="idx_m_ckyc_batch_created_on">
+            <column name="created_on"/>
+        </createIndex>
+    </changeSet>
+
+</databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0254_fix_ckyc_batch_audit_columns.xml

@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.13.xsd">
+
+    <!--
+      0253 declared created_on_utc and created_by_id columns as if m_ckyc_batch would use the
+      Fineract auditable superclass. The entity instead extends AbstractPersistableCustom (no audit
+      columns), so these columns are unused. created_on_utc was NOT NULL which breaks inserts.
+      Drop both.
+    -->
+    <changeSet author="fineract" id="0254-1">
+        <preConditions onFail="MARK_RAN">
+            <columnExists tableName="m_ckyc_batch" columnName="created_on_utc"/>
+        </preConditions>
+        <dropColumn tableName="m_ckyc_batch" columnName="created_on_utc"/>
+    </changeSet>
+
+    <changeSet author="fineract" id="0254-2">
+        <preConditions onFail="MARK_RAN">
+            <columnExists tableName="m_ckyc_batch" columnName="created_by_id"/>
+        </preConditions>
+        <dropColumn tableName="m_ckyc_batch" columnName="created_by_id"/>
+    </changeSet>
+
+</databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0255_add_ckyc_batch_id_to_client_kyc.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.13.xsd">
+
+    <!--
+      Link m_client_kyc rows to the CKYC batch they were shipped in. Populated by the dummy /
+      real CERSAI SFTP upload service so the Mark Non-CKYC dashboard can hide clients that have
+      already been bundled + uploaded.
+    -->
+    <changeSet author="fineract" id="0255-1">
+        <preConditions onFail="MARK_RAN">
+            <not><columnExists tableName="m_client_kyc" columnName="ckyc_batch_id"/></not>
+        </preConditions>
+        <addColumn tableName="m_client_kyc">
+            <column name="ckyc_batch_id" type="VARCHAR(50)"/>
+        </addColumn>
+    </changeSet>
+
+</databaseChangeLog>