import * as rdc from "@distilled.cloud/cloudflare/r2-data-catalog";
import * as Effect from "effect/Effect";
import * as Predicate from "effect/Predicate";
import * as Redacted from "effect/Redacted";
import * as Schedule from "effect/Schedule";

import { Unowned } from "../../AdoptPolicy.ts";
import { isResolved } from "../../Diff.ts";
import * as Provider from "../../Provider.ts";
import { Resource } from "../../Resource.ts";
import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
import type { Providers } from "../Providers.ts";

const TypeId = "Cloudflare.R2.DataCatalog" as const;
type TypeId = typeof TypeId;

/**
 * Whether a maintenance job (compaction or snapshot expiration) runs.
 */
export type MaintenanceState = "enabled" | "disabled";

/**
 * Target output file size, in MB, that catalog compaction rewrites small
 * files into.
 */
export type TargetSizeMb = "64" | "128" | "256" | "512";

/**
 * Catalog-level compaction maintenance settings.
 */
export type Compaction = {
  /**
   * Whether compaction runs for tables in this catalog.
   */
  state?: MaintenanceState;
  /**
   * Target output file size, in MB.
   * @default "128"
   */
  targetSizeMb?: TargetSizeMb;
};

/**
 * Catalog-level snapshot expiration maintenance settings.
 */
export type SnapshotExpiration = {
  /**
   * Whether snapshot expiration runs for tables in this catalog.
   */
  state?: MaintenanceState;
  /**
   * Maximum age a snapshot may reach before it is expired, expressed as a
   * duration string (e.g. `"7d"`).
   */
  maxSnapshotAge?: string;
  /**
   * Minimum number of snapshots retained per table regardless of age.
   */
  minSnapshotsToKeep?: number;
};

export type DataCatalogProps = {
  /**
   * Name of the R2 bucket to enable the Iceberg data catalog on. The bucket
   * must already exist — pass `bucket.bucketName` from a `Cloudflare.R2.Bucket`
   * resource to order catalog-after-bucket. Changing the bucket replaces the
   * catalog (the old bucket's catalog is disabled; table data is untouched).
   */
  bucketName: string;
  /**
   * Compaction maintenance configuration. Only the fields you specify are
   * enforced; omitted fields keep Cloudflare's defaults.
   */
  compaction?: Compaction;
  /**
   * Snapshot expiration maintenance configuration. Only the fields you
   * specify are enforced; omitted fields keep Cloudflare's defaults.
   */
  snapshotExpiration?: SnapshotExpiration;
  /**
   * Cloudflare API token (with R2 read/write access) the catalog uses to
   * run maintenance jobs against the bucket. Write-only: Cloudflare exposes
   * only `credentialStatus: "present" | "absent"`, never the token itself.
   * Maintenance jobs stay pending until a credential is provided.
   */
  token?: Redacted.Redacted<string>;
};

export type DataCatalogAttributes = {
  /**
   * Unique identifier of the catalog (stable across disable/enable cycles).
   */
  catalogId: string;
  /**
   * Catalog (warehouse) name, generated by Cloudflare as
   * `{accountId}_{bucketName}`.
   */
  name: string;
  /**
   * Name of the R2 bucket backing the catalog.
   */
  bucketName: string;
  /**
   * The Cloudflare account the catalog belongs to.
   */
  accountId: string;
  /**
   * Catalog status. A reconciled catalog is always `active`.
   */
  status: "active" | "inactive";
  /**
   * Whether a maintenance credential is registered for this catalog.
   */
  credentialStatus: "present" | "absent" | (string & {});
  /**
   * Observed compaction maintenance configuration.
   */
  compaction:
    | {
        state: MaintenanceState;
        targetSizeMb: TargetSizeMb;
      }
    | undefined;
  /**
   * Observed snapshot expiration maintenance configuration.
   */
  snapshotExpiration:
    | {
        state: MaintenanceState;
        maxSnapshotAge: string;
        minSnapshotsToKeep: number;
      }
    | undefined;
  /**
   * Iceberg REST catalog URI for this warehouse — point PyIceberg, Spark, or
   * any Iceberg REST client at this endpoint.
   */
  catalogUri: string;
};

