import { defineAction, type ActionRunContext } from "@agent-native/core/action";
import { accessFilter, assertAccess } from "@agent-native/core/sharing";
import { and, eq } from "drizzle-orm";
import { z } from "zod";

import { createNativeCrmAdapter } from "../server/crm/native-adapter.js";
import { getDb, schema } from "../server/db/index.js";
import {
  assertCrmStatusTransitionAllowed,
  CrmLifecycleError,
  loadCrmStatusLifecycle,
} from "../server/lib/lifecycle.js";
import { decideCrmWritePolicy, type CrmValue } from "../shared/crm-contract.js";
import { resolveCrmSalesDelegatedWrite } from "../shared/crm-sales-config.js";
import {
  crmInitiatedBy,
  crmWriteRisk,
  isSafeCrmMutationFields,
  MAX_CRM_FIELDS_PER_MUTATION,
  scopedCrmIdempotencyKey,
  toJson,
} from "./_crm-action-utils.js";

export const fieldPatchSchema = z
  .record(z.string().trim().min(1).max(120), z.unknown())
  .refine(
    (fields) =>
      Object.keys(fields).length >= 1 &&
      Object.keys(fields).length <= MAX_CRM_FIELDS_PER_MUTATION,
    `Provide between 1 and ${MAX_CRM_FIELDS_PER_MUTATION} fields.`,
  )
  .refine(
    isSafeCrmMutationFields,
    "CRM fields and values cannot contain media, transcripts, data URLs, base64, or oversized JSON.",
  );

type CrmTransaction = Parameters<
  Parameters<ReturnType<typeof getDb>["transaction"]>[0]
>[0];

function fieldColumns(value: CrmValue) {
  if (typeof value === "string") {
    return {
      stringValue: value,
      numberValue: null,
      booleanValue: null,
      jsonValue: null,
    };
  }
  if (typeof value === "number") {
    return {
      stringValue: null,
      numberValue: value,
      booleanValue: null,
      jsonValue: null,
    };
  }
  if (typeof value === "boolean") {
    return {
      stringValue: null,
      numberValue: null,
      booleanValue: value,
      jsonValue: null,
    };
  }
  if (value === null) {
    return {
      stringValue: null,
      numberValue: null,
      booleanValue: null,
      jsonValue: null,
    };
  }
  return {
    stringValue: null,
    numberValue: null,
    booleanValue: null,
    jsonValue: toJson(value, 8_000),
  };
}

function needsLocalApproval(args: {
  target: "local" | "provider";
  fields: Record<string, unknown>;
}) {
  if (args.target !== "local") return false;
  const risk = crmWriteRisk(Object.keys(args.fields));
  return risk !== "routine" || Object.keys(args.fields).length > 1;
}

function isUniqueConstraintError(error: unknown): boolean {
  const message = error instanceof Error ? error.message : "";
  return /unique constraint|duplicate key|already exists/i.test(message);
}

