hypercore.ts

import type { Db } from "./db/db.ts";
import { sql } from "drizzle-orm";
import {
  createErrorString,
  createFail,
  createSuccess,
  isSuccess,
  type Result,
} from "@joyautomation/dark-matter";
import { log } from "./log.ts";
import type { getBuilder } from "@joyautomation/conch";
import { GraphQLError } from "graphql";

const historianEnabled = Deno.env.get("MANTLE_HISTORIAN_ENABLED") !== "false";
const retentionDays = parseInt(
  Deno.env.get("MANTLE_RETENTION_DAYS") || "30",
  10,
);

// Types for storage stats
type TableStorageStats = {
  tableName: string;
  totalBytes: number;
  compressedBytes: number | null;
  uncompressedBytes: number | null;
  compressionRatio: number | null;
};

type StorageStats = {
  hypercoreAvailable: boolean;
  compressionEnabled: boolean;
  tables: TableStorageStats[];
  totalStorageBytes: number;
  totalCompressedBytes: number | null;
  totalUncompressedBytes: number | null;
  overallCompressionRatio: number | null;
};

type CompressionStatus = {
  tableName: string;
  compressionEnabled: boolean;
  policyExists: boolean;
};

/**
 * Check if TimescaleDB compression (hypercore) is available
 */
export async function isHypercoreAvailable(db: Db): Promise<Result<boolean>> {
  try {
    // Check if timescaledb extension is installed and has compression support
    const result = await db.execute(sql`
      SELECT EXISTS (
        SELECT 1 FROM pg_extension WHERE extname = 'timescaledb'
      ) as has_timescaledb
    `);

    if (!result.rows[0]?.has_timescaledb) {
      return createSuccess(false);
    }

    // Check if compression is available (it's part of TimescaleDB)
    // Try to query compression-related catalog
    try {
      await db.execute(sql`
        SELECT 1 FROM timescaledb_information.compression_settings LIMIT 0
      `);
      return createSuccess(true);
    } catch {
      // Compression catalog doesn't exist - older version without compression
      return createSuccess(false);
    }
  } catch (error) {
    return createFail(createErrorString(error));
  }
}

/**
 * Check compression status for a specific table
 */
export async function getCompressionStatus(
  db: Db,
  tableName: string,
): Promise<Result<CompressionStatus>> {
  try {
    // Check if compression is enabled on the hypertable
    const compressionResult = await db.execute(sql.raw(`
      SELECT compression_enabled
      FROM timescaledb_information.hypertables
      WHERE hypertable_name = '${tableName}'
    `));

    const compressionEnabled = compressionResult.rows[0]?.compression_enabled === true;

    // Check if compression policy exists
    const policyResult = await db.execute(sql.raw(`
      SELECT 1
      FROM timescaledb_information.jobs j
      JOIN timescaledb_information.job_stats js ON j.job_id = js.job_id
      WHERE j.proc_name = 'policy_compression'
      AND j.hypertable_name = '${tableName}'
      LIMIT 1
    `));

    const policyExists = (policyResult.rows?.length ?? 0) > 0;

    return createSuccess({
      tableName,
      compressionEnabled,
      policyExists,
    });
  } catch (error) {
    return createFail(createErrorString(error));
  }
}

/**
 * Enable compression on the history table
 */
export async function enableHistoryCompression(db: Db): Promise<Result<void>> {
  try {
    log.info("Enabling compression on history table...");

    await db.execute(sql`
      ALTER TABLE history SET (
        timescaledb.compress,
        timescaledb.compress_segmentby = 'group_id, node_id, device_id, metric_id',
        timescaledb.compress_orderby = 'timestamp DESC NULLS FIRST'
      )
    `);

    log.info("Compression enabled on history table");
    return createSuccess(undefined);
  } catch (error) {
    return createFail(createErrorString(error));
  }
}

/**
 * Add compression policy for history table (compress chunks older than 1 hour)
 */
export async function addHistoryCompressionPolicy(db: Db): Promise<Result<void>> {
  try {
    log.info("Adding compression policy for history table...");

    await db.execute(sql`
      SELECT add_compression_policy('history', INTERVAL '1 hour')
    `);

    log.info("Compression policy added for history table");
    return createSuccess(undefined);
  } catch (error) {
    // Policy might already exist
    const errorStr = createErrorString(error);
    if (errorStr.includes("already exists")) {
      log.info("Compression policy already exists for history table");
      return createSuccess(undefined);
    }
    return createFail(errorStr);
  }
}

