Feature/add user data cy

apps/database-abstractor/src/main/java/com/akto/action/DbAction.java

@@ -43,6 +43,7 @@
 import com.akto.dto.graph.SvcToSvcGraphNode;
 import com.akto.dto.metrics.MetricData;
 import com.akto.dto.monitoring.ModuleInfo;
+import com.akto.dto.agentic_sessions.AgentQueryData;
 import com.akto.dto.agentic_sessions.SessionDocument;
 import com.akto.dto.notifications.SlackWebhook;
 import com.akto.dto.runtime_filters.RuntimeFilter;
@@ -190,6 +191,9 @@ public class DbAction extends ActionSupport {
     @Getter @Setter
     private List<SessionDocument> sessionDocuments;
 
+    @Getter @Setter
+    private AgentQueryData agentQueryData;
+
     private static final LoggerMaker loggerMaker = new LoggerMaker(DbAction.class, LogDb.DB_ABS);
 
     public List<BulkUpdates> getWritesForTestingRunIssues() {
@@ -3635,6 +3639,16 @@ public String storeSpans() {
         return Action.SUCCESS.toUpperCase();
     }
 
+    public String storeAgentQueryData() {
+        try {
+            DbLayer.storeAgentQueryData(agentQueryData);
+        } catch (Exception e) {
+            loggerMaker.errorAndAddToDb(e, "Error in storeAgentQueryData " + e.toString());
+            return Action.ERROR.toUpperCase();
+        }
+        return Action.SUCCESS.toUpperCase();
+    }
+
     private TestingRunWebhook testingRunWebhook;
 
     public TestingRunWebhook getTestingRunWebhook() {

apps/database-abstractor/src/main/resources/struts.xml

@@ -2378,6 +2378,16 @@
                 <param name="includeProperties">^actionErrors.*</param>
             </result>
         </action>
+        <action name="api/storeAgentQueryData" class="com.akto.action.DbAction" method="storeAgentQueryData">
+            <interceptor-ref name="json"/>
+            <interceptor-ref name="defaultStack" />
+            <result name="SUCCESS" type="json"/>
+            <result name="ERROR" type="json">
+                <param name="statusCode">422</param>
+                <param name="ignoreHierarchy">false</param>
+                <param name="includeProperties">^actionErrors.*</param>
+            </result>
+        </action>
 
     </package>
 

libs/dao/src/main/java/com/akto/dao/agentic_sessions/AgentQueryDataDao.java

@@ -0,0 +1,71 @@
+package com.akto.dao.agentic_sessions;
+
+import com.akto.dao.AccountsContextDao;
+import com.akto.dao.MCollection;
+import com.akto.dao.context.Context;
+import com.akto.dto.HttpResponseParams;
+import com.akto.dto.agentic_sessions.AgentQueryData;
+import com.mongodb.client.model.IndexOptions;
+import com.mongodb.client.model.Indexes;
+import org.bson.conversions.Bson;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class AgentQueryDataDao extends AccountsContextDao<AgentQueryData> {
+
+    public static final AgentQueryDataDao instance = new AgentQueryDataDao();
+
+    private static final long TTL_DAYS = 14;
+
+    public void createIndicesIfAbsent() {
+        // TTL index: auto-delete records older than 14 days
+        Bson ttlIndex = Indexes.ascending(AgentQueryData.TIME_STAMP);
+        IndexOptions ttlOptions = new IndexOptions()
+                .name("timeStamp_ttl")
+                .expireAfter(TTL_DAYS * 24 * 60 * 60, TimeUnit.SECONDS);
+        MCollection.createIndexIfAbsent(getDBName(), getCollName(), ttlIndex, ttlOptions);
+
+        // Compound index for cron queries: fetch by user within a time window
+        String[] compoundFields = new String[]{
+                AgentQueryData.SERVICE_ID,
+                AgentQueryData.UNIQUE_USER_ID,
+                AgentQueryData.TIME_STAMP
+        };
+        MCollection.createIndexIfAbsent(getDBName(), getCollName(), compoundFields, false);
+    }
+
+    public AgentQueryData createAgentQueryDataFromHttpResponseParams(HttpResponseParams httpResponseParam) {
+        AgentQueryData agentQueryData = new AgentQueryData();
+
+        Map<String, List<String>> requestHeaders = httpResponseParam.getRequestParams().getHeaders();
+
+        agentQueryData.setServiceId(getFirstHeader(requestHeaders, "host"));
+        agentQueryData.setUserName(getFirstHeader(requestHeaders, (AgentQueryData.HEADER_PREFIX + AgentQueryData.HEADER_USER_EMAIL)));
+        agentQueryData.setSessionIdentifier(getFirstHeader(requestHeaders, (AgentQueryData.HEADER_PREFIX + AgentQueryData.HEADER_SESSION_ID)));
+        agentQueryData.setConversationId(getFirstHeader(requestHeaders, (AgentQueryData.HEADER_PREFIX + AgentQueryData.HEADER_CONVERSATION_ID)));
+
+        agentQueryData.setQueryPayload(httpResponseParam.getRequestParams().getPayload());
+        agentQueryData.setResponsePayload(httpResponseParam.getPayload());
+        agentQueryData.setTimeStamp(Context.now());
+
+        return agentQueryData;
+    }
+
+    private static String getFirstHeader(Map<String, List<String>> headers, String name) {
+        if (headers == null) return null;
+        List<String> values = headers.get(name);
+        return (values != null && !values.isEmpty()) ? values.get(0) : null;
+    }
+
+    @Override
+    public String getCollName() {
+        return "agent_query_data";
+    }
+
+    @Override
+    public Class<AgentQueryData> getClassT() {
+        return AgentQueryData.class;
+    }
+}

libs/dao/src/main/java/com/akto/dto/agentic_sessions/AgentQueryData.java

@@ -0,0 +1,42 @@
+package com.akto.dto.agentic_sessions;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@AllArgsConstructor
+@NoArgsConstructor
+@Getter
+@Setter
+
+// use this in future for learning the user type, build the flow graph of the agent
+public class AgentQueryData {
+
+    // DB field name constants
+    public static final String SERVICE_ID = "serviceId";
+    public static final String UNIQUE_USER_ID = "uniqueUserId";
+    public static final String SESSION_IDENTIFIER = "sessionIdentifier";
+    public static final String QUERY_PAYLOAD = "queryPayload";
+    public static final String TIME_STAMP = "timeStamp";
+    public static final String INPUT_TOKENS = "inputTokens";
+    public static final String OUTPUT_TOKENS = "outputTokens";
+
+    // Request header name constants
+    public static final String HEADER_PREFIX    = "x-akto-installer-";
+    public static final String HEADER_USER_EMAIL   = "user_email";
+    public static final String HEADER_SESSION_ID   = "session_id";
+    public static final String HEADER_CONVERSATION_ID = "conversation_id";
+    public static final String HEADER_GENERATION_ID   = "generation_id";
+
+    private String serviceId;
+    private String deviceId;
+    private String userName;
+    private String sessionIdentifier;
+    private String conversationId;
+    private String queryPayload;
+    private String responsePayload;
+    private long timeStamp;
+    private int inputTokens;
+    private int outputTokens;
+} 

libs/utils/src/main/java/com/akto/data_actor/ClientActor.java

@@ -15,6 +15,7 @@
 import com.akto.database_abstractor_authenticator.JwtAuthenticator;
 import com.akto.dto.*;
 import com.akto.dto.ApiInfo.ApiInfoKey;
+import com.akto.dto.agentic_sessions.AgentQueryData;
 import com.akto.dto.billing.Organization;
 import com.akto.dto.billing.Tokens;
 import com.akto.dto.bulk_updates.BulkUpdates;
@@ -4460,4 +4461,18 @@ public void storeTestingRunWebhook(TestingRunWebhook testingRunWebhook) {
             return;
         }
     }
+
+    @Override
+    public void storeAgentQueryData(AgentQueryData agentQueryData) {
+        Map<String, List<String>> headers = buildHeaders();
+        BasicDBObject obj = new BasicDBObject();
+        obj.put("agentQueryData", agentQueryData);
+        OriginalHttpRequest request = new OriginalHttpRequest(url + "/storeAgentQueryData", "", "POST", obj.toString(), headers, "");
+        try {
+            OriginalHttpResponse response = ApiExecutor.sendRequest(request, true, null, false, null);
+        } catch (Exception e) {
+            loggerMaker.errorAndAddToDb("error in storeAgentQueryData" + e, LoggerMaker.LogDb.RUNTIME);
+            return;
+        }
+    }   
 }

libs/utils/src/main/java/com/akto/data_actor/DataActor.java

@@ -1,6 +1,7 @@
 package com.akto.data_actor;
 
 import com.akto.dto.*;
+import com.akto.dto.agentic_sessions.AgentQueryData;
 import com.akto.dto.billing.Organization;
 import com.akto.dto.billing.Tokens;
 import com.akto.dto.dependency_flow.Node;
@@ -402,4 +403,6 @@ public void updateNewNodesInBatches(List<SvcToSvcGraphNode> updateNodes) {
     public abstract void storeSpans(List<Span> spans);
 
     public abstract void storeTestingRunWebhook(TestingRunWebhook testingRunWebhook);
+
+    public abstract void storeAgentQueryData(AgentQueryData agentQueryData);
 }

libs/utils/src/main/java/com/akto/data_actor/DbActor.java

@@ -2,6 +2,7 @@
 
 import com.akto.dto.*;
 import com.akto.dto.ApiInfo.ApiInfoKey;
+import com.akto.dto.agentic_sessions.AgentQueryData;
 import com.akto.dto.billing.Organization;
 import com.akto.dto.billing.Tokens;
 import com.akto.dto.dependency_flow.Node;
@@ -744,4 +745,9 @@ public void storeSpans(List<Span> spans) {
     public void storeTestingRunWebhook(TestingRunWebhook testingRunWebhook) {
         DbLayer.storeTestingRunWebhook(testingRunWebhook);
     }
+
+    @Override
+    public void storeAgentQueryData(AgentQueryData agentQueryData) {
+        DbLayer.storeAgentQueryData(agentQueryData);
+    }
 }

libs/utils/src/main/java/com/akto/data_actor/DbLayer.java

@@ -21,6 +21,7 @@
 
 import com.akto.bulk_update_util.ApiInfoBulkUpdate;
 import com.akto.dao.*;
+import com.akto.dao.agentic_sessions.AgentQueryDataDao;
 import com.akto.dao.filter.MergedUrlsDao;
 import com.akto.dao.graph.SvcToSvcGraphEdgesDao;
 import com.akto.dao.graph.SvcToSvcGraphNodesDao;
@@ -76,6 +77,7 @@
 import com.akto.dao.traffic_metrics.TrafficMetricsDao;
 import com.akto.dao.upload.FileUploadsDao;
 import com.akto.dto.ApiInfo.ApiInfoKey;
+import com.akto.dto.agentic_sessions.AgentQueryData;
 import com.akto.dto.billing.Organization;
 import com.akto.dto.billing.Tokens;
 import com.akto.dto.dependency_flow.Node;
@@ -2762,4 +2764,8 @@ public static void bulkUpsertAgenticSessionContext(List<SessionDocument> session
             SessionDocumentDao.instance.getMCollection().bulkWrite(bulkUpdates);
         }
     }
-}
+
+    public static void storeAgentQueryData(AgentQueryData agentQueryData) {
+        AgentQueryDataDao.instance.insertOne(agentQueryData);
+    }
+}