async function updateLocalFields(input: {
  tx: CrmTransaction;
  record: typeof schema.crmRecords.$inferSelect;
  fields: Record<string, CrmValue>;
  policies: Array<typeof schema.crmFieldPolicies.$inferSelect>;
}) {
  const now = new Date().toISOString();
  const policyByName = new Map(
    input.policies.map((policy) => [policy.fieldName, policy]),
  );
  for (const [fieldName, value] of Object.entries(input.fields)) {
    const policy = policyByName.get(fieldName)!;
    const values = {
      ...fieldColumns(value),
      valueType: policy.valueType,
      storagePolicy: "local-authoritative" as const,
      provenanceJson: toJson(
        [
          {
            provider: "native",
            connectionId: input.record.connectionId,
            objectType: input.record.objectType,
            remoteId: input.record.remoteId,
            fieldName,
            observedAt: now,
          },
        ],
        2_000,
      ),
      accessScopeKey: input.record.accessScopeKey,
      accessScopeJson: input.record.accessScopeJson,
      updatedAt: now,
    };
    const existing = await input.tx
      .select({ id: schema.crmRecordFields.id })
      .from(schema.crmRecordFields)
      .where(
        and(
          eq(schema.crmRecordFields.recordId, input.record.id),
          eq(schema.crmRecordFields.fieldName, fieldName),
          accessFilter(schema.crmRecordFields, schema.crmRecordFieldShares),
        ),
      )
      .limit(1);
    if (existing[0]) {
      await input.tx
        .update(schema.crmRecordFields)
        .set(values)
        .where(
          and(
            eq(schema.crmRecordFields.id, existing[0].id),
            accessFilter(
              schema.crmRecordFields,
              schema.crmRecordFieldShares,
              undefined,
              "editor",
            ),
          ),
        );
    } else {
      await input.tx.insert(schema.crmRecordFields).values({
        id: crypto.randomUUID(),
        recordId: input.record.id,
        fieldPolicyId: policy.id,
        fieldName,
        remoteRevision: null,
        ...values,
        ownerEmail: input.record.ownerEmail,
        orgId: input.record.orgId,
        visibility: input.record.visibility,
        createdAt: now,
      });
    }
  }

  if (
    Object.prototype.hasOwnProperty.call(input.fields, "desiredCadenceDays")
  ) {
    const cadence = input.fields.desiredCadenceDays;
    if (
      cadence !== null &&
      (typeof cadence !== "number" ||
        !Number.isInteger(cadence) ||
        cadence < 1 ||
        cadence > 365)
    ) {
      throw new Error(
        "desiredCadenceDays must be null or an integer from 1 to 365.",
      );
    }
    await input.tx
      .update(schema.crmRecords)
      .set({ desiredCadenceDays: cadence, updatedAt: now })
      .where(
        and(
          eq(schema.crmRecords.id, input.record.id),
          accessFilter(
            schema.crmRecords,
            schema.crmRecordShares,
            undefined,
            "editor",
          ),
        ),
      );
  }
}

