/*!
 * 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.
 */
import { Flex, HStack, Link, Text } from "@chakra-ui/react";
import type { ColumnDef } from "@tanstack/react-table";
import type { TFunction } from "i18next";
import { useTranslation } from "react-i18next";
import { Link as RouterLink, useParams, useSearchParams } from "react-router-dom";

import { useDagRunServiceGetDagRuns } from "openapi/queries";
import type { DAGRunResponse } from "openapi/requests/types.gen";
import { ClearRunButton } from "src/components/Clear";
import { DagVersion } from "src/components/DagVersion";
import { DataTable } from "src/components/DataTable";
import { useTableURLState } from "src/components/DataTable/useTableUrlState";
import { ErrorAlert } from "src/components/ErrorAlert";
import { LimitedItemsList } from "src/components/LimitedItemsList";
import { MarkRunAsButton } from "src/components/MarkAs";
import RenderedJsonField from "src/components/RenderedJsonField";
import { RunTypeIcon } from "src/components/RunTypeIcon";
import { StateBadge } from "src/components/StateBadge";
import Time from "src/components/Time";
import { TruncatedText } from "src/components/TruncatedText";
import { SearchParamsKeys, type SearchParamsKeysType } from "src/constants/searchParams";
import { useAdvancedSearchArg } from "src/hooks/useAdvancedSearch";
import { DagRunsFilters } from "src/pages/DagRunsFilters";
import DeleteRunButton from "src/pages/DeleteRunButton";
import { renderDuration, useAutoRefresh, isStatePending } from "src/utils";

type DagRunRow = { row: { original: DAGRunResponse } };
const {
  BUNDLE_VERSION: BUNDLE_VERSION_PARAM,
  CONF_CONTAINS: CONF_CONTAINS_PARAM,
  CONSUMING_ASSET_PATTERN: CONSUMING_ASSET_PATTERN_PARAM,
  DAG_ID_PATTERN: DAG_ID_PATTERN_PARAM,
  DAG_VERSION: DAG_VERSION_PARAM,
  DURATION_GTE: DURATION_GTE_PARAM,
  DURATION_LTE: DURATION_LTE_PARAM,
  END_DATE_GTE: END_DATE_GTE_PARAM,
  END_DATE_LTE: END_DATE_LTE_PARAM,
  LOGICAL_DATE_GTE: LOGICAL_DATE_GTE_PARAM,
  LOGICAL_DATE_LTE: LOGICAL_DATE_LTE_PARAM,
  PARTITION_KEY_PATTERN: PARTITION_KEY_PATTERN_PARAM,
  RUN_AFTER_GTE: RUN_AFTER_GTE_PARAM,
  RUN_AFTER_LTE: RUN_AFTER_LTE_PARAM,
  RUN_ID_PATTERN: RUN_ID_PATTERN_PARAM,
  RUN_TYPE: RUN_TYPE_PARAM,
  START_DATE_GTE: START_DATE_GTE_PARAM,
  START_DATE_LTE: START_DATE_LTE_PARAM,
  STATE: STATE_PARAM,
  TRIGGERING_USER_NAME_PATTERN: TRIGGERING_USER_NAME_PATTERN_PARAM,
}: SearchParamsKeysType = SearchParamsKeys;

