import {
  AIRTABLE_API_ORIGIN,
  getAirtableApi,
  type AirtableApi,
} from "@automate.ax/integration-contracts/airtable"
import * as z from "zod"
import { defineAction } from "../../automation/actions"
import { airtableAccountRequirement } from "./lib"

const AIRTABLE_BATCH_SIZE = 10
const AIRTABLE_ID_SCHEMA = z.string().trim().min(1)
const AIRTABLE_RECORD_READ_REQUIREMENT =
  airtableAccountRequirement("data.records:read")
const AIRTABLE_RECORD_WRITE_REQUIREMENT =
  airtableAccountRequirement("data.records:write")
const AIRTABLE_FIELDS_SCHEMA = z.record(z.string(), z.json())
const AIRTABLE_RECORD_RESPONSE_SCHEMA = z.object({
  createdTime: z.iso.datetime(),
  fields: AIRTABLE_FIELDS_SCHEMA,
  id: z.string(),
})
const AIRTABLE_RECORDS_RESPONSE_SCHEMA = z.object({
  offset: z.string().optional(),
  records: AIRTABLE_RECORD_RESPONSE_SCHEMA.array(),
})
const AIRTABLE_RECORD_SCHEMA = z.object({
  /** Time at which Airtable created the record. */
  createdTime: z.date(),

  /** Field values keyed by field name or ID. */
  fields: AIRTABLE_FIELDS_SCHEMA,

  /** Stable Airtable record ID. */
  id: z.string(),
})
const AIRTABLE_SORT_SCHEMA = z.object({
  /** Sort direction for this field. */
  direction: z.enum(["asc", "desc"]).optional(),

  /** Field name or ID to sort by. */
  field: z.string().min(1),
})
const AIRTABLE_TABLE_SCHEMA = z.object({
  /** Optional table description. */
  description: z.string().optional(),

  /** Fields belonging to the table. */
  fields: z
    .object({
      /** Optional field description. */
      description: z.string().optional(),

      /** Stable Airtable field ID. */
      id: z.string(),

      /** Human-readable field name. */
      name: z.string(),

      /** Provider-specific field configuration. */
      options: z.record(z.string(), z.json()).optional(),

      /** Airtable field type. */
      type: z.string(),
    })
    .array(),

  /** Stable Airtable table ID. */
  id: z.string(),

  /** Human-readable table name. */
  name: z.string(),

  /** ID of the table's primary field. */
  primaryFieldId: z.string(),

  /** Views configured for the table. */
  views: z
    .object({
      /** Stable Airtable view ID. */
      id: z.string(),

      /** Human-readable view name. */
      name: z.string(),

      /** Airtable view type. */
      type: z.string(),
    })
    .array(),
})
const AIRTABLE_TABLE_RESPONSE_SCHEMA = AIRTABLE_TABLE_SCHEMA.extend({
  description: z.string().nullable().optional(),
  fields: AIRTABLE_TABLE_SCHEMA.shape.fields.element
    .extend({
      description: z.string().nullable().optional(),
    })
    .array(),
})
const AIRTABLE_WRITE_INPUT_SCHEMA = z.object({
  /** Stable Airtable base ID. */
  baseId: AIRTABLE_ID_SCHEMA,

  /** Return record fields keyed by stable field ID instead of field name. */
  returnFieldsByFieldId: z.boolean().optional(),

  /** Airtable table name or ID. */
  table: z.string().min(1),

  /** Ask Airtable to coerce string values to the destination field type. */
  typecast: z.boolean().optional(),
})
const AIRTABLE_BASES_RESPONSE_SCHEMA = z.object({
  bases: z
    .object({
      id: z.string(),
      name: z.string(),
      permissionLevel: z.string(),
    })
    .array(),
  offset: z.string().optional(),
})
const AIRTABLE_BASE_SCHEMA_RESPONSE_SCHEMA = z.object({
  tables: AIRTABLE_TABLE_RESPONSE_SCHEMA.array(),
})
const AIRTABLE_DELETED_RECORD_SCHEMA = z.object({
  deleted: z.literal(true),
  id: z.string(),
})

