🪟 🔧 Refactor Jobs APIs (#6204)

airbyte-webapp/src/components/JobItem/components/JobLogs.tsx

@@ -8,8 +8,8 @@ import { StatusIcon } from "components/ui/StatusIcon";
 import { StatusIconStatus } from "components/ui/StatusIcon/StatusIcon";
 import { Text } from "components/ui/Text";
 
+import { useGetDebugInfoJob } from "core/api";
 import { AttemptRead, AttemptStatus, SynchronousJobRead } from "core/request/AirbyteClient";
-import { useGetDebugInfoJob } from "services/job/JobService";
 
 import styles from "./JobLogs.module.scss";
 import { LogsDetails } from "./LogsDetails";

airbyte-webapp/src/components/connection/ConnectionSync/ConnectionSyncContext.tsx

@@ -4,6 +4,7 @@ import { createContext, useCallback, useContext, useEffect, useMemo, useState }
 import { Status as ConnectionSyncStatus } from "components/EntityTable/types";
 import { getConnectionSyncStatus } from "components/EntityTable/utils";
 
+import { useCancelJob } from "core/api";
 import {
   JobRead,
   ConnectionStatus,
@@ -15,7 +16,6 @@ import {
 } from "core/request/AirbyteClient";
 import { useConnectionEditService } from "hooks/services/ConnectionEdit/ConnectionEditService";
 import { useResetConnection, useResetConnectionStream, useSyncConnection } from "hooks/services/useConnectionHook";
-import { useCancelJob } from "services/job/JobService";
 import { moveTimeToFutureByPeriod } from "utils/time";
 
 interface ConnectionSyncContext {

airbyte-webapp/src/core/api/hooks/index.ts

@@ -1,5 +1,6 @@
 export * from "./geographies";
 export * from "./health";
+export * from "./jobs";
 export * from "./logs";
 export * from "./notifications";
 export * from "./operations";

airbyte-webapp/src/core/api/hooks/jobs.ts

@@ -0,0 +1,60 @@
+import { useMutation, useQuery } from "react-query";
+
+import { SCOPE_WORKSPACE } from "services/Scope";
+
+import { cancelJob, getJobDebugInfo, listJobsFor } from "../generated/AirbyteClient";
+import { JobDebugInfoRead, JobListRequestBody, JobReadList } from "../types/AirbyteClient";
+import { useRequestOptions } from "../useRequestOptions";
+import { useSuspenseQuery } from "../useSuspenseQuery";
+
+export const useListJobs = (listParams: JobListRequestBody, keepPreviousData = true) => {
+  const requestOptions = useRequestOptions();
+  const result = useQuery(
+    [SCOPE_WORKSPACE, "jobs", "list", listParams.configId, listParams.includingJobId, listParams.pagination],
+    () => listJobsFor(listParams, requestOptions),
+    {
+      // 2.5 second refresh
+      refetchInterval: 2500,
+      keepPreviousData,
+      suspense: true,
+    }
+  );
+  // cast to JobReadList because (suspense: true) means we will never get undefined
+  const jobReadList = result.data as JobReadList;
+  return { jobs: jobReadList.jobs, totalJobCount: jobReadList.totalJobCount, isPreviousData: result.isPreviousData };
+};
+
+export const useGetDebugInfoJob = (
+  id: number,
+  enabled = true,
+  refetchWhileRunning = false
+): JobDebugInfoRead | undefined => {
+  const requestOptions = useRequestOptions();
+
+  return useSuspenseQuery(
+    [SCOPE_WORKSPACE, "jobs", "getDebugInfo", id],
+    () => getJobDebugInfo({ id }, requestOptions),
+    {
+      refetchInterval: !refetchWhileRunning
+        ? false
+        : (data) => {
+            // If refetchWhileRunning was true, we keep refetching debug info (including logs), while the job is still
+            // running or hasn't ended too long ago. We need some time after the last attempt has stopped, since logs
+            // keep incoming for some time after the job has already been marked as finished.
+            const lastAttemptEndTimestamp =
+              data?.attempts.length && data.attempts[data.attempts.length - 1].attempt.endedAt;
+            // While no attempt ended timestamp exists yet (i.e. the job is still running) or it hasn't ended
+            // more than 2 minutes (2 * 60 * 1000ms) ago, keep refetching
+            return lastAttemptEndTimestamp && Date.now() - lastAttemptEndTimestamp * 1000 > 2 * 60 * 1000
+              ? false
+              : 2500;
+          },
+      enabled,
+    }
+  );
+};
+
+export const useCancelJob = () => {
+  const requestOptions = useRequestOptions();
+  return useMutation((id: number) => cancelJob({ id }, requestOptions));
+};

airbyte-webapp/src/hooks/services/useSourceHook.tsx

@@ -6,15 +6,14 @@ import { Action, Namespace } from "core/analytics";
 import { SyncSchema } from "core/domain/catalog";
 import { ConnectionConfiguration } from "core/domain/connection";
 import { SourceService } from "core/domain/connector/SourceService";
-import { JobInfo } from "core/domain/job";
 import { useInitService } from "services/useInitService";
 import { isDefined } from "utils/common";
 
 import { useAnalyticsService } from "./Analytics";
 import { useRemoveConnectionsFromList } from "./useConnectionHook";
 import { useRequestErrorHandler } from "./useRequestErrorHandler";
 import { useCurrentWorkspace } from "./useWorkspace";
-import { SourceRead, WebBackendConnectionListItem } from "../../core/request/AirbyteClient";
+import { SourceRead, SynchronousJobRead, WebBackendConnectionListItem } from "../../core/request/AirbyteClient";
 import { useSuspenseQuery } from "../../services/connector/useSuspenseQuery";
 import { SCOPE_WORKSPACE } from "../../services/Scope";
 import { useDefaultRequestMiddlewares } from "../../services/useDefaultRequestMiddlewares";
@@ -164,7 +163,7 @@ const useUpdateSource = () => {
   );
 };
 
-export type SchemaError = (Error & { status: number; response: JobInfo }) | null;
+export type SchemaError = (Error & { status: number; response: SynchronousJobRead }) | null;
 
 const useDiscoverSchema = (
   sourceId: string,

airbyte-webapp/src/pages/connections/ConnectionJobHistoryPage/ConnectionJobHistoryPage.tsx

@@ -12,9 +12,9 @@ import { Link } from "components/ui/Link";
 
 import { Action, Namespace } from "core/analytics";
 import { getFrequencyFromScheduleData } from "core/analytics/utils";
+import { useListJobs } from "core/api";
 import { useTrackPage, PageTrackingCodes, useAnalyticsService } from "hooks/services/Analytics";
 import { useConnectionEditService } from "hooks/services/ConnectionEdit/ConnectionEditService";
-import { useListJobs } from "services/job/JobService";
 
 import styles from "./ConnectionJobHistoryPage.module.scss";
 import JobsList from "./JobsList";

airbyte-webapp/src/pages/connections/StreamStatusPage/StreamStatusPage.tsx

@@ -1,8 +1,8 @@
 import { ConnectionSyncContextProvider } from "components/connection/ConnectionSync/ConnectionSyncContext";
 
+import { useListJobs } from "core/api";
 import { useConnectionEditService } from "hooks/services/ConnectionEdit/ConnectionEditService";
 import { useExperiment } from "hooks/services/Experiment";
-import { useListJobs } from "services/job/JobService";
 
 import { StreamsList } from "./StreamsList";
 import { StreamsListContextProvider } from "./StreamsListContext";