import type { RuntimeLogger } from "../../../../core/types.ts";
import type { MigrationDb } from "../runtime-migration-runner.ts";
import type { DialectAdapter, IdempotencyInsertValues, RuntimeDialect } from "./dialect-adapter.ts";
import type { MySql2Database } from "drizzle-orm/mysql2";
import type mysql from "mysql2/promise";

import { sql } from "drizzle-orm";
import { drizzle } from "drizzle-orm/mysql2";
import { createPool } from "mysql2/promise";
import { runRuntimeMigrations } from "../runtime-migration-runner.ts";
import * as schema from "../schema/mysql.ts";
import { resolveDrizzleSession, runWithDrizzleSession } from "./session.ts";

export interface MysqlDialectOptions {
  connectionString: string;
  poolMax?: number;
  logger?: RuntimeLogger;
}

export interface MysqlDialect extends RuntimeDialect {
  readonly kind: "mysql";
  readonly db: MySql2Database<typeof schema>;
  readonly schema: typeof schema;
  readonly pool: mysql.Pool;
}

const migrationDirectory = new URL("../../../../../drizzle/mysql/", import.meta.url);

export async function createMysqlDialect(options: MysqlDialectOptions): Promise<MysqlDialect> {
  const pool = createPool({
    uri: options.connectionString,
    connectionLimit: options.poolMax ?? 10,
  });

  try {
    // Hold one connection for the entire migration run so START TRANSACTION/COMMIT share a session.
    const migrationConnection = await pool.getConnection();
    try {
      await runRuntimeMigrations({
        db: wrapMysqlMigrationConnection(migrationConnection),
        migrationDirectory,
        logger: options.logger,
        dialect: "mysql",
      });
    } finally {
      migrationConnection.release();
    }
  } catch (error) {
    await pool.end();
    throw error;
  }

  const rootDb = drizzle(pool, { schema, mode: "default" });
  const dialect: MysqlDialect = {
    kind: "mysql",
    get db(): MySql2Database<typeof schema> {
      return resolveDrizzleSession(rootDb);
    },
    schema,
    adapter: null as unknown as DialectAdapter,
    pool,
    async close(): Promise<void> {
      await pool.end();
    },
  };
  dialect.adapter = createMysqlAdapter(rootDb, () => dialect.db);

  return dialect;
}

function createMysqlAdapter(
  rootDb: MySql2Database<typeof schema>,
  getDb: () => MySql2Database<typeof schema>,
): DialectAdapter {
  return {
    kind: "mysql",

    async transaction<T>(work: () => Promise<T>): Promise<T> {
      return await rootDb.transaction(async (tx) => {
        return await runWithDrizzleSession(tx, work);
      });
    },

    async deleteRetentionTail(keepCount: number): Promise<void> {
      await getDb().execute(sql`
        delete from runs
        where id not in (
          select id from (
            select id from runs
            order by started_at desc, id desc
            limit ${keepCount}
          ) as keep_ids
        )
      `);
    },

    async insertIgnoreIdempotency(values: IdempotencyInsertValues): Promise<boolean> {
      const result = await getDb().execute(sql`
        insert ignore into idempotency_records (
          key_hash, claim_id, request_hash, state, response_value, created_at, expires_at
        ) values (
          ${values.keyHash},
          ${values.claimId},
          ${values.requestHash},
          'in_progress',
          null,
          ${values.createdAt},
          ${values.expiresAt}
        )
      `);
      const header = Array.isArray(result) ? result[0] : result;
      return Boolean(header && typeof header === "object" && (header as { affectedRows?: number }).affectedRows);
    },
  };
}

function wrapMysqlMigrationConnection(connection: mysql.PoolConnection): MigrationDb {
  return {
    async exec(sqlText: string): Promise<void> {
      await connection.query(sqlText);
    },
    prepare(sqlText: string) {
      return {
        async run(...params: unknown[]): Promise<{ changes: number }> {
          const [result] = await connection.query<mysql.ResultSetHeader>(sqlText, params);
          return { changes: result.affectedRows };
        },
        async all(...params: unknown[]): Promise<unknown[]> {
          const [rows] = await connection.query(sqlText, params);
          return rows as unknown[];
        },
        async get(...params: unknown[]): Promise<unknown> {
          const [rows] = await connection.query(sqlText, params);
          const list = rows as unknown[];
          return list[0];
        },
      };
    },
  };
}