export type DataCatalog = Resource<
  TypeId,
  DataCatalogProps,
  DataCatalogAttributes,
  never,
  Providers
>;

/**
 * Apache Iceberg data catalog attached to a Cloudflare R2 bucket.
 *
 * R2 Data Catalog exposes an Iceberg REST catalog endpoint backed by an R2
 * bucket, so engines like Spark, PyIceberg, and DuckDB can create and query
 * Iceberg tables stored in R2. The catalog is a singleton per bucket: this
 * resource enables it, keeps its maintenance configuration in sync, and
 * disables it on destroy (table data in the bucket is never deleted).
 * ### Enabling a catalog
 * **Example:** Enable the catalog on an R2 bucket
 * ```typescript
 * const bucket = yield* Cloudflare.R2.Bucket("LakehouseBucket");
 *
 * const catalog = yield* Cloudflare.R2.R2DataCatalog("Lakehouse", {
 *   bucketName: bucket.bucketName,
 * });
 *
 * // Point any Iceberg REST client at the warehouse:
 * const uri = catalog.catalogUri;
 * const warehouse = catalog.name;
 * ```
 *
 * ### Maintenance
 * **Example:** Configure compaction and snapshot expiration
 * ```typescript
 * const catalog = yield* Cloudflare.R2.R2DataCatalog("Lakehouse", {
 *   bucketName: bucket.bucketName,
 *   compaction: { state: "enabled", targetSizeMb: "256" },
 *   snapshotExpiration: {
 *     state: "enabled",
 *     maxSnapshotAge: "3d",
 *     minSnapshotsToKeep: 5,
 *   },
 * });
 * ```
 *
 * **Example:** Register a maintenance credential
 * ```typescript
 * // Maintenance jobs need an API token with R2 read/write on the bucket.
 * const catalog = yield* Cloudflare.R2.R2DataCatalog("Lakehouse", {
 *   bucketName: bucket.bucketName,
 *   compaction: { state: "enabled" },
 *   token: maintenanceToken, // Redacted<string>
 * });
 * ```
 *
 * @see https://developers.cloudflare.com/r2/data-catalog/
 *
 * @resource
 * @product R2 Data Catalog
 * @category Storage & Databases
 */
export const DataCatalog = Resource<DataCatalog>(TypeId);

/**
 * Returns true if the given value is an R2DataCatalog resource.
 */
export const isDataCatalog = (value: unknown): value is DataCatalog =>
  Predicate.hasProperty(value, "Type") && value.Type === TypeId;

