import { KafkaV1 as Kafka } from '@superblocksteam/types';

import { EditorLanguage, FormComponentType, InputDataType, Plugin, PluginResponseType, PluginType } from '../../types/index.js';
import { PUBLIC_INTEGRATIONS_LOGO_URL } from './constants.js';

const KafkaPluginVersions = {
  V1: '0.0.1'
};

// some constants to share with the ui
export const KAFKA_MODULE_NAME = 'KafkaPlugin';
export const OPERATION_NAME = 'operation';
export const OPERATION_PRODUCE = 'OPERATION_PRODUCE';
export const OPERATION_CONSUME = 'OPERATION_CONSUME';

const MESSAGES_PLACEHOLDER = `[
  {
    "topic": "my_topic_1",
    "value": "Superblocks!"
  },
  {
    "topic": "my_topic_2",
    "value": {
      "koala": "bear"
    }
  }
]`;

const ALL_SASL_MECHANISMS_AND_DISPLAY_NAMES: Record<string, string> = {
  [Kafka.SASL_Mechanism.PLAIN]: 'PLAIN',
  [Kafka.SASL_Mechanism.SCRAM_SHA256]: 'SCRAM SHA256',
  [Kafka.SASL_Mechanism.SCRAM_SHA512]: 'SCRAM SHA512',
  [Kafka.SASL_Mechanism.AWS]: 'AWS'
};

const EXCLUDE_AWS_SASL_MECHANISMS_AND_DISPLAY_NAMES: Record<string, string> = Object.fromEntries(
  Object.entries(ALL_SASL_MECHANISMS_AND_DISPLAY_NAMES).filter(([key, value]) => key !== String(Kafka.SASL_Mechanism.AWS))
);

const KAFKA_ACTIONS_AND_DISPLAY_NAMES: Record<string, string> = {
  OPERATION_CONSUME: 'Consume',
  OPERATION_PRODUCE: 'Produce'
};

const KAFKA_COMPRESSIONS_AND_DISPLAY_NAMES: Record<string, string> = {
  COMPRESSION_GZIP: 'GZIP',
  COMPRESSION_SNAPPY: 'SNAPPY',
  COMPRESSION_LZ4: 'LZ4',
  COMPRESSION_ZSTD: 'ZSTD',
  COMPRESSION_UNSPECIFIED: 'None'
};

const CONSUME_FROM_VALUES_AND_DISPLAY_NAMES: Record<string, string> = {
  // FROM_SEEK: 'Offset', // We found bugs here when testing
  FROM_LATEST: 'Latest',
  FROM_BEGINNING: 'Beginning'
};

const ACKS_FROM_VALUES_AND_DISPLAY_NAMES: Record<string, string> = {
  ACKS_ALL: 'All',
  ACKS_NONE: 'None',
  ACKS_LEADER: 'Leaders'
};

const makeDropdownItem = (value: string, displayName?: string, subText?: string) => {
  return {
    key: value,
    value: value,
    displayName: displayName ?? value,
    subText: subText ?? ''
  };
};

// TODO(docs-v2): no confluent page exists in docs-v2; add one and restore docsUrl
export const ConfluentPlugin = generateKafkaPlugin(
  'confluent',
  'Confluent',
  `${PUBLIC_INTEGRATIONS_LOGO_URL}/confluent.png`,
  undefined,
  EXCLUDE_AWS_SASL_MECHANISMS_AND_DISPLAY_NAMES
);

// TODO(docs-v2): no kafka page exists in docs-v2; add one and restore docsUrl
export const KafkaPlugin = generateKafkaPlugin(
  'kafka',
  'Kafka',
  `${PUBLIC_INTEGRATIONS_LOGO_URL}/kafka.png`,
  undefined,
  ALL_SASL_MECHANISMS_AND_DISPLAY_NAMES
);

// TODO(docs-v2): no redpanda page exists in docs-v2; add one and restore docsUrl
export const RedpandaPlugin = generateKafkaPlugin(
  'redpanda',
  'Redpanda',
  `${PUBLIC_INTEGRATIONS_LOGO_URL}/redpanda.png`,
  undefined,
  EXCLUDE_AWS_SASL_MECHANISMS_AND_DISPLAY_NAMES
);