/**
 * Compress all eligible chunks for a hypertable.
 * Compresses any uncompressed chunk whose time range has fully elapsed
 * (range_end is in the past), plus the configured grace period.
 */
export async function compressEligibleChunks(
  db: Db,
  tableName: string,
  olderThan: string = "1 hour",
): Promise<Result<number>> {
  try {
    const result = await db.execute(sql.raw(`
      SELECT chunk_schema, chunk_name
      FROM timescaledb_information.chunks
      WHERE hypertable_name = '${tableName}'
        AND NOT is_compressed
        AND range_end < NOW() - INTERVAL '${olderThan}'
      ORDER BY range_start
    `));

    let compressed = 0;
    for (const row of result.rows) {
      const chunkFqn = `${row.chunk_schema}.${row.chunk_name}`;
      try {
        await db.execute(sql.raw(`SELECT compress_chunk('${chunkFqn}')`));
        compressed++;
        log.info(`Compressed chunk ${chunkFqn}`);
      } catch (error) {
        log.warn(`Failed to compress chunk ${chunkFqn}: ${createErrorString(error)}`);
      }
    }

    return createSuccess(compressed);
  } catch (error) {
    return createFail(createErrorString(error));
  }
}

/**
 * Initialize hypercore compression on all tables if available
 * This should be called on startup after migrations
 */
export async function initializeHypercore(db: Db): Promise<Result<void>> {
  log.info("Checking hypercore (TimescaleDB compression) availability...");

  const availableResult = await isHypercoreAvailable(db);
  if (!isSuccess(availableResult)) {
    log.warn(`Could not check hypercore availability: ${availableResult.error}`);
    return createFail(availableResult.error);
  }

  if (!availableResult.output) {
    log.info("Hypercore (TimescaleDB compression) is not available");
    return createSuccess(undefined);
  }

  log.info("Hypercore is available, checking compression status...");

  // Check and enable compression on history table
  const historyStatus = await getCompressionStatus(db, "history");
  if (isSuccess(historyStatus)) {
    if (!historyStatus.output.compressionEnabled) {
      const enableResult = await enableHistoryCompression(db);
      if (!isSuccess(enableResult)) {
        log.error(`Failed to enable compression on history: ${enableResult.error}`);
      }
    } else {
      log.info("Compression already enabled on history table");
    }

    if (!historyStatus.output.policyExists) {
      const policyResult = await addHistoryCompressionPolicy(db);
      if (!isSuccess(policyResult)) {
        log.error(`Failed to add compression policy for history: ${policyResult.error}`);
      }
    } else {
      log.info("Compression policy already exists for history table");
    }
  }

  // Directly compress any eligible chunks on startup.
  // The bgw scheduler policies are unreliable, so this ensures
  // compression actually happens on every restart/deploy.
  log.info("Compressing eligible chunks...");
  const historyCompressed = await compressEligibleChunks(db, "history", "1 hour");
  if (isSuccess(historyCompressed)) {
    log.info(`Compressed ${historyCompressed.output} history chunk(s)`);
  }
  log.info("Hypercore initialization complete");
  return createSuccess(undefined);
}

/**
 * Get the current retention policy for a hypertable, if any.
 */
export async function getRetentionPolicy(
  db: Db,
  tableName: string,
): Promise<Result<{ exists: boolean; intervalDays: number | null }>> {
  try {
    const result = await db.execute(sql.raw(`
      SELECT config->>'drop_after' as drop_after
      FROM timescaledb_information.jobs
      WHERE proc_name = 'policy_retention'
      AND hypertable_name = '${tableName}'
      LIMIT 1
    `));

    if (result.rows.length > 0) {
      const dropAfter = String(result.rows[0].drop_after);
      const daysMatch = dropAfter.match(/(\d+)\s*days?/i);
      return createSuccess({
        exists: true,
        intervalDays: daysMatch ? parseInt(daysMatch[1], 10) : null,
      });
    }

    return createSuccess({ exists: false, intervalDays: null });
  } catch (error) {
    return createFail(createErrorString(error));
  }
}

/**
 * Set the retention policy for a hypertable.
 * Removes any existing policy first, then adds one with the specified interval.
 */