const runColumns = (translate: TFunction, dagId?: string): Array<ColumnDef<DAGRunResponse>> => [
  ...(Boolean(dagId)
    ? []
    : [
        {
          accessorKey: "dag_display_name",
          cell: ({ row: { original } }: DagRunRow) => (
            <Link asChild color="fg.info">
              <RouterLink to={`/dags/${original.dag_id}`}>
                <TruncatedText text={original.dag_display_name} />
              </RouterLink>
            </Link>
          ),
          enableSorting: false,
          header: translate("dagId"),
        },
      ]),
  {
    accessorKey: "dag_run_id",
    cell: ({ row: { original } }: DagRunRow) => (
      <Link asChild color="fg.info" fontWeight="bold">
        <RouterLink to={`/dags/${original.dag_id}/runs/${original.dag_run_id}`}>
          <TruncatedText text={original.dag_run_id} />
        </RouterLink>
      </Link>
    ),
    header: translate("dagRunId"),
  },
  {
    accessorKey: "run_after",
    cell: ({ row: { original } }: DagRunRow) => (
      <Link asChild color="fg.info" fontWeight="bold">
        <RouterLink to={`/dags/${original.dag_id}/runs/${original.dag_run_id}`}>
          <Time datetime={original.run_after} />
        </RouterLink>
      </Link>
    ),
    header: translate("dagRun.runAfter"),
  },
  {
    accessorKey: "state",
    cell: ({
      row: {
        original: { state },
      },
    }) => <StateBadge state={state}>{translate(`common:states.${state}`)}</StateBadge>,
    header: () => translate("state"),
  },
  {
    accessorKey: "run_type",
    cell: ({ row: { original } }) => (
      <HStack>
        <RunTypeIcon runType={original.run_type} />
        <Text>{translate(`common:runTypes.${original.run_type}`)}</Text>
      </HStack>
    ),
    enableSorting: false,
    header: translate("dagRun.runType"),
  },
  {
    accessorKey: "triggering_user_name",
    cell: ({ row: { original } }) => <Text>{original.triggering_user_name ?? ""}</Text>,
    enableSorting: false,
    header: translate("dagRun.triggeringUser"),
  },
  {
    accessorKey: "start_date",
    cell: ({ row: { original } }) => <Time datetime={original.start_date} />,
    header: translate("startDate"),
  },
  {
    accessorKey: "end_date",
    cell: ({ row: { original } }) => <Time datetime={original.end_date} />,
    header: translate("endDate"),
  },
  {
    accessorKey: "duration",
    cell: ({ row: { original } }) => renderDuration(original.duration),
    header: translate("duration"),
  },
  {
    accessorKey: "dag_versions",
    cell: ({ row: { original } }) => (
      <LimitedItemsList
        items={original.dag_versions.map((version) => (
          <DagVersion key={version.id} version={version} />
        ))}
        maxItems={4}
      />
    ),
    enableSorting: false,
    header: translate("dagRun.dagVersions"),
  },
  {
    accessorKey: "partition_key",
    cell: ({ row: { original } }) => <Text>{original.partition_key ?? ""}</Text>,
    enableSorting: false,
    header: translate("dagRun.partitionKey"),
  },
  {
    accessorKey: "conf",
    cell: ({ row: { original } }) =>
      original.conf && Object.keys(original.conf).length > 0 ? (
        <RenderedJsonField collapsed content={original.conf} />
      ) : undefined,
    header: translate("dagRun.conf"),
  },
  {
    accessorKey: "actions",
    cell: ({ row }) => (
      <Flex justifyContent="end">
        <ClearRunButton dagRun={row.original} />
        <MarkRunAsButton dagRun={row.original} />
        <DeleteRunButton dagRun={row.original} />
      </Flex>
    ),
    enableSorting: false,
    header: "",
    meta: {
      skeletonWidth: 10,
    },
  },
];