function generateKafkaPlugin(
  id: string,
  name: string,
  iconLocation: string,
  docsUrl: string | undefined,
  saslMechanisms: Record<string, string>
): Plugin {
  let auto_create_tooltip = `Automatically create the topic if it doesn't already exist.`;
  if (id === 'confluent') {
    auto_create_tooltip +=
      ' Note, this setting only works for customers running a [Dedicated Cluster](https://docs.confluent.io/cloud/current/clusters/broker-config.html) on Confluent Cloud.';
  }
  return {
    id,
    name,
    moduleName: KAFKA_MODULE_NAME,
    modulePath: 'plugins/kafka/KafkaPlugin',
    iconLocation,
    docsUrl,
    type: PluginType.API,
    responseType: PluginResponseType.JSON,
    hasRawRequest: false,
    hasMetadata: true,
    rawRequestName: 'Executed request',
    isStreamable: true,
    datasourceTemplate: {
      sections: [
        {
          name: 'main',
          items: [
            {
              label: 'Display name',
              name: 'name',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              placeholder: `${name} Prod`,
              rules: [{ required: true, message: 'Display name is required' }]
            },
            {
              label: 'Brokers',
              placeholder: 'pkc-5aaa5.us-west4.gcp.confluent.cloud:9092',
              name: 'cluster.brokers',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              rules: [{ required: true, message: 'Brokers are required' }]
            }
          ]
        },
        {
          name: 'auth',
          borderThreshold: 1,
          items: [
            {
              label: 'SASL mechanism',
              name: 'cluster.sasl.mechanism',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.DROPDOWN,
              initialValue: String(Kafka.SASL_Mechanism.PLAIN),
              rules: [{ required: true }],
              options: Object.entries(saslMechanisms).map(([value, displayName]) => makeDropdownItem(value, displayName))
            },
            {
              label: 'Username',
              name: 'cluster.sasl.username',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              rules: [{ required: true, message: 'Username is required' }],
              display: {
                show: {
                  'cluster.sasl.mechanism': [
                    String(Kafka.SASL_Mechanism.PLAIN),
                    String(Kafka.SASL_Mechanism.SCRAM_SHA256),
                    String(Kafka.SASL_Mechanism.SCRAM_SHA512)
                  ]
                }
              }
            },
            {
              label: 'Password',
              name: 'cluster.sasl.password',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              dataType: InputDataType.PASSWORD,
              rules: [{ required: true, message: 'Password is required' }],
              display: {
                show: {
                  'cluster.sasl.mechanism': [
                    String(Kafka.SASL_Mechanism.PLAIN),
                    String(Kafka.SASL_Mechanism.SCRAM_SHA256),
                    String(Kafka.SASL_Mechanism.SCRAM_SHA512)
                  ]
                }
              }
            },
            {
              label: 'IAM role ARN',
              name: 'cluster.sasl.authorizationIdentity',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              rules: [{ required: true, message: 'IAM role ARN is required' }],
              display: {
                show: {
                  'cluster.sasl.mechanism': [String(Kafka.SASL_Mechanism.AWS)]
                }
              }
            },
            {
              label: 'Access key ID',
              name: 'cluster.sasl.accessKeyId',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              dataType: InputDataType.PASSWORD,
              rules: [{ required: true, message: 'Access key ID is required' }],
              display: {
                show: {
                  'cluster.sasl.mechanism': [String(Kafka.SASL_Mechanism.AWS)]
                }
              }
            },
            {
              label: 'Secret key',
              name: 'cluster.sasl.secretKey',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              dataType: InputDataType.PASSWORD,
              rules: [{ required: true, message: 'Secret key is required' }],
              display: {
                show: {
                  'cluster.sasl.mechanism': [String(Kafka.SASL_Mechanism.AWS)]
                }
              }
            }
          ]
        },
        {
          name: 'mainBottom',
          items: [
            {
              label: 'Enable SSL',
              name: 'cluster.ssl',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.CHECKBOX,
              initialValue: true
            }
          ]
        }
      ]
    },
    actionTemplate: {
      sections: [
        {
          name: 'main',
          items: [
            {
              label: 'Action',
              name: OPERATION_NAME,
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.DROPDOWN,
              rules: [{ required: true }],
              // logic in the front-end will choose the right default based on
              // the context. we should bring that in the form.
              initialValue: 'OPERATION_PRODUCE',
              options: Object.entries(KAFKA_ACTIONS_AND_DISPLAY_NAMES).map(([value, displayName]) => makeDropdownItem(value, displayName)),
              tooltip: {
                markdownText: 'The Kafka action you wish to perform'
              }
            },
            // produce
            {
              label: 'Messages',
              name: 'produce.messages',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.CODE_EDITOR,
              language: EditorLanguage.JSON,
              tooltip: {
                markdownText: docsUrl
                  ? `A JSON array of messages to produce. See [our documentation](${docsUrl}?tab=produce) for the complete schema`
                  : `A JSON array of messages to produce.`
              },
              placeholder: MESSAGES_PLACEHOLDER,
              display: {
                show: {
                  operation: [OPERATION_PRODUCE]
                }
              }
            },
            // consume
            {
              label: 'Topic',
              name: 'consume.topic',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              display: {
                show: {
                  operation: [OPERATION_CONSUME]
                }
              },
              tooltip: {
                markdownText: 'The Kafka topic to consume from'
              }
            },
            {
              label: 'Consume from',
              name: 'consume.from',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.DROPDOWN,
              rules: [{ required: true }],
              initialValue: 'FROM_BEGINNING',
              options: Object.entries(CONSUME_FROM_VALUES_AND_DISPLAY_NAMES).map(([value, displayName]) =>
                makeDropdownItem(value, displayName)
              ),
              display: {
                show: {
                  operation: [OPERATION_CONSUME]
                }
              },
              tooltip: {
                markdownText: 'Where to begin consuming messages from'
              }
            },
            {
              label: 'Offset',
              name: 'consume.seek.offset',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              display: {
                show: {
                  operation: [OPERATION_CONSUME],
                  'consume.from': ['FROM_SEEK']
                }
              }
            },
            {
              label: 'Partition',
              name: 'consume.seek.partition',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              display: {
                show: {
                  operation: [OPERATION_CONSUME],
                  'consume.from': ['FROM_SEEK']
                }
              }
            }
          ]
        },
        {
          name: 'advancedConfiguration',
          sectionHeader: 'Additional settings',
          defaultCollapsed: true,
          isCollapsible: true,
          asTitle: false,
          items: [
            // produce
            {
              label: 'Compression',
              name: 'produce.compression',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.DROPDOWN,
              initialValue: 'COMPRESSION_UNSPECIFIED',
              rules: [{ required: true }],
              display: {
                show: {
                  operation: [OPERATION_PRODUCE]
                }
              },
              options: Object.entries(KAFKA_COMPRESSIONS_AND_DISPLAY_NAMES).map(([value, displayName]) =>
                makeDropdownItem(value, displayName)
              ),
              tooltip: {
                markdownText: 'The compression algorithm to use when producing messages'
              }
            },
            {
              label: 'Auto create topic',
              name: 'produce.autoCreateTopic',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.CHECKBOX,
              initialValue: true,
              display: {
                show: {
                  operation: [OPERATION_PRODUCE]
                }
              },
              tooltip: {
                markdownText: auto_create_tooltip
              }
            },
            {
              label: 'Transaction',
              name: 'produce.transaction',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.CHECKBOX,
              initialValue: true,
              display: {
                show: {
                  operation: [OPERATION_PRODUCE]
                }
              },
              tooltip: {
                markdownText: 'Whether to produce messages as part of a transaction'
              }
            },
            {
              label: 'Transaction ID (optional)',
              name: 'produce.transactionId',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              placeholder: 'superblocks-fds01k',
              display: {
                show: {
                  operation: [OPERATION_PRODUCE],
                  'produce.transaction': ['true']
                }
              },
              tooltip: {
                markdownText: 'The transaction ID to use when producing messages. If not provided, one will be created'
              }
            },
            {
              label: 'Idempotent',
              name: 'produce.idempotent',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.CHECKBOX,
              initialValue: true,
              display: {
                show: {
                  operation: [OPERATION_PRODUCE],
                  'produce.transaction': ['false']
                }
              },
              tooltip: {
                markdownText: 'Whether to produce messages idempotently'
              }
            },
            {
              label: 'Acks',
              name: 'produce.acks',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.DROPDOWN,
              initialValue: 'ACKS_ALL',
              rules: [{ required: true }],
              display: {
                show: {
                  operation: [OPERATION_PRODUCE],
                  'produce.transaction': ['false'],
                  'produce.idempotent': ['false']
                }
              },
              options: Object.entries(ACKS_FROM_VALUES_AND_DISPLAY_NAMES).map(([value, displayName]) =>
                makeDropdownItem(value, displayName)
              ),
              tooltip: {
                markdownText: 'Who to wait for acknowledgement from before considering a message produced'
              }
            },
            {
              label: 'Client ID (optional)',
              name: 'produce.clientId',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              placeholder: 'superblocks-kd82k',
              display: {
                show: {
                  operation: [OPERATION_PRODUCE]
                }
              },
              hidden: id === 'confluent', // We want to use our partner client id for the confluent plugin.
              tooltip: {
                markdownText: 'The specific client ID to use when producing messages. If not provided, one will be created'
              }
            },
            // consume
            {
              label: 'Read uncommitted',
              name: 'consume.readUncommitted',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.CHECKBOX,
              initialValue: true,
              display: {
                show: {
                  operation: [OPERATION_CONSUME]
                }
              },
              tooltip: {
                markdownText:
                  'Whether or not to return all messages, even those that are not committed such as transactional messages which have been aborted'
              }
            },
            {
              label: 'Group ID (optional)',
              name: 'consume.groupId',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              placeholder: 'superblocks-5kd7s',
              display: {
                show: {
                  operation: [OPERATION_CONSUME]
                }
              },
              tooltip: {
                markdownText: 'The consumer group ID to use when consuming messages. If not provided, one will be created'
              }
            },
            {
              label: 'Client ID (optional)',
              name: 'consume.clientId',
              startVersion: KafkaPluginVersions.V1,
              componentType: FormComponentType.INPUT_TEXT,
              placeholder: 'superblocks-kd82k',
              display: {
                show: {
                  operation: [OPERATION_CONSUME]
                }
              },
              hidden: id === 'confluent', // We want to use our partner client id for the confluent plugin.
              tooltip: {
                markdownText: 'The specific client ID to use when consuming messages. If not provided, one will be created'
              }
            }
          ]
        }
      ]
    }
  };
}