/**
 * Connection Storage Implementation
 *
 * Handles CRUD operations for MCP connections using Kysely (database-agnostic).
 * All connections are organization-scoped.
 */

import {
  type Insertable,
  type Kysely,
  type RawBuilder,
  sql,
  type SqlBool,
  type Updateable,
} from "kysely";
import type {
  OrderByExpression,
  WhereExpression,
} from "@decocms/bindings/collections";
import type { CredentialVault } from "../encryption/credential-vault";
import type {
  ConnectionEntity,
  ConnectionParameters,
  OAuthConfig,
  StdioConnectionParameters,
} from "../tools/connection/schema";
import { isStdioParameters } from "../tools/connection/schema";
import {
  createNoRedirectFetch,
  guardAgainstPrivateUrl,
} from "../tools/registry/discover-tools";
import { generatePrefixedId } from "@decocms/shared/utils/generate-id";
import {
  getWellKnownDecopilotConnection,
  isDecopilot,
} from "@decocms/shared/sdk";
import { getConnectionSlug } from "@decocms/shared/utils/connection-slug";
import { CONNECTION_DECRYPT_DISABLE_THRESHOLD } from "../core/constants";
import {
  isDecryptDisabled,
  markDecryptDisabled,
  recordDecryptFailure,
  recordDecryptSuccess,
} from "./decrypt-failure-tracker";
import type { ConnectionStoragePort } from "./ports";
import type { Database } from "./types";

/** JSON fields that need serialization/deserialization */
const JSON_FIELDS = [
  "connection_headers",
  "oauth_config",
  "configuration_scopes",
  "metadata",
  "bindings",
] as const;

/** Raw database row type */
type RawConnectionRow = {
  id: string;
  organization_id: string;
  created_by: string;
  updated_by: string | null;
  title: string;
  description: string | null;
  icon: string | null;
  app_name: string | null;
  app_id: string | null;
  slug: string | null;
  connection_type: "HTTP" | "SSE" | "Websocket" | "STDIO" | "VIRTUAL";
  connection_url: string | null;
  connection_token: string | null;
  connection_headers: string | null; // JSON, envVars encrypted for STDIO
  oauth_config: string | OAuthConfig | null;
  configuration_state: string | null; // Encrypted
  configuration_scopes: string | string[] | null;
  metadata: string | Record<string, unknown> | null;
  bindings: string | string[] | null;
  status: "active" | "inactive" | "error";
  created_at: Date | string;
  updated_at: Date | string;
};
/** Top-level columns on the connections table that are safe for user-controlled WHERE filtering */
const TOP_LEVEL_COLUMNS = new Set([
  "id",
  "organization_id",
  "created_by",
  "updated_by",
  "title",
  "description",
  "icon",
  "app_name",
  "app_id",
  "slug",
  "connection_type",
  "connection_url",
  // connection_token is intentionally excluded — sensitive
  "status",
  "created_at",
  "updated_at",
]);

/** JSON columns that support nested access via ->>. Excludes sensitive columns. */
const JSON_COLUMNS = new Set([
  "metadata",
  // connection_headers excluded — may contain auth headers
  // oauth_config excluded — contains client secrets and tokens
  "configuration_scopes",
  "bindings",
]);

/**
 * Build a SQL reference for a field path.
 * Single-segment paths that match a column become `"column"`.
 * Multi-segment paths where the first segment is a JSON column become `"column"->>'key'`.
 * Returns null for unsupported paths.
 */
function fieldRef(fieldPath: string[]): RawBuilder<unknown> | null {
  if (fieldPath.length === 0) return null;

  const column = fieldPath[0]!;

  if (fieldPath.length === 1 && TOP_LEVEL_COLUMNS.has(column)) {
    return sql.ref(column);
  }

  if (fieldPath.length === 2 && JSON_COLUMNS.has(column)) {
    const key = fieldPath[1]!;
    return sql`${sql.ref(column)}->>${sql.lit(key)}`;
  }

  // Deeper nesting: use #>> for Postgres JSON path
  if (fieldPath.length > 2 && JSON_COLUMNS.has(column)) {
    const path = `{${fieldPath.slice(1).join(",")}}`;
    return sql`${sql.ref(column)}#>>${sql.lit(path)}`;
  }

  // Unknown column — skip (will be ignored)
  return null;
}