export const DataCatalogProvider = () =>
  Provider.succeed(DataCatalog, {
    stables: ["catalogId", "name", "accountId", "bucketName", "catalogUri"],
    diff: Effect.fn(function* ({ olds, news, output }) {
      const { accountId } = yield* yield* CloudflareEnvironment;
      if (!isResolved(news)) return undefined;
      if ((output?.accountId ?? accountId) !== accountId) {
        return { action: "replace" } as const;
      }
      // The catalog is keyed to its bucket — moving buckets is a replacement
      // (disable on the old bucket, enable on the new one). bucketName is an
      // Input<string>; compare only once both sides are concrete strings.
      const oldBucket =
        output?.bucketName ??
        (typeof olds?.bucketName === "string" ? olds.bucketName : undefined);
      if (
        oldBucket !== undefined &&
        typeof news.bucketName === "string" &&
        oldBucket !== news.bucketName
      ) {
        return { action: "replace" } as const;
      }
      return undefined;
    }),
    read: Effect.fn(function* ({ output, olds }) {
      const { accountId } = yield* yield* CloudflareEnvironment;
      const acct = output?.accountId ?? accountId;
      const bucketName =
        output?.bucketName ??
        (typeof olds?.bucketName === "string" ? olds.bucketName : undefined);
      if (bucketName === undefined) return undefined;

      const observed = yield* getCatalog(acct, bucketName);
      // `inactive` means the catalog was disabled — for reconciliation
      // purposes the resource is gone (re-enable on the next deploy).
      if (!observed || observed.status !== "active") return undefined;
      const attrs = toAttributes(observed, acct);
      // The catalog carries no ownership markers; if we have no state of our
      // own, an active catalog may have been enabled out-of-band. Brand it
      // `Unowned` so the engine gates takeover behind the adopt policy.
      return output ? attrs : Unowned(attrs);
    }),
    list: Effect.fn(function* () {
      const { accountId } = yield* yield* CloudflareEnvironment;
      // R2 Data Catalog is an account-scoped collection: one warehouse per
      // bucket that has the catalog enabled. Only `active` warehouses map to
      // a live resource — `read` treats `inactive` (disabled) as gone.
      return yield* rdc.listR2DataCatalogs({ accountId }).pipe(
        Effect.map(({ warehouses }) =>
          warehouses
            .filter((w) => w.status === "active")
            .map((w) => toAttributes(w, accountId)),
        ),
        // Accounts without R2 Data Catalog access reject the route entirely.
        Effect.catchTag("InvalidRoute", () => Effect.succeed([])),
      );
    }),
    reconcile: Effect.fn(function* ({ news, olds, output }) {
      const { accountId } = yield* yield* CloudflareEnvironment;
      const acct = output?.accountId ?? accountId;
      // Inputs have been resolved to concrete strings by Plan.
      const bucketName = news.bucketName as string;

      // Observe — `output` is only a cache; a disabled or never-enabled
      // catalog observes as missing/inactive and falls through to ensure.
      let observed = yield* getCatalog(acct, bucketName);

      // Ensure — enable is a true idempotent upsert (re-running it on an
      // active catalog returns the same id), so no AlreadyExists race exists.
      // A freshly-created bucket can briefly 404 the enable endpoint
      // (`NoSuchBucket`, code 40406) — ride out that consistency lag.
      if (!observed || observed.status !== "active") {
        yield* rdc.enableR2DataCatalog({ accountId: acct, bucketName }).pipe(
          Effect.retry({
            while: (e) => e._tag === "NoSuchBucket",
            schedule: catalogConsistencySchedule,
          }),
        );
        observed = yield* rdc
          .getR2DataCatalog({ accountId: acct, bucketName })
          .pipe(
            // The catalog can lag behind its own enable call.
            Effect.retry({
              while: (e) => e._tag === "WarehouseNotFound",
              schedule: catalogConsistencySchedule,
            }),
          );
      }

      // Sync maintenance — diff observed config against the fields the user
      // actually specified; skip the API entirely on a no-op.
      let maintenance = observed.maintenanceConfig ?? undefined;
      const observedCompaction = maintenance?.compaction ?? undefined;
      const observedExpiration = maintenance?.snapshotExpiration ?? undefined;
      const compactionDirty =
        news.compaction !== undefined &&
        ((news.compaction.state !== undefined &&
          news.compaction.state !== observedCompaction?.state) ||
          (news.compaction.targetSizeMb !== undefined &&
            news.compaction.targetSizeMb !== observedCompaction?.targetSizeMb));
      const expirationDirty =
        news.snapshotExpiration !== undefined &&
        ((news.snapshotExpiration.state !== undefined &&
          news.snapshotExpiration.state !== observedExpiration?.state) ||
          (news.snapshotExpiration.maxSnapshotAge !== undefined &&
            news.snapshotExpiration.maxSnapshotAge !==
              observedExpiration?.maxSnapshotAge) ||
          (news.snapshotExpiration.minSnapshotsToKeep !== undefined &&
            news.snapshotExpiration.minSnapshotsToKeep !==
              observedExpiration?.minSnapshotsToKeep));
      if (compactionDirty || expirationDirty) {
        const updated = yield* rdc.updateMaintenanceConfig({
          accountId: acct,
          bucketName,
          compaction: news.compaction,
          snapshotExpiration: news.snapshotExpiration,
        });
        maintenance = {
          compaction: updated.compaction,
          snapshotExpiration: updated.snapshotExpiration,
        };
      }

      // Sync credential — the API exposes only present/absent, so `olds`
      // serves as the rotation hint: re-push when the token value changed
      // or no credential is registered (adoption re-pushes; idempotent).
      let credentialStatus = (observed.credentialStatus ?? "absent") as
        | "present"
        | "absent";
      if (news.token !== undefined) {
        const rotated =
          olds?.token === undefined ||
          Redacted.value(olds.token) !== Redacted.value(news.token);
        if (credentialStatus !== "present" || rotated) {
          yield* rdc.createCredential({
            accountId: acct,
            bucketName,
            token: Redacted.value(news.token),
          });
          credentialStatus = "present";
        }
      }

      return toAttributes(
        { ...observed, status: "active", maintenanceConfig: maintenance },
        acct,
        credentialStatus,
      );
    }),
    delete: Effect.fn(function* ({ output }) {
      // Disable is idempotent server-side; a never-enabled (or already
      // disabled-and-bucket-deleted) catalog surfaces `WarehouseNotFound`.
      yield* rdc
        .disableR2DataCatalog({
          accountId: output.accountId,
          bucketName: output.bucketName,
        })
        .pipe(
          Effect.catchTag(
            ["WarehouseNotFound", "NoSuchBucket"],
            () => Effect.void,
          ),
        );
    }),
  });

