import { buildSchema, DocumentNode, GraphQLError, Kind, parse, TypeInfo, validate } from 'graphql';
import PromiseQueue from 'p-queue';
import { z } from 'zod';
import { collectSchemaCoordinates, preprocessOperation } from '@graphql-hive/core';
import { buildOperationS3BucketKey } from '@hive/cdn-script/artifact-storage-reader';
import { ServiceLogger } from '@hive/service-common';
import { sql as c_sql, ClickHouse } from '../../operations/providers/clickhouse-client';
import { S3Config } from '../../shared/providers/s3-config';

type DocumentRecord = {
  appDeploymentId: string;
  /** hash as provided by the user */
  hash: string;
  /** hash as used by usage reporting */
  internalHash: string;
  body: string;
  operationName: string | null;
  schemaCoordinates: Array<string>;
};

const AppDeploymentOperationHashModel = z
  .string()
  .trim()
  .min(1, 'Hash must be at least 1 characters long')
  .max(128, 'Hash must be at most 128 characters long')
  .regex(
    /^([A-Za-z]|[0-9]|_|-)+$/,
    "Operation hash can only contain letters, numbers, '_', and '-'",
  );

const AppDeploymentOperationBodyModel = z.string().min(3, 'Body must be at least 3 character long');

export type BatchProcessEvent = {
  event: 'PROCESS';
  id: string;
  data: {
    schemaSdl: string;
    targetId: string;
    appDeployment: {
      id: string;
      name: string;
      version: string;
    };
    documents: ReadonlyArray<{ hash: string; body: string }>;
  };
};

export type BatchProcessedEvent = {
  event: 'processedBatch';
  id: string;
  data:
    | {
        type: 'error';
        error: {
          message: string;
          details: {
            /** index of the operation causing an issue */
            index: number;
            /** message with additional details (either parse or validation error) */
            message: string;
          };
        };
      }
    | {
        type: 'success';
      };
};

/**
 * Callback invoked when documents are successfully persisted to S3.
 * Can be used to prefill a Redis cache during deployment.
 *
 * @example
 * ```typescript
 * const onDocumentsPersisted: OnDocumentsPersistedCallback = async (documents) => {
 *   for (const { key, body } of documents) {
 *     await redis.set(`hive:pd:${key}`, body, { EX: 3600 });
 *   }
 * };
 * ```
 */
export type OnDocumentsPersistedCallback = (
  documents: Array<{
    key: string; // targetId~appName~appVersion~hash
    body: string;
  }>,
) => Promise<void>;

export class PersistedDocumentIngester {
  private promiseQueue = new PromiseQueue({ concurrency: 30 });
  private logger: ServiceLogger;

  constructor(
    private clickhouse: ClickHouse,
    private s3: S3Config,
    logger: ServiceLogger,
    private onDocumentsPersisted?: OnDocumentsPersistedCallback,
  ) {
    this.logger = logger.child({ source: 'PersistedDocumentIngester' });
  }

  async processBatch(data: BatchProcessEvent['data']) {
    this.logger.debug(
      'Processing batch. (targetId=%s, appDeploymentId=%s, operationCount=%d)',
      data.targetId,
      data.appDeployment.id,
      data.documents.length,
    );

    const schema = buildSchema(data.schemaSdl);
    const typeInfo = new TypeInfo(schema);
    const documents: Array<DocumentRecord> = [];

    let index = 0;
    for (const operation of data.documents) {
      const hashValidation = AppDeploymentOperationHashModel.safeParse(operation.hash);
      const bodyValidation = AppDeploymentOperationBodyModel.safeParse(operation.body);

      if (hashValidation.success === false || bodyValidation.success === false) {
        this.logger.debug(
          'Invalid operation provided. Processing failed. (targetId=%s, appDeploymentId=%s, operationIndex=%d)',
          data.targetId,
          data.appDeployment.id,
          index,
        );

        return {
          type: 'error' as const,
          error: {
            // TODO: we should add more details (what hash is affected etc.)
            message: 'Invalid input, please check the operations.',
            details: {
              index,
              message:
                hashValidation.error?.issues[0].message ??
                bodyValidation.error?.issues[0].message ??
                'Invalid hash or body provided',
            },
          },
        };
      }
      let documentNode: DocumentNode;
      try {
        documentNode = parse(operation.body);
      } catch (err) {
        if (err instanceof GraphQLError) {
          console.error(err);
          this.logger.debug(
            'Failed parsing GraphQL operation. (targetId=%s, appDeploymentId=%s, operationIndex=%d)',
            data.targetId,
            data.appDeployment.id,
            index,
          );

          return {
            type: 'error' as const,
            error: {
              message: 'Failed to parse a GraphQL operation.',
              details: {
                index,
                message: err.message,
              },
            },
          };
        }
        throw err;
      }
      const errors = validate(schema, documentNode, undefined, {
        maxErrors: 1,
      });

      if (errors.length > 0) {
        this.logger.debug(
          'GraphQL operation did not pass validation against latest valid schema version. (targetId=%s, appDeploymentId=%s, operationIndex=%d)',
          data.targetId,
          data.appDeployment.id,
          index,
        );

        return {
          type: 'error' as const,
          error: {
            message: 'The GraphQL operation is not valid against the latest schema version.',
            details: {
              index,
              message: errors[0].message,
            },
          },
        };
      }

      const operationNames = getOperationNames(documentNode);
      if (operationNames.length > 1) {
        return {
          type: 'error' as const,
          error: {
            message: 'Only one executable operation definition is allowed per document.',
            details: {
              index,
              message:
                'Multiple operation definitions found. Only one executable operation definition is allowed per document.',
            },
          },
        };
      }

      const operationName = operationNames[0] ?? null;

      const schemaCoordinates = collectSchemaCoordinates({
        documentNode,
        processVariables: false,
        variables: null,
        schema,
        typeInfo,
      });

      const normalizedOperation = preprocessOperation({
        document: documentNode,
        operationName,
        schemaCoordinates,
      });

      documents.push({
        appDeploymentId: data.appDeployment.id,
        hash: operation.hash,
        internalHash: normalizedOperation?.hash ?? operation.hash,
        body: operation.body,
        operationName,
        schemaCoordinates: Array.from(schemaCoordinates),
      });

      index++;
    }

    if (documents.length) {
      this.logger.debug(
        'inserting documents into clickhouse and s3. (targetId=%s, appDeployment=%s, documentCount=%d)',
        data.targetId,
        data.appDeployment.id,
        documents.length,
      );

      await this.insertDocuments({
        targetId: data.targetId,
        appDeployment: data.appDeployment,
        documents: documents,
      });
    }

    return {
      type: 'success' as const,
    };
  }