export default defineAction({
  description:
    "Submit a revision-aware CRM record update. The initial release always saves HubSpot and Salesforce provider updates as proposals and fails closed unless an atomic expected-revision write path is proven. Local-authoritative fields may be applied after the applicable approval policy is satisfied.",
  schema: z.object({
    recordId: z.string().trim().min(1).max(128),
    target: z.enum(["local", "provider"]).default("provider"),
    fields: fieldPatchSchema,
    expectedRemoteRevision: z.string().trim().min(1).max(256).optional(),
    idempotencyKey: z.string().trim().min(1).max(256).optional(),
  }),
  needsApproval: needsLocalApproval,
  audit: {
    target: (_args, result) => {
      const response = result as {
        recordId: string;
        ownerEmail: string;
        orgId: string | null;
        visibility: "private" | "org";
      };
      return {
        type: "crm-record",
        id: response.recordId,
        ownerEmail: response.ownerEmail,
        orgId: response.orgId,
        visibility: response.visibility,
      };
    },
    summary: (args) =>
      `Submitted CRM ${args.target} update for record ${args.recordId}`,
    recordInputs: false,
  },
  run: async (args, ctx?: ActionRunContext) => {
    await assertAccess("crm-record", args.recordId, "editor");
    const db = getDb();
    const [record] = await db
      .select()
      .from(schema.crmRecords)
      .where(
        and(
          eq(schema.crmRecords.id, args.recordId),
          accessFilter(
            schema.crmRecords,
            schema.crmRecordShares,
            undefined,
            "editor",
          ),
        ),
      )
      .limit(1);
    if (!record || record.tombstone)
      throw new Error("CRM record is unavailable for updates.");

    const fields = args.fields as Record<string, CrmValue>;
    const fieldNames = Object.keys(fields);
    const policies = await db
      .select()
      .from(schema.crmFieldPolicies)
      .where(
        and(
          eq(schema.crmFieldPolicies.connectionId, record.connectionId),
          eq(schema.crmFieldPolicies.objectType, record.objectType),
          accessFilter(schema.crmFieldPolicies, schema.crmFieldPolicyShares),
        ),
      );
    const policyByName = new Map(
      policies.map((policy) => [policy.fieldName, policy]),
    );
    const missing = fieldNames.filter(
      (fieldName) => !policyByName.has(fieldName),
    );
    const blocked = fieldNames.filter((fieldName) => {
      const policy = policyByName.get(fieldName);
      return (
        !policy || !policy.updateable || policy.storagePolicy === "redacted"
      );
    });
    if (missing.length || blocked.length) {
      const error = new Error(
        `Only discovered, updateable CRM fields can be changed. Unsupported: ${[...new Set([...missing, ...blocked])].join(", ")}`,
      ) as Error & { statusCode?: number };
      error.statusCode = 422;
      throw error;
    }

    const wrongAuthority = fieldNames.filter((fieldName) => {
      const storagePolicy = policyByName.get(fieldName)?.storagePolicy;
      return args.target === "local"
        ? storagePolicy !== "local-authoritative"
        : storagePolicy === "derived-local" ||
            storagePolicy === "local-authoritative";
    });
    if (wrongAuthority.length) {
      throw new Error(
        `${args.target === "local" ? "Local" : "Provider"} authority does not own: ${wrongAuthority.join(", ")}`,
      );
    }

    // A status value is a lifecycle transition, not just another local value:
    // the enterable set is the attribute's own options, so a move into a
    // retired or undeclared stage is refused with the reason a person can act
    // on. Only local writes are gated — a provider-target status change is a
    // proposal, which is the working flow, not a blocked transition.
    if (args.target === "local") {
      for (const fieldName of fieldNames) {
        const policy = policyByName.get(fieldName)!;
        if (policy.attributeType !== "status") continue;
        const value = fields[fieldName];
        // Leaving a status is always allowed, including a stage that is now
        // retired; only entering one is governed.
        if (value === null) continue;
        if (typeof value !== "string") {
          throw new CrmLifecycleError(
            "crm-status-value-type",
            `"${fieldName}" is a status attribute; its value must be one of its declared options.`,
          );
        }
        await assertCrmStatusTransitionAllowed({
          db,
          lifecycle: await loadCrmStatusLifecycle(db, policy.id),
          target: { recordId: record.id },
          to: value,
        });
      }
    }

    const risk = crmWriteRisk(fieldNames);
    const initiatedBy = crmInitiatedBy(ctx);
    const writePolicy = {
      initiatedBy,
      target: args.target,
      reversibility: "compensatable",
      scope: fieldNames.length === 1 ? "single-field" : "single-record",
      risk,
    } as const;
    const delegation = resolveCrmSalesDelegatedWrite({
      context: ctx,
      operation: "update",
      policy: {
        ...writePolicy,
        delegatedAuthority: false,
        storedAutomationPolicy: false,
      },
    });
    const decision = decideCrmWritePolicy({ ...writePolicy, ...delegation });
    if (decision === "deny")
      throw new Error(
        "This CRM update is not authorized by the current write policy.",
      );
    if (
      initiatedBy === "automation" &&
      args.target === "local" &&
      decision !== "execute"
    ) {
      throw new Error(
        "This CRM automation may only execute an explicitly delegated routine local update.",
      );
    }

    const ownership = {
      ownerEmail: record.ownerEmail,
      orgId: record.orgId,
      visibility: record.visibility,
    };
    const patchJson = toJson({ fields }, 12_000);
    const expectedRemoteRevision =
      args.target === "provider"
        ? (args.expectedRemoteRevision ?? record.remoteRevision)
        : (args.expectedRemoteRevision ?? null);
    if (args.target === "provider" && !expectedRemoteRevision) {
      throw new Error(
        "CRM provider updates require a current remote revision. Refresh the record before proposing a change.",
      );
    }
    const idempotencyKey = await scopedCrmIdempotencyKey({
      ...ownership,
      recordId: record.id,
      key: args.idempotencyKey ?? crypto.randomUUID(),
    });
    const findExisting = async () => {
      const [existing] = await db
        .select()
        .from(schema.crmMutations)
        .where(
          and(
            eq(schema.crmMutations.idempotencyKey, idempotencyKey),
            accessFilter(schema.crmMutations, schema.crmMutationShares),
          ),
        )
        .limit(1);
      return existing;
    };
    const replay = (existing: typeof schema.crmMutations.$inferSelect) => {
      if (
        existing.recordId !== record.id ||
        existing.target !== args.target ||
        existing.patchJson !== patchJson ||
        existing.expectedRemoteRevision !== expectedRemoteRevision
      ) {
        throw new Error(
          "CRM idempotency key was already used for a different mutation.",
        );
      }
      return {
        mutationId: existing.id,
        recordId: record.id,
        status: existing.status,
        decision: existing.policyDecision,
        replayed: true as const,
        ownerEmail: record.ownerEmail,
        orgId: record.orgId,
        visibility: record.visibility,
      };
    };
    const existing = await findExisting();
    if (existing) return replay(existing);

    const now = new Date().toISOString();
    const mutationId = crypto.randomUUID();
    if (args.target === "local" && record.provider === "native") {
      if (!args.expectedRemoteRevision) {
        throw new Error(
          "Native SQL CRM updates require the current record revision. Refresh the record and retry.",
        );
      }
      const adapter = await createNativeCrmAdapter({
        connectionId: record.connectionId,
        initiatedBy,
      });
      const result = await adapter.applyMutation({
        operation: "update",
        record: {
          connectionId: record.connectionId,
          provider: "native",
          objectType: record.objectType,
          kind: record.kind,
          remoteId: record.remoteId,
          localId: record.id,
        },
        fields,
        expectedRemoteRevision: args.expectedRemoteRevision,
        idempotencyKey,
      });
      if (result.status !== "applied") {
        throw new Error(
          result.message ?? `Native SQL CRM update ${result.status}.`,
        );
      }
      const applied = await findExisting();
      if (!applied) {
        throw new Error(
          "Native SQL CRM update succeeded but its audit mutation could not be verified.",
        );
      }
      return {
        mutationId: applied.id,
        recordId: record.id,
        status: "applied" as const,
        revision: result.remoteRevision,
        ownerEmail: record.ownerEmail,
        orgId: record.orgId,
        visibility: record.visibility,
      };
    }
    if (args.target === "local") {
      try {
        await db.transaction(async (tx) => {
          await updateLocalFields({ tx, record, fields, policies });
          await tx.insert(schema.crmMutations).values({
            id: mutationId,
            recordId: record.id,
            connectionId: record.connectionId,
            operation: "update",
            initiatedBy,
            target: "local",
            policyDecision: decision,
            risk,
            status: "applied",
            patchJson,
            beforeJson: "{}",
            afterJson: patchJson,
            idempotencyKey,
            expectedRemoteRevision,
            appliedAt: now,
            ...ownership,
            createdAt: now,
            updatedAt: now,
          });
        });
      } catch (error) {
        if (!isUniqueConstraintError(error)) throw error;
        const raced = await findExisting();
        if (raced) return replay(raced);
        throw error;
      }
      return {
        mutationId,
        recordId: record.id,
        status: "applied" as const,
        ownerEmail: record.ownerEmail,
        orgId: record.orgId,
        visibility: record.visibility,
      };
    }

    try {
      await db.insert(schema.crmMutations).values({
        id: mutationId,
        recordId: record.id,
        connectionId: record.connectionId,
        operation: "update",
        initiatedBy,
        target: "provider",
        policyDecision: decision,
        risk,
        status: "pending",
        patchJson,
        beforeJson: toJson({ remoteRevision: record.remoteRevision }, 1_000),
        afterJson: "{}",
        idempotencyKey,
        expectedRemoteRevision,
        ...ownership,
        createdAt: now,
        updatedAt: now,
      });
    } catch (error) {
      if (!isUniqueConstraintError(error)) throw error;
      const raced = await findExisting();
      if (raced) return replay(raced);
      throw error;
    }
    return {
      mutationId,
      recordId: record.id,
      status: "pending" as const,
      decision,
      ownerEmail: record.ownerEmail,
      orgId: record.orgId,
      visibility: record.visibility,
    };
  },
});