/** Lists every Airtable base accessible to the selected account. */
export const listAirtableBases = defineAction("List Airtable bases")
  .describe("Lists every Airtable base accessible to the connected account.")
  .account("airtable", airtableAccountRequirement("schema.bases:read"))
  .output(
    z.object({
      /** Accessible Airtable bases. */
      bases: z
        .object({
          /** Stable Airtable base ID. */
          id: z.string(),

          /** Human-readable base name. */
          name: z.string(),

          /** Effective permission level for the connected account. */
          permissionLevel: z.string(),
        })
        .array(),
    }),
  )
  .retry({ replaySafety: "safe" })
  .handler(async ({ account }) => {
    return { bases: await listAllAirtableBases(getAirtableApi(account.secret)) }
  })

/** Returns the complete table and view schema for an Airtable base. */
export const getAirtableBaseSchema = defineAction("Get Airtable base schema")
  .describe("Returns the tables, fields, and views configured for a base.")
  .account("airtable", airtableAccountRequirement("schema.bases:read"))
  .input(
    z.object({
      /** Stable Airtable base ID. */
      baseId: AIRTABLE_ID_SCHEMA,
    }),
  )
  .output(
    z.object({
      /** Tables configured in the base. */
      tables: AIRTABLE_TABLE_SCHEMA.array(),
    }),
  )
  .retry({ replaySafety: "safe" })
  .handler(async ({ account, input }) => {
    return {
      tables: AIRTABLE_BASE_SCHEMA_RESPONSE_SCHEMA.parse(
        await (
          await getAirtableApi(account.secret).request(
            `meta/bases/${encodeURIComponent(input.baseId)}/tables`,
          )
        ).json(),
      ).tables.map((table) => ({
        ...table,
        description: table.description ?? undefined,
        fields: table.fields.map((field) => ({
          ...field,
          description: field.description ?? undefined,
        })),
      })),
    }
  })

/** Gets one Airtable record by its stable record ID. */
export const getAirtableRecord = defineAction("Get Airtable record")
  .describe("Gets one record and all visible field values.")
  .account("airtable", AIRTABLE_RECORD_READ_REQUIREMENT)
  .input(
    z.object({
      /** Stable Airtable base ID. */
      baseId: AIRTABLE_ID_SCHEMA,

      /** Stable Airtable record ID. */
      recordId: z.string().min(1),

      /** Return fields keyed by stable field ID instead of field name. */
      returnFieldsByFieldId: z.boolean().optional(),

      /** Airtable table name or ID. */
      table: z.string().min(1),
    }),
  )
  .output(AIRTABLE_RECORD_SCHEMA)
  .retry({ replaySafety: "safe" })
  .handler(async ({ account, input }) => {
    const url = getRecordUrl(input.baseId, input.table, input.recordId)
    setBooleanParam(url, "returnFieldsByFieldId", input.returnFieldsByFieldId)
    return toAirtableRecord(
      AIRTABLE_RECORD_RESPONSE_SCHEMA.parse(
        await (
          await getAirtableApi(account.secret).request(toApiPath(url))
        ).json(),
      ),
    )
  })