  private async insertClickHouseDocuments(args: {
    targetId: string;
    appDeployment: {
      id: string;
    };
    documents: Array<DocumentRecord>;
  }) {
    // 1. Insert into ClickHouse
    this.logger.debug(
      'Inserting documents into ClickHouse. (targetId=%s, appDeployment=%s, documentCount=%d)',
      args.targetId,
      args.appDeployment.id,
      args.documents.length,
    );

    await this.clickhouse.insert({
      query: c_sql`
        INSERT INTO "app_deployment_documents" (
          "target_id"
          , "app_deployment_id"
          , "document_hash"
          , "document_body"
          , "operation_name"
          , "schema_coordinates"
          , "hash"
        )
        FORMAT CSV`,
      data: args.documents.map(document => [
        args.targetId,
        document.appDeploymentId,
        document.hash,
        document.body,
        document.operationName ?? '',
        document.schemaCoordinates,
        document.internalHash,
      ]),
      timeout: 10_000,
      queryId: 'insert_app_deployment_documents',
    });

    this.logger.debug(
      'Inserting documents into ClickHouse finished. (targetId=%s, appDeployment=%s, documentCount=%d)',
      args.targetId,
      args.appDeployment.id,
      args.documents.length,
    );
  }

  private async insertS3Documents(args: {
    targetId: string;
    appDeployment: {
      id: string;
      name: string;
      version: string;
    };
    documents: Array<DocumentRecord>;
  }) {
    this.logger.debug(
      'Inserting documents into S3. (targetId=%s, appDeployment=%s, documentCount=%d)',
      args.targetId,
      args.appDeployment.id,
      args.documents.length,
    );

    /** We parallelize and queue the requests. */
    const tasks: Array<Promise<void>> = [];

    for (const document of args.documents) {
      const s3Key = buildOperationS3BucketKey(
        args.targetId,
        args.appDeployment.name,
        args.appDeployment.version,
        document.hash,
      );

      tasks.push(
        this.promiseQueue.add(async () => {
          for (const s3 of this.s3) {
            const response = await s3.client.fetch([s3.endpoint, s3.bucket, s3Key].join('/'), {
              method: 'PUT',
              headers: {
                'content-type': 'text/plain',
              },
              body: document.body,
              aws: {
                // This boolean makes Google Cloud Storage & AWS happy.
                signQuery: true,
              },
            });

            if (response.statusCode !== 200) {
              throw new Error(
                `Failed to upload operation to S3: [${response.statusCode}] ${response.statusMessage}`,
              );
            }
          }
        }),
      );
    }

    await Promise.all(tasks);

    this.logger.debug(
      'Inserting documents into S3 finished. (targetId=%s, appDeployment=%s, documentCount=%d)',
      args.targetId,
      args.appDeployment.id,
      args.documents.length,
    );

    // Trigger cache prefill callback if configured
    if (this.onDocumentsPersisted) {
      const docsForCache = args.documents.map(doc => ({
        // Key format matches what the SDK uses for lookups: targetId~appName~appVersion~hash
        key: `${args.targetId}~${args.appDeployment.name}~${args.appDeployment.version}~${doc.hash}`,
        body: doc.body,
      }));

      try {
        await this.onDocumentsPersisted(docsForCache);
        this.logger.debug(
          'Cache prefill callback completed. (targetId=%s, appDeployment=%s, documentCount=%d)',
          args.targetId,
          args.appDeployment.id,
          docsForCache.length,
        );
      } catch (error) {
        // Don't fail the deployment for cache prefill failures
        this.logger.warn(
          { error },
          'Cache prefill callback failed. (targetId=%s, appDeployment=%s)',
          args.targetId,
          args.appDeployment.id,
        );
      }
    }
  }

  /** inserts operations of an app deployment into clickhouse and s3 */
  private async insertDocuments(args: {
    targetId: string;
    appDeployment: {
      id: string;
      name: string;
      version: string;
    };
    documents: Array<DocumentRecord>;
  }) {
    await Promise.all([
      // prettier-ignore
      this.insertClickHouseDocuments(args),
      this.insertS3Documents(args),
    ]);
  }
}

function getOperationNames(query: DocumentNode): Array<string> {
  const names: Array<string> = [];
  for (const node of query.definitions) {
    if (node.kind === Kind.OPERATION_DEFINITION && node.name?.value) {
      names.push(node.name.value);
    }
  }

  return names;
}
