feat: Correlation CRUD operations (#417)

parseablehq · Jan 17, 2025 · e94a62e · e94a62e
1 parent fd51eee
commit e94a62e
Show file tree

Hide file tree

Showing 19 changed files with 1,287 additions and 102 deletions.
diff --git a/src/@types/parseable/api/correlation.ts b/src/@types/parseable/api/correlation.ts
@@ -0,0 +1,18 @@
+export type Correlation = {
+	version: string;
+	id: string;
+	title: string;
+	tableConfigs: Array<{
+		selectedFields: string[];
+		tableName: string;
+	}>;
+	joinConfig: {
+		joinConditions: Array<{
+			tableName: string;
+			field: string;
+		}>;
+	};
+	filter: null;
+	startTime: string;
+	endTime: string;
+};
diff --git a/src/api/constants.ts b/src/api/constants.ts
@@ -26,6 +26,10 @@ export const LOG_QUERY_URL = (params?: Params, resourcePath = 'query') =>
 export const LOG_STREAMS_ALERTS_URL = (streamName: string) => `${LOG_STREAM_LIST_URL}/${streamName}/alert`;
 export const LIST_SAVED_FILTERS_URL = `${API_V1}/filters`;
 export const LIST_DASHBOARDS = `${API_V1}/dashboards`;
+export const LIST_CORRELATIONS = `${API_V1}/correlation`;
+export const UPDATE_CORRELATION_URL = (correlationId: string) => `${API_V1}/correlation/${correlationId}`;
+export const DELETE_SAVED_CORRELATION_URL = (correlationId: string) => `${API_V1}/correlation/${correlationId}`;
+export const GET_SAVED_CORRELATION_URL = (correlationId: string) => `${API_V1}/correlation/${correlationId}`;
 export const UPDATE_SAVED_FILTERS_URL = (filterId: string) => `${API_V1}/filters/${filterId}`;
 export const UPDATE_DASHBOARDS_URL = (dashboardId: string) => `${API_V1}/dashboards/${dashboardId}`;
 export const DELETE_DASHBOARDS_URL = (dashboardId: string) => `${API_V1}/dashboards/${dashboardId}`;

diff --git a/src/api/correlations.ts b/src/api/correlations.ts
@@ -0,0 +1,28 @@
+import { Correlation } from '@/@types/parseable/api/correlation';
+import { Axios } from './axios';
+import {
+	DELETE_SAVED_CORRELATION_URL,
+	GET_SAVED_CORRELATION_URL,
+	LIST_CORRELATIONS,
+	UPDATE_CORRELATION_URL,
+} from './constants';
+
+export const getCorrelations = () => {
+	return Axios().get<Correlation[]>(LIST_CORRELATIONS);
+};
+
+export const getCorrelationById = (correlationId: string) => {
+	return Axios().get(GET_SAVED_CORRELATION_URL(correlationId));
+};
+
+export const deleteSavedCorrelation = (correlationId: string) => {
+	return Axios().delete(DELETE_SAVED_CORRELATION_URL(correlationId));
+};
+
+export const saveCorrelation = (correlationData: Correlation) => {
+	return Axios().post(LIST_CORRELATIONS, correlationData);
+};
+
+export const updateCorrelation = (correlationData: Correlation) => {
+	return Axios().put(UPDATE_CORRELATION_URL(correlationData.id), correlationData);
+};
diff --git a/src/hooks/useCorrelationQueryLogs.tsx b/src/hooks/useCorrelationQueryLogs.tsx
@@ -1,5 +1,4 @@
 import { getCorrelationQueryLogsWithHeaders } from '@/api/query';
-import { StatusCodes } from 'http-status-codes';
 import useMountedState from './useMountedState';
 import { useAppStore } from '@/layouts/MainLayout/providers/AppProvider';
 import _ from 'lodash';
@@ -55,9 +54,9 @@ export const useCorrelationQueryLogs = () => {
 			enabled: false,
 			refetchOnWindowFocus: false,
 			onSuccess: async (responses) => {
-				responses.map((data: { data: LogsResponseWithHeaders; status: StatusCodes }) => {
+				responses.map((data: { data: LogsResponseWithHeaders }) => {
 					const logs = data.data;
-					const isInvalidResponse = _.isEmpty(logs) || _.isNil(logs) || data.status !== StatusCodes.OK;
+					const isInvalidResponse = _.isEmpty(logs) || _.isNil(logs);
 					if (isInvalidResponse) return setError('Failed to query logs');
 
 					const { records, fields } = logs;

diff --git a/src/hooks/useCorrelations.tsx b/src/hooks/useCorrelations.tsx
@@ -0,0 +1,157 @@
+import { useMutation, useQuery } from 'react-query';
+import _ from 'lodash';
+
+import {
+	deleteSavedCorrelation,
+	getCorrelationById,
+	getCorrelations,
+	saveCorrelation,
+	updateCorrelation,
+} from '@/api/correlations';
+import { correlationStoreReducers, useCorrelationStore } from '@/pages/Correlation/providers/CorrelationProvider';
+import { notifyError, notifySuccess } from '@/utils/notification';
+import { AxiosError, isAxiosError } from 'axios';
+import { appStoreReducers, useAppStore } from '@/layouts/MainLayout/providers/AppProvider';
+import dayjs from 'dayjs';
+
+const {
+	setCorrelations,
+	setActiveCorrelation,
+	setCorrelationId,
+	setSavedCorrelationId,
+	cleanCorrelationStore,
+	toggleSavedCorrelationsModal,
+} = correlationStoreReducers;
+const { setTimeRange, syncTimeRange } = appStoreReducers;
+export const useCorrelationsQuery = () => {
+	const [{ correlationId }, setCorrelatedStore] = useCorrelationStore((store) => store);
+	const [, setAppStore] = useAppStore((store) => store);
+	const {
+		isError: fetchCorrelationsError,
+		isSuccess: fetchCorrelationsSuccess,
+		isLoading: fetchCorrelationsLoading,
+		refetch: fetchCorrelations,
+	} = useQuery(['correlations'], () => getCorrelations(), {
+		retry: false,
+		enabled: false,
+		refetchOnWindowFocus: false,
+		onSuccess: (data) => {
+			setCorrelatedStore((store) => setCorrelations(store, data.data || []));
+		},
+		onError: () => {
+			setCorrelatedStore((store) => setCorrelations(store, []));
+			notifyError({ message: 'Failed to fetch correlations' });
+		},
+	});
+
+	const {
+		mutate: getCorrelationByIdMutation,
+		isError: fetchCorrelationIdError,
+		isSuccess: fetchCorrelationIdSuccess,
+		isLoading: fetchCorrelationIdLoading,
+	} = useMutation((correlationId: string) => getCorrelationById(correlationId), {
+		onSuccess: (data: any) => {
+			data.data.startTime &&
+				data.data.endTime &&
+				setAppStore((store) =>
+					setTimeRange(store, {
+						startTime: dayjs(data.data.startTime),
+						endTime: dayjs(data.data.endTime),
+						type: 'custom',
+					}),
+				);
+			setCorrelatedStore((store) => setCorrelationId(store, data.data.id));
+			setCorrelatedStore((store) => setActiveCorrelation(store, data.data));
+		},
+		onError: () => {
+			notifyError({ message: 'Failed to fetch correlation' });
+		},
+	});
+
+	const { mutate: deleteSavedCorrelationMutation, isLoading: isDeleting } = useMutation(
+		(data: { correlationId: string; onSuccess?: () => void }) => deleteSavedCorrelation(data.correlationId),
+		{
+			onSuccess: (_data, variables) => {
+				variables.onSuccess && variables.onSuccess();
+				if (variables.correlationId === correlationId) {
+					setCorrelatedStore(cleanCorrelationStore);
+					setAppStore(syncTimeRange);
+				}
+				fetchCorrelations();
+				setCorrelatedStore((store) => toggleSavedCorrelationsModal(store, false));
+				notifySuccess({ message: 'Deleted Successfully' });
+			},
+			onError: (data: AxiosError) => {
+				if (isAxiosError(data) && data.response) {
+					const error = data.response?.data as string;
+					typeof error === 'string' && notifyError({ message: error });
+				} else if (data.message && typeof data.message === 'string') {
+					notifyError({ message: data.message });
+				}
+			},
+		},
+	);
+
+	const { mutate: saveCorrelationMutation, isLoading: isCorrelationSaving } = useMutation(
+		(data: { correlationData: any; onSuccess?: () => void }) => saveCorrelation(data.correlationData),
+		{
+			onSuccess: (data, variables) => {
+				variables.onSuccess && variables.onSuccess();
+				setCorrelatedStore((store) => setCorrelationId(store, data.data.id));
+				setCorrelatedStore((store) => setSavedCorrelationId(store, data.data.id));
+				fetchCorrelations();
+				notifySuccess({ message: 'Correlation saved successfully' });
+			},
+			onError: (data: AxiosError) => {
+				if (isAxiosError(data) && data.response) {
+					const error = data.response?.data as string;
+					typeof error === 'string' && notifyError({ message: error });
+				} else if (data.message && typeof data.message === 'string') {
+					notifyError({ message: data.message });
+				}
+			},
+		},
+	);
+
+	const { mutate: updateCorrelationMutation, isLoading: isCorrelationUpdating } = useMutation(
+		(data: { correlationData: any; onSuccess?: () => void }) => updateCorrelation(data.correlationData),
+		{
+			onSuccess: (data, variables) => {
+				variables.onSuccess && variables.onSuccess();
+				setCorrelatedStore((store) => setCorrelationId(store, data.data.id));
+				setCorrelatedStore((store) => setActiveCorrelation(store, data.data));
+				fetchCorrelations();
+				notifySuccess({ message: 'Correlation updated successfully' });
+			},
+			onError: (data: AxiosError) => {
+				if (isAxiosError(data) && data.response) {
+					const error = data.response?.data as string;
+					typeof error === 'string' && notifyError({ message: error });
+				} else if (data.message && typeof data.message === 'string') {
+					notifyError({ message: data.message });
+				}
+			},
+		},
+	);
+
+	return {
+		fetchCorrelationsError,
+		fetchCorrelationsSuccess,
+		fetchCorrelationsLoading,
+		fetchCorrelations,
+
+		deleteSavedCorrelationMutation,
+		isDeleting,
+
+		fetchCorrelationIdError,
+		fetchCorrelationIdSuccess,
+		fetchCorrelationIdLoading,
+		getCorrelationByIdMutation,
+
+		saveCorrelationMutation,
+		isCorrelationSaving,
+
+		updateCorrelationMutation,
+		isCorrelationUpdating,
+	};
+};
diff --git a/src/hooks/useFetchStreamData.tsx b/src/hooks/useFetchStreamData.tsx
@@ -12,7 +12,6 @@ import {
 } from '@/pages/Correlation/providers/CorrelationProvider';
 import { notifyError } from '@/utils/notification';
 import { useQuery } from 'react-query';
-import { LogsResponseWithHeaders } from '@/@types/parseable/api/query';
 import { useRef, useEffect } from 'react';
 
 const { setStreamData } = correlationStoreReducers;
@@ -23,7 +22,6 @@ export const useFetchStreamData = () => {
 		(store) => store,
 	);
 	const [streamInfo] = useStreamStore((store) => store.info);
-	const [currentStream] = useAppStore((store) => store.currentStream);
 	const timePartitionColumn = _.get(streamInfo, 'time_partition', 'p_timestamp');
 	const [timeRange] = useAppStore((store) => store.timeRange);
 	const [
@@ -66,26 +64,29 @@ export const useFetchStreamData = () => {
 
 			const fetchPromises = streamsToFetch.map((streamName) => {
 				const queryOpts = { ...defaultQueryOpts, streamNames: [streamName] };
-				return getStreamDataWithHeaders(queryOpts);
+				return getStreamDataWithHeaders(queryOpts).then((data) => ({ streamName, data }));
 			});
 			return Promise.all(fetchPromises);
 		},
 		{
 			enabled: false,
 			refetchOnWindowFocus: false,
 			onSuccess: async (responses) => {
-				responses.map((data: { data: LogsResponseWithHeaders; status: StatusCodes }) => {
+				responses.forEach(({ streamName, data }) => {
 					const logs = data.data;
 					const isInvalidResponse = _.isEmpty(logs) || _.isNil(logs) || data.status !== StatusCodes.OK;
-					if (isInvalidResponse) return setError('Failed to query logs');
+					if (isInvalidResponse) {
+						setError('Failed to query logs');
+						return;
+					}
 
 					const { records, fields } = logs;
 					if (fields.length > 0 && !correlationCondition) {
-						return setCorrelationStore((store) => setStreamData(store, currentStream || '', records));
+						setCorrelationStore((store) => setStreamData(store, streamName, records));
 					} else if (fields.length > 0 && correlationCondition) {
-						return setCorrelationStore((store) => setStreamData(store, 'correlatedStream', records));
+						setCorrelationStore((store) => setStreamData(store, 'correlatedStream', records));
 					} else {
-						notifyError({ message: `${currentStream} doesn't have any fields` });
+						notifyError({ message: `${streamName} doesn't have any fields` });
 					}
 				});
 			},

diff --git a/src/hooks/useGetCorrelationStreamSchema.ts b/src/hooks/useGetCorrelationStreamSchema.ts
@@ -1,7 +1,7 @@
 import { getLogStreamSchema } from '@/api/logStream';
 import { AxiosError, isAxiosError } from 'axios';
 import _ from 'lodash';
-import { useQuery } from 'react-query';
+import { useQueries, useQuery } from 'react-query';
 import { useState } from 'react';
 import { correlationStoreReducers, useCorrelationStore } from '@/pages/Correlation/providers/CorrelationProvider';
 
@@ -43,3 +43,46 @@ export const useGetStreamSchema = (opts: { streamName: string }) => {
 		isRefetching,
 	};
 };
+
+// Multiple stream schemas hook
+export const useGetMultipleStreamSchemas = (streams: string[]) => {
+	const [, setCorrelationStore] = useCorrelationStore((_store) => null);
+	const [errors, setErrors] = useState<Record<string, string>>({});
+
+	const queries = useQueries(
+		streams.map((streamName) => ({
+			queryKey: ['stream-schema', streamName],
+			queryFn: () => getLogStreamSchema(streamName),
+			retry: false,
+			enabled: streamName !== '' && streamName !== 'correlatedStream',
+			refetchOnWindowFocus: false,
+			onSuccess: (data: any) => {
+				setErrors((prev) => _.omit(prev, streamName));
+				setCorrelationStore((store) => setStreamSchema(store, data.data, streamName));
+			},
+			onError: (error: AxiosError) => {
+				let errorMessage = 'An unknown error occurred';
+				if (isAxiosError(error) && error.response?.data) {
+					errorMessage = typeof error.response.data === 'string' ? error.response.data : error.message;
+				}
+				setErrors((prev) => ({
+					...prev,
+					[streamName]: errorMessage,
+				}));
+			},
+		})),
+	);
+
+	const isLoading = queries.some((query) => query.isLoading);
+	const isError = queries.some((query) => query.isError);
+	const isSuccess = queries.every((query) => query.isSuccess);
+	const isRefetching = queries.some((query) => query.isRefetching);
+
+	return {
+		isLoading,
+		isError,
+		isSuccess,
+		isRefetching,
+		errors,
+	};
+};