export const DagRuns = () => {
  const { t: translate } = useTranslation();
  const { dagId } = useParams();
  const [searchParams] = useSearchParams();

  const { setTableURLState, tableURLState } = useTableURLState({
    columnVisibility: {
      conf: false,
      dag_version: false,
      end_date: false,
      partition_key: false,
    },
  });
  const { cursor, pagination, sorting } = tableURLState;
  const [sort] = sorting;
  const orderBy = sort ? [`${sort.desc ? "-" : ""}${sort.id}`] : ["-run_after"];

  const { pageSize } = pagination;
  const filteredState = searchParams.get(STATE_PARAM);
  const filteredType = searchParams.get(RUN_TYPE_PARAM);
  const filteredRunIdPattern = searchParams.get(RUN_ID_PATTERN_PARAM);
  const filteredTriggeringUserNamePattern = searchParams.get(TRIGGERING_USER_NAME_PATTERN_PARAM);
  const filteredDagIdPattern = searchParams.get(DAG_ID_PATTERN_PARAM);
  const filteredDagVersion = searchParams.get(DAG_VERSION_PARAM);
  const filteredConsumingAsset = searchParams.get(CONSUMING_ASSET_PATTERN_PARAM);
  const bundleVersion = searchParams.get(BUNDLE_VERSION_PARAM);
  const startDateGte = searchParams.get(START_DATE_GTE_PARAM);
  const startDateLte = searchParams.get(START_DATE_LTE_PARAM);
  const endDateGte = searchParams.get(END_DATE_GTE_PARAM);
  const endDateLte = searchParams.get(END_DATE_LTE_PARAM);
  const logicalDateGte = searchParams.get(LOGICAL_DATE_GTE_PARAM);
  const logicalDateLte = searchParams.get(LOGICAL_DATE_LTE_PARAM);
  const runAfterGte = searchParams.get(RUN_AFTER_GTE_PARAM);
  const runAfterLte = searchParams.get(RUN_AFTER_LTE_PARAM);
  const durationGte = searchParams.get(DURATION_GTE_PARAM);
  const durationLte = searchParams.get(DURATION_LTE_PARAM);
  const confContains = searchParams.get(CONF_CONTAINS_PARAM);
  const partitionKeyPattern = searchParams.get(PARTITION_KEY_PATTERN_PARAM);

  const refetchInterval = useAutoRefresh({});

  const dagIdPatternArg = useAdvancedSearchArg({
    patternApiKey: "dagIdPattern",
    prefixApiKey: "dagIdPrefixPattern",
    storageKey: DAG_ID_PATTERN_PARAM,
    value: filteredDagIdPattern,
  });
  const runIdPatternArg = useAdvancedSearchArg({
    patternApiKey: "runIdPattern",
    prefixApiKey: "runIdPrefixPattern",
    storageKey: RUN_ID_PATTERN_PARAM,
    value: filteredRunIdPattern,
  });
  const triggeringUserArg = useAdvancedSearchArg({
    patternApiKey: "triggeringUserNamePattern",
    prefixApiKey: "triggeringUserNamePrefixPattern",
    storageKey: TRIGGERING_USER_NAME_PATTERN_PARAM,
    value: filteredTriggeringUserNamePattern,
  });
  const partitionKeyArg = useAdvancedSearchArg({
    patternApiKey: "partitionKeyPattern",
    prefixApiKey: "partitionKeyPrefixPattern",
    storageKey: PARTITION_KEY_PATTERN_PARAM,
    value: partitionKeyPattern,
  });

  const { data, error, isLoading } = useDagRunServiceGetDagRuns(
    {
      bundleVersion: bundleVersion ?? undefined,
      confContains: confContains !== null && confContains !== "" ? confContains : undefined,
      consumingAssetPattern: filteredConsumingAsset ?? undefined,
      cursor: cursor ?? "",
      dagId: dagId ?? "~",
      ...dagIdPatternArg,
      dagVersion:
        filteredDagVersion !== null && filteredDagVersion !== "" ? [Number(filteredDagVersion)] : undefined,
      durationGte: durationGte !== null && durationGte !== "" ? Number(durationGte) : undefined,
      durationLte: durationLte !== null && durationLte !== "" ? Number(durationLte) : undefined,
      endDateGte: endDateGte ?? undefined,
      endDateLte: endDateLte ?? undefined,
      limit: pageSize,
      logicalDateGte: logicalDateGte ?? undefined,
      logicalDateLte: logicalDateLte ?? undefined,
      orderBy,
      ...partitionKeyArg,
      runAfterGte: runAfterGte ?? undefined,
      runAfterLte: runAfterLte ?? undefined,
      ...runIdPatternArg,
      runType: filteredType === null ? undefined : [filteredType],
      startDateGte: startDateGte ?? undefined,
      startDateLte: startDateLte ?? undefined,
      state: filteredState === null ? undefined : [filteredState],
      ...triggeringUserArg,
    },
    undefined,
    {
      placeholderData: (prev) => prev,
      refetchInterval: (query) =>
        query.state.data?.dag_runs.some((run) => isStatePending(run.state)) ? refetchInterval : false,
    },
  );

  const columns = runColumns(translate, dagId);

  const nextCursor = data?.next_cursor ?? undefined;
  const previousCursor = data?.previous_cursor ?? undefined;

  return (
    <>
      <DagRunsFilters dagId={dagId} />
      <DataTable
        columns={columns}
        data={data?.dag_runs ?? []}
        errorMessage={<ErrorAlert error={error} />}
        initialState={tableURLState}
        isLoading={isLoading}
        modelName="common:dagRun"
        nextCursor={nextCursor}
        onStateChange={setTableURLState}
        previousCursor={previousCursor}
      />
    </>
  );
};