/** Lists Airtable records with provider-native filtering, views, and sorting. */
export const listAirtableRecords = defineAction("List Airtable records")
  .describe(
    "Lists records with automatic pagination and Airtable-native filtering.",
  )
  .account("airtable", AIRTABLE_RECORD_READ_REQUIREMENT)
  .input(
    z.object({
      /** Stable Airtable base ID. */
      baseId: AIRTABLE_ID_SCHEMA,

      /** Field names or IDs to include in each record. */
      fields: z.string().min(1).array().min(1).optional(),

      /** Airtable formula used to filter records. */
      filterByFormula: z.string().min(1).optional(),

      /** Maximum number of records to return across all pages. */
      maxRecords: z.number().int().positive().optional(),

      /** Number of records requested per provider page. */
      pageSize: z.number().int().min(1).max(100).optional(),

      /** Return fields keyed by stable field ID instead of field name. */
      returnFieldsByFieldId: z.boolean().optional(),

      /** Ordered Airtable sort rules. */
      sort: AIRTABLE_SORT_SCHEMA.array().min(1).optional(),

      /** Airtable table name or ID. */
      table: z.string().min(1),

      /** Airtable view name or ID used to constrain and order records. */
      view: z.string().min(1).optional(),
    }),
  )
  .output(
    z.object({
      /** Number of returned records. */
      count: z.number().int().nonnegative(),

      /** Matching Airtable records. */
      records: AIRTABLE_RECORD_SCHEMA.array(),
    }),
  )
  .retry({ replaySafety: "safe" })
  .handler(async ({ account, input }) => {
    const api = getAirtableApi(account.secret)
    const baseId = input.baseId
    const records: z.output<typeof AIRTABLE_RECORD_SCHEMA>[] = []
    let offset: string | undefined

    do {
      const url = getTableUrl(baseId, input.table)
      for (const field of input.fields ?? []) {
        url.searchParams.append("fields[]", field)
      }
      if (input.filterByFormula) {
        url.searchParams.set("filterByFormula", input.filterByFormula)
      }
      if (input.view) url.searchParams.set("view", input.view)
      setBooleanParam(url, "returnFieldsByFieldId", input.returnFieldsByFieldId)
      input.sort?.forEach(({ direction = "asc", field }, index) => {
        url.searchParams.set(`sort[${index}][field]`, field)
        url.searchParams.set(`sort[${index}][direction]`, direction)
      })
      if (offset) url.searchParams.set("offset", offset)

      const remaining =
        input.maxRecords === undefined
          ? undefined
          : input.maxRecords - records.length
      url.searchParams.set(
        "pageSize",
        String(
          Math.min(
            input.pageSize ?? 100,
            remaining === undefined ? 100 : remaining,
          ),
        ),
      )

      const page = AIRTABLE_RECORDS_RESPONSE_SCHEMA.parse(
        await (await api.request(toApiPath(url))).json(),
      )
      records.push(...page.records.map(toAirtableRecord))
      offset = page.offset
    } while (
      offset &&
      (input.maxRecords === undefined || records.length < input.maxRecords)
    )

    const limitedRecords = records.slice(0, input.maxRecords)
    return { count: limitedRecords.length, records: limitedRecords }
  })

/** Creates one Airtable record. */
export const createAirtableRecord = defineAction("Create Airtable record")
  .describe("Creates one record and returns the provider-created value.")
  .account("airtable", AIRTABLE_RECORD_WRITE_REQUIREMENT)
  .input(
    AIRTABLE_WRITE_INPUT_SCHEMA.extend({
      /** Field values keyed by field name or ID. */
      fields: AIRTABLE_FIELDS_SCHEMA,
    }),
  )
  .output(AIRTABLE_RECORD_SCHEMA)
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const url = getTableUrl(input.baseId, input.table)
    setBooleanParam(url, "returnFieldsByFieldId", input.returnFieldsByFieldId)
    return toAirtableRecord(
      AIRTABLE_RECORD_RESPONSE_SCHEMA.parse(
        await (
          await getAirtableApi(account.secret).request(toApiPath(url), {
            body: JSON.stringify({
              fields: input.fields,
              typecast: input.typecast,
            }),
            method: "POST",
          })
        ).json(),
      ),
    )
  })

/** Creates Airtable records in provider-sized batches. */
export const createAirtableRecords = defineAction("Create Airtable records")
  .describe(
    "Creates one or more records in provider-sized batches and returns every created record.",
  )
  .account("airtable", AIRTABLE_RECORD_WRITE_REQUIREMENT)
  .input(
    AIRTABLE_WRITE_INPUT_SCHEMA.extend({
      /** Records to create, each containing field values by name or ID. */
      records: z
        .object({
          fields: AIRTABLE_FIELDS_SCHEMA,
        })
        .array()
        .min(1),
    }),
  )
  .output(AIRTABLE_RECORD_SCHEMA.array())
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const api = getAirtableApi(account.secret)
    const url = getTableUrl(input.baseId, input.table)
    setBooleanParam(url, "returnFieldsByFieldId", input.returnFieldsByFieldId)
    const records: z.output<typeof AIRTABLE_RECORD_SCHEMA>[] = []

    for (
      let index = 0;
      index < input.records.length;
      index += AIRTABLE_BATCH_SIZE
    ) {
      records.push(
        ...AIRTABLE_RECORDS_RESPONSE_SCHEMA.parse(
          await (
            await api.request(toApiPath(url), {
              body: JSON.stringify({
                records: input.records.slice(
                  index,
                  index + AIRTABLE_BATCH_SIZE,
                ),
                typecast: input.typecast,
              }),
              method: "POST",
            })
          ).json(),
        ).records.map(toAirtableRecord),
      )
    }

    return records
  })