/**
 * Translate a WhereExpression tree into a Kysely SQL expression.
 */
function applyWhereToSql(where: WhereExpression): RawBuilder<SqlBool> {
  if ("conditions" in where) {
    const { operator, conditions } = where;
    if (conditions.length === 0) return sql<SqlBool>`true`;

    const parts = conditions.map((c) => applyWhereToSql(c));

    switch (operator) {
      case "and":
        return sql<SqlBool>`(${sql.join(parts, sql` AND `)})`;
      case "or":
        return sql<SqlBool>`(${sql.join(parts, sql` OR `)})`;
      case "not":
        return sql<SqlBool>`NOT (${sql.join(parts, sql` AND `)})`;
      default:
        return sql<SqlBool>`true`;
    }
  }

  const { field, operator, value } = where;
  const ref = fieldRef(field);
  if (!ref) return sql<SqlBool>`true`; // Unknown field — no-op

  switch (operator) {
    case "eq":
      return value === null
        ? sql<SqlBool>`${ref} IS NULL`
        : sql<SqlBool>`${ref} = ${sql.val(value)}`;
    case "gt":
      return sql<SqlBool>`${ref} > ${sql.val(value)}`;
    case "gte":
      return sql<SqlBool>`${ref} >= ${sql.val(value)}`;
    case "lt":
      return sql<SqlBool>`${ref} < ${sql.val(value)}`;
    case "lte":
      return sql<SqlBool>`${ref} <= ${sql.val(value)}`;
    case "in":
      if (!Array.isArray(value) || value.length === 0)
        return sql<SqlBool>`false`;
      return sql<SqlBool>`${ref} IN (${sql.join(value.map((v) => sql.val(v)))})`;
    case "like":
      return sql<SqlBool>`${ref} ILIKE ${sql.val(value)}`;
    case "contains": {
      // Escape LIKE metacharacters so they match literally
      const escaped = String(value).replace(/[%_\\]/g, "\\$&");
      return sql<SqlBool>`${ref} ILIKE ${sql.val(`%${escaped}%`)}`;
    }
    default:
      return sql<SqlBool>`true`;
  }
}

export class ConnectionStorage implements ConnectionStoragePort {
  constructor(
    private db: Kysely<Database>,
    private vault: CredentialVault,
  ) {}

  async create(data: Partial<ConnectionEntity>): Promise<ConnectionEntity> {
    const id = data.id ?? generatePrefixedId("conn");
    const now = new Date().toISOString();

    const existing = await this.findById(id);

    if (existing) {
      // Only allow update if same organization - prevent cross-org hijacking
      if (existing.organization_id !== data.organization_id) {
        throw new Error("Connection ID already exists");
      }
      return this.update(id, data);
    }

    const slug = getConnectionSlug(data);
    const serialized = await this.serializeConnection({
      ...data,
      id: data.id ?? id,
      slug,
      status: "active",
      created_at: now,
      updated_at: now,
    });
    await this.db
      .insertInto("connections")
      .values(serialized as Insertable<Database["connections"]>)
      .execute();

    const connection = await this.findById(id);
    if (!connection) {
      throw new Error(`Failed to create connection with id: ${id}`);
    }

    return connection;
  }

  async createNew(data: Partial<ConnectionEntity>): Promise<ConnectionEntity> {
    const id = data.id ?? generatePrefixedId("conn");
    const now = new Date().toISOString();
    const slug = getConnectionSlug(data);
    const serialized = await this.serializeConnection({
      ...data,
      id,
      slug,
      status: "active",
      created_at: now,
      updated_at: now,
    });

    try {
      await this.db
        .insertInto("connections")
        .values(serialized as Insertable<Database["connections"]>)
        .execute();
    } catch (error) {
      if (
        error instanceof Error &&
        (error.message.includes("duplicate key") ||
          error.message.includes("unique constraint"))
      ) {
        throw new Error("Connection ID already exists");
      }
      throw error;
    }

    const connection = await this.findById(id);
    if (!connection) {
      throw new Error(`Failed to create connection with id: ${id}`);
    }

    return connection;
  }