export async function setRetentionPolicy(
  db: Db,
  tableName: string,
  days: number,
): Promise<Result<void>> {
  try {
    // Remove existing policy if any
    try {
      await db.execute(
        sql.raw(`SELECT remove_retention_policy('${tableName}')`),
      );
      log.info(`Removed existing retention policy for ${tableName}`);
    } catch {
      // No existing policy — that's fine
    }

    // Add new retention policy
    await db.execute(
      sql.raw(
        `SELECT add_retention_policy('${tableName}', INTERVAL '${days} days')`,
      ),
    );
    log.info(
      `Added retention policy for ${tableName}: drop data older than ${days} days`,
    );
    return createSuccess(undefined);
  } catch (error) {
    return createFail(createErrorString(error));
  }
}

/**
 * Initialize retention policy on the history table.
 * Should be called on startup regardless of historian status.
 */
export async function initializeRetention(db: Db): Promise<Result<void>> {
  log.info(`Checking retention policy (configured: ${retentionDays} days)...`);

  const availableResult = await isHypercoreAvailable(db);
  if (!isSuccess(availableResult)) {
    log.warn(
      `Could not check TimescaleDB availability: ${availableResult.error}`,
    );
    return createFail(availableResult.error);
  }

  if (!availableResult.output) {
    log.info("TimescaleDB not available, skipping retention policy");
    return createSuccess(undefined);
  }

  const currentPolicy = await getRetentionPolicy(db, "history");
  if (!isSuccess(currentPolicy)) {
    log.error(`Failed to get retention policy: ${currentPolicy.error}`);
    return createFail(currentPolicy.error);
  }

  const needsUpdate =
    !currentPolicy.output.exists ||
    currentPolicy.output.intervalDays !== retentionDays;

  if (needsUpdate) {
    const result = await setRetentionPolicy(db, "history", retentionDays);
    if (!isSuccess(result)) {
      log.error(`Failed to set retention policy: ${result.error}`);
      return createFail(result.error);
    }
  } else {
    log.info(
      `Retention policy already set to ${retentionDays} days, no change needed`,
    );
  }

  return createSuccess(undefined);
}

/**
 * Get storage statistics including compression info
 */
export async function getStorageStats(db: Db): Promise<Result<StorageStats>> {
  try {
    const availableResult = await isHypercoreAvailable(db);
    const hypercoreAvailable = isSuccess(availableResult) && availableResult.output;

    const tables: TableStorageStats[] = [];
    let compressionEnabled = false;

    if (hypercoreAvailable) {
      // Get hypertable info and sizes using TimescaleDB functions
      const hypertablesResult = await db.execute(sql`
        SELECT
          hypertable_name as table_name,
          compression_enabled
        FROM timescaledb_information.hypertables
        WHERE hypertable_name IN ('history', 'history_properties')
      `);

      for (const row of hypertablesResult.rows) {
        const tableName = String(row.table_name);

        // Get total size using hypertable_size function
        const sizeResult = await db.execute(sql.raw(`
          SELECT hypertable_size('${tableName}') as total_bytes
        `));
        const totalBytes = Number(sizeResult.rows[0]?.total_bytes) || 0;

        // Get compression stats using hypertable_compression_stats function
        let compressedBytes: number | null = null;
        let uncompressedBytes: number | null = null;
        let compressionRatio: number | null = null;

        if (row.compression_enabled) {
          compressionEnabled = true;

          const compressionResult = await db.execute(sql.raw(`
            SELECT
              COALESCE(SUM(after_compression_total_bytes), 0) as compressed_bytes,
              COALESCE(SUM(before_compression_total_bytes), 0) as uncompressed_bytes
            FROM hypertable_compression_stats('${tableName}')
          `));

          if (compressionResult.rows[0]) {
            compressedBytes = Number(compressionResult.rows[0].compressed_bytes) || 0;
            uncompressedBytes = Number(compressionResult.rows[0].uncompressed_bytes) || 0;

            if (compressedBytes > 0 && uncompressedBytes > 0) {
              compressionRatio = uncompressedBytes / compressedBytes;
            }
          }
        }

        tables.push({
          tableName,
          totalBytes,
          compressedBytes,
          uncompressedBytes,
          compressionRatio,
        });
      }
    } else {
      // Fall back to basic pg_total_relation_size for non-TimescaleDB
      const sizeResult = await db.execute(sql`
        SELECT
          'history' as table_name,
          pg_total_relation_size('history') as total_bytes
        UNION ALL
        SELECT
          'history_properties' as table_name,
          pg_total_relation_size('history_properties') as total_bytes
      `);

      for (const row of sizeResult.rows) {
        tables.push({
          tableName: String(row.table_name),
          totalBytes: Number(row.total_bytes) || 0,
          compressedBytes: null,
          uncompressedBytes: null,
          compressionRatio: null,
        });
      }
    }

    // Calculate totals
    const totalStorageBytes = tables.reduce((sum, t) => sum + t.totalBytes, 0);
    const totalCompressedBytes = tables.every(t => t.compressedBytes !== null)
      ? tables.reduce((sum, t) => sum + (t.compressedBytes ?? 0), 0)
      : null;
    const totalUncompressedBytes = tables.every(t => t.uncompressedBytes !== null)
      ? tables.reduce((sum, t) => sum + (t.uncompressedBytes ?? 0), 0)
      : null;

    let overallCompressionRatio: number | null = null;
    if (totalCompressedBytes !== null && totalUncompressedBytes !== null && totalCompressedBytes > 0) {
      overallCompressionRatio = totalUncompressedBytes / totalCompressedBytes;
    }

    return createSuccess({
      hypercoreAvailable,
      compressionEnabled,
      tables,
      totalStorageBytes,
      totalCompressedBytes,
      totalUncompressedBytes,
      overallCompressionRatio,
    });
  } catch (error) {
    return createFail(createErrorString(error));
  }
}