/** Updates one Airtable record by merging or replacing its fields. */
export const updateAirtableRecord = defineAction("Update Airtable record")
  .describe("Merges into or replaces one record's writable fields.")
  .account("airtable", AIRTABLE_RECORD_WRITE_REQUIREMENT)
  .input(
    AIRTABLE_WRITE_INPUT_SCHEMA.extend({
      /** New field values keyed by field name or ID. */
      fields: AIRTABLE_FIELDS_SCHEMA,

      /** Merge supplied fields or replace every writable field. */
      mode: z.enum(["merge", "replace"]).optional(),

      /** Stable Airtable record ID. */
      recordId: z.string().min(1),
    }),
  )
  .output(AIRTABLE_RECORD_SCHEMA)
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const url = getRecordUrl(input.baseId, input.table, input.recordId)
    setBooleanParam(url, "returnFieldsByFieldId", input.returnFieldsByFieldId)
    return toAirtableRecord(
      AIRTABLE_RECORD_RESPONSE_SCHEMA.parse(
        await (
          await getAirtableApi(account.secret).request(toApiPath(url), {
            body: JSON.stringify({
              fields: input.fields,
              typecast: input.typecast,
            }),
            method: input.mode === "replace" ? "PUT" : "PATCH",
          })
        ).json(),
      ),
    )
  })

/** Updates Airtable records in provider-sized batches. */
export const updateAirtableRecords = defineAction("Update Airtable records")
  .describe(
    "Merges into or replaces records in provider-sized batches and returns their new values.",
  )
  .account("airtable", AIRTABLE_RECORD_WRITE_REQUIREMENT)
  .input(
    AIRTABLE_WRITE_INPUT_SCHEMA.extend({
      /** Merge supplied fields or replace every writable field. */
      mode: z.enum(["merge", "replace"]).optional(),

      /** Record IDs and field values to update. */
      records: z
        .object({
          /** New field values keyed by field name or ID. */
          fields: AIRTABLE_FIELDS_SCHEMA,

          /** Stable Airtable record ID. */
          id: z.string().min(1),
        })
        .array()
        .min(1),
    }),
  )
  .output(AIRTABLE_RECORD_SCHEMA.array())
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const api = getAirtableApi(account.secret)
    const url = getTableUrl(input.baseId, input.table)
    setBooleanParam(url, "returnFieldsByFieldId", input.returnFieldsByFieldId)
    const records: z.output<typeof AIRTABLE_RECORD_SCHEMA>[] = []

    for (
      let index = 0;
      index < input.records.length;
      index += AIRTABLE_BATCH_SIZE
    ) {
      records.push(
        ...AIRTABLE_RECORDS_RESPONSE_SCHEMA.parse(
          await (
            await api.request(toApiPath(url), {
              body: JSON.stringify({
                records: input.records.slice(
                  index,
                  index + AIRTABLE_BATCH_SIZE,
                ),
                typecast: input.typecast,
              }),
              method: input.mode === "replace" ? "PUT" : "PATCH",
            })
          ).json(),
        ).records.map(toAirtableRecord),
      )
    }

    return records
  })

/** Deletes one Airtable record. */
export const deleteAirtableRecord = defineAction("Delete Airtable record")
  .describe("Deletes one record and returns Airtable's deletion receipt.")
  .account("airtable", AIRTABLE_RECORD_WRITE_REQUIREMENT)
  .input(
    z.object({
      /** Stable Airtable base ID. */
      baseId: AIRTABLE_ID_SCHEMA,

      /** Stable Airtable record ID. */
      recordId: z.string().min(1),

      /** Airtable table name or ID. */
      table: z.string().min(1),
    }),
  )
  .output(
    z.object({
      /** Whether Airtable deleted the record. */
      deleted: z.literal(true),

      /** Stable ID of the deleted record. */
      id: z.string(),
    }),
  )
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    return AIRTABLE_DELETED_RECORD_SCHEMA.parse(
      await (
        await getAirtableApi(account.secret).request(
          toApiPath(getRecordUrl(input.baseId, input.table, input.recordId)),
          { method: "DELETE" },
        )
      ).json(),
    )
  })