  async findById(
    id: string,
    organizationId?: string,
  ): Promise<ConnectionEntity | null> {
    // Handle Decopilot ID - return Decopilot connection entity
    const decopilotOrgId = isDecopilot(id);
    if (decopilotOrgId) {
      const resolvedOrgId = organizationId ?? decopilotOrgId;
      return getWellKnownDecopilotConnection(resolvedOrgId);
    }

    let query = this.db
      .selectFrom("connections")
      .selectAll()
      .where("id", "=", id);

    if (organizationId) {
      query = query.where("organization_id", "=", organizationId);
    }

    const row = await query.executeTakeFirst();
    return row ? this.deserializeConnection(row as RawConnectionRow) : null;
  }

  async list(
    organizationId: string,
    options?: {
      includeVirtual?: boolean;
      slug?: string;
      where?: WhereExpression;
      orderBy?: OrderByExpression[];
      limit?: number;
      offset?: number;
    },
  ): Promise<{ items: ConnectionEntity[]; totalCount: number }> {
    let query = this.db
      .selectFrom("connections")
      .selectAll()
      .where("organization_id", "=", organizationId);

    // By default, exclude VIRTUAL connections unless explicitly requested
    if (!options?.includeVirtual) {
      query = query.where("connection_type", "!=", "VIRTUAL");
    }

    if (options?.slug) {
      query = query.where("slug", "=", options.slug);
    }

    // Apply where expression to SQL
    if (options?.where) {
      query = query.where(applyWhereToSql(options.where));
    }

    // Count before pagination
    const countQuery = this.db
      .selectFrom(query.as("filtered"))
      .select(sql<number>`count(*)::int`.as("count"));
    const countResult = await countQuery.executeTakeFirst();
    const totalCount = countResult?.count ?? 0;

    // Apply orderBy
    if (options?.orderBy && options.orderBy.length > 0) {
      for (const order of options.orderBy) {
        const ref = fieldRef(order.field);
        if (!ref) continue;
        const dir = order.direction === "desc" ? sql`desc` : sql`asc`;
        const nulls =
          order.nulls === "first" ? sql`nulls first` : sql`nulls last`;
        query = query.orderBy(sql`${ref} ${dir} ${nulls}`);
      }
    }

    // Apply pagination
    if (options?.limit) {
      query = query.limit(options.limit);
    }
    if (options?.offset) {
      query = query.offset(options.offset);
    }

    const rows = await query.execute();

    const items = await Promise.all(
      rows.map((row) => this.deserializeConnection(row as RawConnectionRow)),
    );

    return { items, totalCount };
  }

  async update(
    id: string,
    data: Partial<ConnectionEntity>,
  ): Promise<ConnectionEntity> {
    if (Object.keys(data).length === 0) {
      const connection = await this.findById(id);
      if (!connection) throw new Error("Connection not found");
      return connection;
    }

    // The slug column is a denormalization of app_name/connection_url/title, and
    // the UI links to a connection by the slug it derives from those same fields
    // (getConnectionSlug). Any drift makes the connection unreachable — its detail
    // page 404s while the list still shows it. So recompute from the merged row on
    // every update, using presence (not `??`) so an explicit null clears the field
    // it was derived from instead of silently keeping the old slug.
    const slugData: Record<string, unknown> = { ...data };
    const existing = await this.findById(id);
    if (existing) {
      const merged = <K extends keyof ConnectionEntity>(key: K) =>
        data[key] === undefined ? existing[key] : data[key];
      slugData.slug = getConnectionSlug({
        app_name: merged("app_name"),
        connection_url: merged("connection_url"),
        title: merged("title"),
        id,
      });
    }

    const serialized = await this.serializeConnection({
      ...slugData,
      updated_at: new Date().toISOString(),
    });

    await this.db
      .updateTable("connections")
      .set(serialized)
      .where("id", "=", id)
      .execute();

    const connection = await this.findById(id);
    if (!connection) {
      throw new Error("Connection not found after update");
    }

    return connection;
  }

  async delete(id: string): Promise<void> {
    await this.db.deleteFrom("connections").where("id", "=", id).execute();
  }