/**
 * Add hypercore/storage GraphQL schema
 */
export function addHypercoreToSchema(
  builder: ReturnType<typeof getBuilder>,
  db: Db,
) {
  // Table storage stats type
  const TableStorageStatsRef = builder.objectRef<TableStorageStats>("TableStorageStats");
  TableStorageStatsRef.implement({
    fields: (t) => ({
      tableName: t.exposeString("tableName"),
      totalBytes: t.exposeFloat("totalBytes"),
      compressedBytes: t.exposeFloat("compressedBytes", { nullable: true }),
      uncompressedBytes: t.exposeFloat("uncompressedBytes", { nullable: true }),
      compressionRatio: t.exposeFloat("compressionRatio", { nullable: true }),
    }),
  });

  // Storage stats type
  const StorageStatsRef = builder.objectRef<StorageStats>("StorageStats");
  StorageStatsRef.implement({
    fields: (t) => ({
      hypercoreAvailable: t.exposeBoolean("hypercoreAvailable"),
      compressionEnabled: t.exposeBoolean("compressionEnabled"),
      tables: t.field({
        type: [TableStorageStatsRef],
        resolve: (parent) => parent.tables,
      }),
      totalStorageBytes: t.exposeFloat("totalStorageBytes"),
      totalCompressedBytes: t.exposeFloat("totalCompressedBytes", { nullable: true }),
      totalUncompressedBytes: t.exposeFloat("totalUncompressedBytes", { nullable: true }),
      overallCompressionRatio: t.exposeFloat("overallCompressionRatio", { nullable: true }),
    }),
  });

  // Query: hypercoreAvailable
  builder.queryField("hypercoreAvailable", (t) =>
    t.field({
      type: "Boolean",
      description: "Check if TimescaleDB compression (hypercore) is available",
      resolve: async () => {
        const result = await isHypercoreAvailable(db);
        if (isSuccess(result)) {
          return result.output;
        } else {
          throw new GraphQLError(result.error);
        }
      },
    })
  );

  // Query: storageStats
  builder.queryField("storageStats", (t) =>
    t.field({
      type: StorageStatsRef,
      description: "Get storage statistics including compression info",
      resolve: async () => {
        if (!historianEnabled) {
          throw new GraphQLError(
            "Historian is not enabled for this space.",
          );
        }
        const result = await getStorageStats(db);
        if (isSuccess(result)) {
          return result.output;
        } else {
          throw new GraphQLError(result.error);
        }
      },
    })
  );

  // Retention policy type
  const RetentionPolicyRef = builder.objectRef<{
    exists: boolean;
    intervalDays: number | null;
    configuredDays: number;
  }>("RetentionPolicy");

  RetentionPolicyRef.implement({
    fields: (t) => ({
      exists: t.exposeBoolean("exists"),
      intervalDays: t.exposeInt("intervalDays", { nullable: true }),
      configuredDays: t.exposeInt("configuredDays"),
    }),
  });

  // Query: retentionPolicy
  builder.queryField("retentionPolicy", (t) =>
    t.field({
      type: RetentionPolicyRef,
      description: "Get the current data retention policy configuration",
      resolve: async () => {
        const result = await getRetentionPolicy(db, "history");
        if (isSuccess(result)) {
          return { ...result.output, configuredDays: retentionDays };
        }
        throw new GraphQLError(result.error);
      },
    })
  );
}