/** Deletes Airtable records in provider-sized batches. */
export const deleteAirtableRecords = defineAction("Delete Airtable records")
  .describe(
    "Deletes one or more records in provider-sized batches and returns every deletion receipt.",
  )
  .account("airtable", AIRTABLE_RECORD_WRITE_REQUIREMENT)
  .input(
    z.object({
      /** Stable Airtable base ID. */
      baseId: AIRTABLE_ID_SCHEMA,

      /** Stable IDs of records to delete. */
      recordIds: z.string().min(1).array().min(1),

      /** Airtable table name or ID. */
      table: z.string().min(1),
    }),
  )
  .output(AIRTABLE_DELETED_RECORD_SCHEMA.array())
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const api = getAirtableApi(account.secret)
    const baseId = input.baseId
    const deletedRecords: z.output<typeof AIRTABLE_DELETED_RECORD_SCHEMA>[] = []

    for (
      let index = 0;
      index < input.recordIds.length;
      index += AIRTABLE_BATCH_SIZE
    ) {
      const url = getTableUrl(baseId, input.table)
      for (const recordId of input.recordIds.slice(
        index,
        index + AIRTABLE_BATCH_SIZE,
      )) {
        url.searchParams.append("records[]", recordId)
      }
      deletedRecords.push(
        ...z
          .object({ records: AIRTABLE_DELETED_RECORD_SCHEMA.array() })
          .parse(
            await (
              await api.request(toApiPath(url), { method: "DELETE" })
            ).json(),
          ).records,
      )
    }

    return deletedRecords
  })

/**
 * Lists every Airtable base visible to the connected account.
 *
 * @param api - Authenticated Airtable API client.
 */
async function listAllAirtableBases(api: AirtableApi) {
  const bases: z.output<typeof AIRTABLE_BASES_RESPONSE_SCHEMA>["bases"] = []
  let offset: string | undefined

  do {
    const url = new URL("meta/bases", AIRTABLE_API_ORIGIN)
    if (offset) url.searchParams.set("offset", offset)
    const response = AIRTABLE_BASES_RESPONSE_SCHEMA.parse(
      await (await api.request(toApiPath(url))).json(),
    )
    bases.push(...response.bases)
    offset = response.offset
  } while (offset)

  return bases
}

/**
 * Converts an Airtable record response to its public representation.
 *
 * @param record - Schema-validated provider record.
 */
function toAirtableRecord(
  record: z.output<typeof AIRTABLE_RECORD_RESPONSE_SCHEMA>,
): z.output<typeof AIRTABLE_RECORD_SCHEMA> {
  return {
    ...record,
    createdTime: new Date(record.createdTime),
  }
}

/**
 * Constructs an Airtable table URL with encoded path components.
 *
 * @param baseId - Airtable base ID.
 * @param table - Airtable table name or ID.
 */
function getTableUrl(baseId: string, table: string) {
  return new URL(
    `${encodeURIComponent(baseId)}/${encodeURIComponent(table)}`,
    AIRTABLE_API_ORIGIN,
  )
}

/**
 * Constructs an Airtable record URL with encoded path components.
 *
 * @param baseId - Airtable base ID.
 * @param table - Airtable table name or ID.
 * @param recordId - Stable Airtable record ID.
 */
function getRecordUrl(baseId: string, table: string, recordId: string) {
  const url = getTableUrl(baseId, table)
  url.pathname += `/${encodeURIComponent(recordId)}`
  return url
}

/**
 * Sets an optional boolean query parameter without conflating false and absent.
 *
 * @param url - Airtable URL receiving the parameter.
 * @param name - Query parameter name.
 * @param value - Optional boolean value.
 */
function setBooleanParam(url: URL, name: string, value: boolean | undefined) {
  if (value !== undefined) url.searchParams.set(name, String(value))
}

/**
 * Preserves the query string while making an API URL relative to the v0 root.
 *
 * @param url - Absolute Airtable API URL.
 */
function toApiPath(url: URL) {
  return `${url.pathname.slice(new URL(AIRTABLE_API_ORIGIN).pathname.length)}${url.search}`
}