  async testConnection(
    id: string,
    headers?: Record<string, string>,
  ): Promise<{ healthy: boolean; latencyMs: number }> {
    const connection = await this.findById(id);
    if (!connection) {
      throw new Error("Connection not found");
    }

    const startTime = Date.now();

    // STDIO connections can't be tested via HTTP
    if (connection.connection_type === "STDIO") {
      // For STDIO, we'd need to spawn the process - skip for now
      return {
        healthy: true, // Assume healthy, actual health checked on first use
        latencyMs: Date.now() - startTime,
      };
    }

    if (!connection.connection_url) {
      return {
        healthy: false,
        latencyMs: Date.now() - startTime,
      };
    }

    try {
      // connection_url is user-supplied — a plain `fetch` here would let a
      // URL that resolves privately (or 302s to one) reach an internal/
      // metadata address on every test call, even if the address was safe at
      // create/update time. Reuse the same SSRF guard the fetch-tools path
      // applies, plus a no-redirect fetch to close the redirect-based bypass.
      const guardError = await guardAgainstPrivateUrl(
        connection.connection_url,
      );
      if (guardError) {
        return {
          healthy: false,
          latencyMs: Date.now() - startTime,
        };
      }

      const httpParams = connection.connection_headers as {
        headers?: Record<string, string>;
      } | null;

      const response = await createNoRedirectFetch()(
        connection.connection_url,
        {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
            ...(connection.connection_token && {
              Authorization: `Bearer ${connection.connection_token}`,
            }),
            ...httpParams?.headers,
            ...headers,
          },
          body: JSON.stringify({
            jsonrpc: "2.0",
            method: "ping",
            id: 1,
          }),
        },
      );

      return {
        healthy: response.ok || response.status === 404,
        latencyMs: Date.now() - startTime,
      };
    } catch {
      return {
        healthy: false,
        latencyMs: Date.now() - startTime,
      };
    }
  }

  /**
   * Serialize entity data to database format
   */
  private async serializeConnection(
    data: Partial<ConnectionEntity>,
  ): Promise<Updateable<Database["connections"]>> {
    const result: Record<string, unknown> = {};

    for (const [key, value] of Object.entries(data)) {
      if (value === undefined) continue;
      // tools column was dropped — skip to avoid inserting into non-existent column
      if (key === "tools") continue;

      if (key === "connection_token" && value) {
        result[key] = await this.vault.encrypt(value as string);
      } else if (key === "configuration_state" && value) {
        // Encrypt configuration state
        const stateJson = JSON.stringify(value);
        result[key] = await this.vault.encrypt(stateJson);
      } else if (key === "connection_headers" && value) {
        // For STDIO, encrypt envVars before storing
        const params = value as ConnectionParameters;
        if (isStdioParameters(params) && params.envVars) {
          const encryptedEnvVars: Record<string, string> = {};
          for (const [envKey, envValue] of Object.entries(params.envVars)) {
            encryptedEnvVars[envKey] = await this.vault.encrypt(envValue);
          }
          result[key] = JSON.stringify({
            ...params,
            envVars: encryptedEnvVars,
          });
        } else {
          result[key] = JSON.stringify(params);
        }
      } else if (JSON_FIELDS.includes(key as (typeof JSON_FIELDS)[number])) {
        result[key] = value ? JSON.stringify(value) : null;
      } else {
        result[key] = value;
      }
    }

    return result as Updateable<Database["connections"]>;
  }

  /**
   * Handle one or more credential-decryption failures for a connection.
   *
   * Logs each failure until CONNECTION_DECRYPT_DISABLE_THRESHOLD consecutive
   * failures are reached, then durably disables the connection (status="error")
   * and suppresses further logs. Already-disabled connections (this replica or a
   * non-active DB status from a prior disable) are suppressed immediately.
   */
  private async handleDecryptFailures(
    row: RawConnectionRow,
    failures: Array<{ label: string; error: unknown }>,
  ): Promise<void> {
    const context = `connection ${row.id} (org: ${row.organization_id}, type: ${row.connection_type}, title: ${row.title})`;

    // Already disabled in this replica — stay quiet.
    if (isDecryptDisabled(row.id)) return;

    // Disabled in a prior run (or paused by a user): sync local state and suppress.
    if (row.status !== "active") {
      markDecryptDisabled(row.id);
      return;
    }

    const { thresholdCrossed } = recordDecryptFailure(row.id);

    if (!thresholdCrossed) {
      for (const { label, error } of failures) {
        console.error(`Failed to decrypt ${label} for ${context}:`, error);
      }
      return;
    }

    // Threshold reached — disable durably, then mark suppressed only on success
    // so a failed write is retried on the next read instead of silently dropped.
    try {
      await this.db
        .updateTable("connections")
        .set({ status: "error" })
        .where("id", "=", row.id)
        .execute();
      markDecryptDisabled(row.id);
      console.error(
        `Disabled ${context} after ${CONNECTION_DECRYPT_DISABLE_THRESHOLD} consecutive decryption failures.`,
        failures[0]?.error,
      );
    } catch (err) {
      console.error(
        `Failed to disable ${context} after decryption failures:`,
        err,
      );
    }
  }

  /**
   * Deserialize database row to entity
   */
  private async deserializeConnection(
    row: RawConnectionRow,
  ): Promise<ConnectionEntity> {
    const decryptErrors: Array<{ label: string; error: unknown }> = [];

    let decryptedToken: string | null = null;
    if (row.connection_token) {
      try {
        decryptedToken = await this.vault.decrypt(row.connection_token);
      } catch (error) {
        decryptErrors.push({ label: "connection token", error });
      }
    }

    // Decrypt configuration state
    let decryptedConfigState: Record<string, unknown> | null = null;
    if (row.configuration_state) {
      try {
        const decryptedJson = await this.vault.decrypt(row.configuration_state);
        decryptedConfigState = JSON.parse(decryptedJson);
      } catch (error) {
        decryptErrors.push({ label: "configuration state", error });
      }
    }

    if (decryptErrors.length > 0) {
      await this.handleDecryptFailures(row, decryptErrors);
    } else if (row.connection_token || row.configuration_state) {
      recordDecryptSuccess(row.id);
    }

    // Parse and decrypt connection_headers
    let connectionParameters: ConnectionParameters | null = null;
    if (row.connection_headers) {
      try {
        const parsed = JSON.parse(row.connection_headers);
        // For STDIO, decrypt envVars
        if (isStdioParameters(parsed) && parsed.envVars) {
          const decryptedEnvVars: Record<string, string> = {};
          for (const [envKey, envValue] of Object.entries(parsed.envVars)) {
            try {
              decryptedEnvVars[envKey] = await this.vault.decrypt(
                envValue as string,
              );
            } catch {
              // If decryption fails, keep encrypted value (migration case)
              decryptedEnvVars[envKey] = envValue as string;
            }
          }
          connectionParameters = {
            ...parsed,
            envVars: decryptedEnvVars,
          } as StdioConnectionParameters;
        } else {
          connectionParameters = parsed;
        }
      } catch (error) {
        console.error(
          `Failed to parse connection_headers for connection ${row.id} (org: ${row.organization_id}, type: ${row.connection_type}, title: ${row.title}):`,
          error,
        );
      }
    }

    const parseJson = <T>(value: string | T | null): T | null => {
      if (value === null) return null;
      if (typeof value === "string") {
        try {
          return JSON.parse(value) as T;
        } catch {
          return null;
        }
      }
      return value as T;
    };

    return {
      id: row.id,
      organization_id: row.organization_id,
      created_by: row.created_by,
      updated_by: row.updated_by ?? undefined,
      title: row.title,
      description: row.description,
      icon: row.icon,
      app_name: row.app_name,
      app_id: row.app_id,
      slug: row.slug,
      connection_type: row.connection_type,
      connection_url: row.connection_url,
      connection_token: decryptedToken,
      connection_headers: connectionParameters,
      oauth_config: parseJson<OAuthConfig>(row.oauth_config),
      configuration_state: decryptedConfigState,
      configuration_scopes: parseJson<string[]>(row.configuration_scopes),
      metadata: parseJson<Record<string, unknown>>(row.metadata),
      tools: null,
      bindings: parseJson<string[]>(row.bindings),
      status: row.status,
      created_at: row.created_at as string,
      updated_at: row.updated_at as string,
    };
  }
}