// A freshly-created bucket (or freshly-enabled catalog) can briefly 404 on
// the r2-catalog endpoints before the warehouse propagates.
const catalogConsistencySchedule = Schedule.max([
  Schedule.exponential(100),
  Schedule.recurs(5),
]);

/**
 * Read a catalog by bucket name, mapping "gone" (`WarehouseNotFound`, code
 * 40401, or a missing bucket) to `undefined`.
 */
const getCatalog = (accountId: string, bucketName: string) =>
  rdc
    .getR2DataCatalog({ accountId, bucketName })
    .pipe(
      Effect.catchTag(["WarehouseNotFound", "NoSuchBucket"], () =>
        Effect.succeed(undefined),
      ),
    );

const toAttributes = (
  catalog: rdc.GetR2DataCatalogResponse,
  accountId: string,
  credentialStatus?: "present" | "absent",
): DataCatalogAttributes => ({
  catalogId: catalog.id,
  name: catalog.name,
  bucketName: catalog.bucket,
  accountId,
  // Distilled widens generated string enums to open unions (`string & {}`).
  status: catalog.status as "active" | "inactive",
  credentialStatus: credentialStatus ?? catalog.credentialStatus ?? "absent",
  compaction: catalog.maintenanceConfig?.compaction
    ? {
        state: catalog.maintenanceConfig.compaction.state as MaintenanceState,
        targetSizeMb: catalog.maintenanceConfig.compaction
          .targetSizeMb as TargetSizeMb,
      }
    : undefined,
  snapshotExpiration: catalog.maintenanceConfig?.snapshotExpiration
    ? {
        state: catalog.maintenanceConfig.snapshotExpiration
          .state as MaintenanceState,
        maxSnapshotAge:
          catalog.maintenanceConfig.snapshotExpiration.maxSnapshotAge,
        minSnapshotsToKeep:
          catalog.maintenanceConfig.snapshotExpiration.minSnapshotsToKeep,
      }
    : undefined,
  catalogUri: `https://catalog.cloudflarestorage.com/${accountId}/${catalog.bucket}`,
});