import { CollectionSchemaBase, SchemaBase, Field, isAggregate } from './dsl.js';
import type { FrontAppSchema, ProjectApiSchema } from './project.js';
import type { DtoMessage } from './dto.js';
import { TableSchema } from './db.js';
import type { EntitySchema } from './entity.js';
import type { FilterSchema } from './filter.js';
import type { SetExpr, ValueExpr } from './expr.js';

/** A data-access layer bound to one frontend app, or to the api-level common
 *  domain layer (app unset — shared across modules). */
export interface DaoSchema extends CollectionSchemaBase {
  type: 'dao';
  /** The backend api module this DAO belongs to (shared instance from
   *  project.config.ts apis). DAOs are always backend-side, so storage is
   *  dao_schema/{api.name}/{app.name}/dao/ — app unset = the api-level common
   *  domain layer, stored at dao_schema/{api.name}/common/dao/. */
  api: ProjectApiSchema;
  /** The frontend app this DAO belongs to (shared instance from project.config).
   *  Unset = api-level common domain DAO shared by all modules of the api. */
  app?: FrontAppSchema;
  /** The table this DAO operates on (single-table atomicity). */
  table: TableSchema;
  /** Methods keyed by name — the map key is written back as the method name. */
  methods: Record<string, DaoMethodSchema>;
}

/** Method input for defineDao: name is written back from the methods map key. */
export type DaoMethodDef =
  | Omit<FindSchema, 'schema' | 'name'>
  | Omit<GetSchema, 'schema' | 'name'>
  | Omit<InsertSchema, 'schema' | 'name'>
  | Omit<UpdateSchema, 'schema' | 'name'>
  | Omit<DeleteSchema, 'schema' | 'name'>
  | Omit<UpsertSchema, 'schema' | 'name'>
  | Omit<AggregateSchema, 'schema' | 'name'>;

/** The tenant column of the dao table when the app declares a tenant.
 *  Deterministic name derivation: `{tenant.phrase}_{tenant.pk}` (e.g. shop
 *  with pk id → `shop_id`) — checked directly against the table columns,
 *  no FK traversal. Tables without that column are global tables (valid:
 *  system config tables carry no tenant id). Exported for generator/linter. */
export function tenantFkOf(dao: DaoSchema): Field | undefined {
  const app = dao.app;
  const tenant = app?.tenant;
  if (!tenant || !app) return undefined;
  const pk = tenant.primaryKey;
  const phrase = tenant.phrase;
  if (!pk || Array.isArray(pk) || !phrase) {
    throw new Error(`app '${app.name}' tenant table '${tenant.name}' must declare a single-column primaryKey and a phrase (tenant column name = {phrase}_{pk})`);
  }
  return dao.table.columns[`${phrase.name}_${pk.name}`];
}

/** The optimistic-lock version column of the dao table (table-level
 *  declaration), machine-enforced: update args must carry it, set must not
 *  touch it. */
function versionOf(dao: DaoSchema): Field | undefined {
  return dao.table.version;
}

/** Validates a ValueExpr: column refs belong to the dao table, param names
 *  are unique, literal/bin types are numeric-compatible. */
function validateValueExpr(dao: DaoSchema, expr: ValueExpr, params: Set<string>, methodName: string): void {
  const table = dao.table;
  const cols = Object.values(table.columns);
  switch (expr.kind) {
    case 'col':
      if (!cols.includes(expr.field)) {
        throw new Error(`dao ${dao.name}.${methodName}: expr references column '${expr.field.name}' which is not a column of table ${table.name}`);
      }
      return;
    case 'lit':
      if (typeof expr.value === 'string') {
        throw new Error(`dao ${dao.name}.${methodName}: string literal in a numeric expression — use a param instead`);
      }
      return;
    case 'param':
      if (params.has(expr.name)) {
        throw new Error(`dao ${dao.name}.${methodName}: duplicate param '${expr.name}' in set expressions`);
      }
      params.add(expr.name);
      return;
    case 'bin':
      validateValueExpr(dao, expr.left, params, methodName);
      validateValueExpr(dao, expr.right, params, methodName);
      return;
  }
}

function validateUpdateSet(dao: DaoSchema, methodName: string, m: UpdateSchema): void {
  const argCols = new Set(m.args.columns.map((c) => c.name));
  const version = versionOf(dao);
  // Where criteria columns are locators carried by args, not direct assignments —
  // expression-setting the same column is the conditional-update idiom
  // (WHERE state = row.state, SET state = ?).
  const whereCols = new Set((m.where?.conditions ?? []).map((c) => c.field.name));
  for (const setExpr of m.set ?? []) {
    if (version && setExpr.col === version) {
      throw new Error(`dao ${dao.name}.${methodName}: set must not touch version column '${version.name}' — the optimistic lock manages it`);
    }
    if (!Object.values(dao.table.columns).includes(setExpr.col)) {
      throw new Error(`dao ${dao.name}.${methodName}: set column '${setExpr.col.name}' is not a column of table ${dao.table.name}`);
    }
    if (argCols.has(setExpr.col.name) && !whereCols.has(setExpr.col.name)) {
      throw new Error(`dao ${dao.name}.${methodName}: set column '${setExpr.col.name}' also appears in args '${m.args.name}' — a column is either directly assigned or expression-set, never both`);
    }
    validateValueExpr(dao, setExpr.expr, new Set(), methodName);
  }
}

/** Validates read-method result columns: external reference columns must be
 *  reachable through a main-table foreign key (FK = join condition). */
function assertColumnReachable(dao: DaoSchema, methodName: string, c: Field): void {
  const table = dao.table;
  if (!c.schema) {
    throw new Error(`dao ${dao.name}.${methodName}: row column '${c.name}' has no schema`);
  }
  if (c.schema === table) return;
  const srcTable = c.schema as TableSchema;
  const reachable = Object.values(table.foreignKeys ?? {}).some((fk) => {
    const refs = Array.isArray(fk.references) ? fk.references : [fk.references];
    return refs.some((r) => r.schema === srcTable);
  });
  if (!reachable) {
    throw new Error(`dao ${dao.name}.${methodName}: row column '${c.name}' belongs to table ${srcTable.name} but ${table.name} has no foreign key pointing to it`);
  }
}

/** Validates read-method result columns: external reference columns must be
 *  reachable through a main-table foreign key (FK = join condition). */
function validateRowColumns(dao: DaoSchema, methodName: string, row: EntitySchema): void {
  for (const c of row.columns) assertColumnReachable(dao, methodName, c);
}

/** Validates key args (get/delete): every key field must be a main-table
 *  column; the tenant column is injected as a separate parameter and can
 *  never be a key (a duplicate parameter would render). */
function validateKeyArgs(dao: DaoSchema, methodName: string, args: Field | Field[]): void {
  const cols = Object.values(dao.table.columns);
  const keys = Array.isArray(args) ? args : [args];
  for (const k of keys) {
    if (!cols.includes(k)) {
      throw new Error(`dao ${dao.name}.${methodName}: key column '${k.name}' is not a column of table ${dao.table.name}`);
    }
  }
  const tenant = tenantFkOf(dao);
  if (tenant && keys.includes(tenant)) {
    throw new Error(`dao ${dao.name}.${methodName}: key '${tenant.name}' is the tenant column — it is injected as a separate parameter, never a key`);
  }
}

/** Update runs without JOINs, so its where criteria must all reference
 *  main-table columns — a cross-table criterion would render a raw column
 *  name that no table in the query provides. Also requires at least one
 *  locating criterion: without a pk/tenant/version/filter the generated
 *  UPDATE would have an empty WHERE and touch every row. */
function validateUpdateWhere(dao: DaoSchema, methodName: string, m: UpdateSchema): void {
  if (m.where) {
    for (const c of m.where.conditions) {
      if (c.field.schema !== dao.table) {
        const srcName = (c.field.schema as TableSchema | undefined)?.name ?? 'unknown';
        throw new Error(`dao ${dao.name}.${methodName}: where criterion on '${c.field.name}' belongs to table '${srcName}' — update is single-table and cannot join`);
      }
      if (c.optional) {
        throw new Error(`dao ${dao.name}.${methodName}: where criterion on '${c.field.name}' is optional — update criteria must be required (a missing value would silently drop the criterion)`);
      }
      if (!m.args.columns.includes(c.field)) {
        throw new Error(`dao ${dao.name}.${methodName}: where criterion on '${c.field.name}' is not carried by args '${m.args.name}' — the generated UPDATE reads it from the row`);
      }
    }
  }
  const pk = dao.table.primaryKey;
  const hasPk = pk !== undefined && (Array.isArray(pk) ? pk.length > 0 : true);
  const hasLocator = hasPk || tenantFkOf(dao) !== undefined || versionOf(dao) !== undefined || (m.where !== undefined && m.where.conditions.length > 0);
  if (!hasLocator) {
    throw new Error(`dao ${dao.name}.${methodName}: update on table '${dao.table.name}' has no locating criteria — declare a primaryKey, tenant, version column, or a where filter (an empty WHERE updates every row)`);
  }
}

/** Order-by columns must be part of the result row: the generated ORDER BY
 *  uses the row's result keys, which only exist for selected columns. */
function validateOrderBy(dao: DaoSchema, methodName: string, m: FindSchema): void {
  if (!m.orderBy) return;
  const orders = Array.isArray(m.orderBy) ? m.orderBy : [m.orderBy];
  for (const o of orders) {
    if (!m.results.columns.includes(o.column)) {
      throw new Error(`dao ${dao.name}.${methodName}: orderBy column '${o.column.name}' is not part of results '${m.results.name}'`);
    }
  }
}

/** Aggregate results: the result entity must carry at least one column; every
 *  plain column (grouping dimension) and every aggregate field's underlying
 *  column must be reachable (same rule as read results). */
function validateAggregateResults(dao: DaoSchema, methodName: string, m: AggregateSchema): void {
  const columns = m.results.columns;
  if (columns.length === 0) {
    throw new Error(`dao ${dao.name}.${methodName}: aggregate has no results`);
  }
  for (const c of columns) {
    if (isAggregate(c)) {
      if (c.expr.field) assertColumnReachable(dao, methodName, c.expr.field);
    } else {
      assertColumnReachable(dao, methodName, c);
    }
  }
}

/** Upsert conflict keys: must equal the pk or a complete unique index column
 *  set, and every key must be carried by args. Auto-increment tables are
 *  forbidden (MySQL auto_increment burns ids on duplicate-key updates). */
function validateUpsert(dao: DaoSchema, methodName: string, m: UpsertSchema): void {
  const table = dao.table;
  if (table.autoIncrement) {
    throw new Error(`dao ${dao.name}.${methodName}: upsert on table '${table.name}' with auto-increment pk is forbidden (MySQL auto_increment burns ids on duplicate-key updates)`);
  }
  const keys = Array.isArray(m.keys) ? m.keys : [m.keys];
  if (keys.length === 0) {
    throw new Error(`dao ${dao.name}.${methodName}: upsert keys must be non-empty`);
  }
  for (const k of keys) {
    if (k.schema !== table) {
      throw new Error(`dao ${dao.name}.${methodName}: upsert key '${k.name}' is not a column of table ${table.name}`);
    }
  }
  const keyNames = keys.map((k) => k.name).sort();
  const candidates: Array<string[]> = [];
  const uniqueCols = new Set<string>();
  const pk = table.primaryKey;
  if (pk) {
    const cs = (Array.isArray(pk) ? pk : [pk]).map((c) => c.name);
    candidates.push(cs.sort());
    for (const n of cs) uniqueCols.add(n);
  }
  for (const idx of table.indexes ?? []) {
    if (idx.unique) {
      const cs = (Array.isArray(idx.columns) ? idx.columns : [idx.columns]).map((c) => c.name);
      candidates.push(cs.sort());
      for (const n of cs) uniqueCols.add(n);
    }
  }
  const matches = candidates.some((c) => c.length === keyNames.length && c.every((n, i) => n === keyNames[i]));
  if (!matches) {
    throw new Error(`dao ${dao.name}.${methodName}: upsert keys [${keys.map((k) => k.name).join(', ')}] must equal the pk or a complete unique index column set of table ${table.name}`);
  }
  for (const k of keys) {
    if (!m.args.columns.includes(k)) {
      throw new Error(`dao ${dao.name}.${methodName}: upsert args '${m.args.name}' must carry conflict key '${k.name}'`);
    }
  }
  // ON DUPLICATE KEY UPDATE has no WHERE — tenant isolation is only possible
  // when the tenant column participates in the conflict key.
  const tenant = tenantFkOf(dao);
  if (tenant && !keys.includes(tenant)) {
    throw new Error(`dao ${dao.name}.${methodName}: upsert on tenant-scoped table '${table.name}' must include tenant column '${tenant.name}' in keys (ON DUPLICATE KEY UPDATE has no WHERE — tenant isolation requires the tenant column to be part of the conflict key)`);
  }
  // The merge clause (col = new.col) only writes non-key writable columns —
  // conflict keys are the conflict identity, readOnly columns stay
  // DB-managed, and every pk/unique column is excluded: a merge writing a
  // unique column could collide with another row's value and chain-fire the
  // duplicate-key handler (unique columns are conflict identity, not data).
  const mergeCols = m.args.columns.filter((c) => !keys.includes(c) && !c.readOnly && !uniqueCols.has(c.name));
  if (mergeCols.length === 0) {
    throw new Error(`dao ${dao.name}.${methodName}: upsert has no merge columns — args '${m.args.name}' carries only conflict keys, pk/unique and readOnly columns (ON DUPLICATE KEY UPDATE needs at least one writable non-unique column)`);
  }
}

/** Validates write-method args: every column must come from the dao table
 *  (single-table atomicity). */
function validateWriteArgs(dao: DaoSchema, methodName: string, args: EntitySchema): void {
  const cols = Object.values(dao.table.columns);
  for (const c of args.columns) {
    if (!cols.includes(c)) {
      throw new Error(`dao ${dao.name}.${methodName}: args column '${c.name}' of '${args.name}' is not a column of table ${dao.table.name}`);
    }
  }
}

/** Enforces the tenant/version column presence on row-carried methods
 *  (decisions: insert = declared in args; update = extracted from row).
 *  Tenant is required on both writes (the column is always scoped); version
 *  is required only on update — inserts let the DB default initialize it. */
function validateRowCarriedColumns(dao: DaoSchema, methodName: string, method: InsertSchema | UpdateSchema | UpsertSchema): void {
  const tenant = tenantFkOf(dao);
  if (tenant) {
    if (!method.args.columns.includes(tenant)) {
      throw new Error(`dao ${dao.name}.${methodName}: args '${method.args.name}' must include tenant column '${tenant.name}'`);
    }
  }
  if (method.type === 'update') {
    const version = versionOf(dao);
    if (version && !method.args.columns.includes(version)) {
      throw new Error(`dao ${dao.name}.${methodName}: args '${method.args.name}' must include version column '${version.name}' — the optimistic lock reads it from the row`);
    }
  }
}

export function defineDao(options: {
  name: string;
  api: ProjectApiSchema;
  app?: FrontAppSchema;
  table: TableSchema;
  methods: Record<string, DaoMethodDef>;
  description?: string;
}): DaoSchema {
  if (options.app && !options.api.apps.includes(options.app)) {
    throw new Error(`dao ${options.name}: api '${options.api.name}' does not serve app '${options.app.name}'`);
  }
  const schema: DaoSchema = {
    type: 'dao',
    name: options.name,
    description: options.description,
    api: options.api,
    app: options.app,
    table: options.table,
    methods: {},
  };
  for (const key of Object.keys(options.methods)) {
    const method = options.methods[key] as DaoMethodDef;
    for (const ref of [method.args, 'where' in method ? method.where : undefined]) {
      if (isFilterSchema(ref) && ref.api !== options.api) {
        throw new Error(
          `dao ${options.name}: method '${key}' references filter '${ref.name}' bound to ${ref.api.name}/${ref.app?.name ?? 'common'} but the dao is bound to ${options.api.name}/${options.app?.name ?? 'common'}`,
        );
      }
      // app-bound dao may use common or same-app filters; common dao may only use common filters.
      if (isFilterSchema(ref) && ref.api === options.api && !options.app && ref.app) {
        throw new Error(
          `dao ${options.name}: method '${key}' references app-bound filter '${ref.name}' (${ref.app.name}) but the dao is api-level common`,
        );
      }
      if (isFilterSchema(ref) && ref.api === options.api && options.app && ref.app && ref.app !== options.app) {
        throw new Error(
          `dao ${options.name}: method '${key}' references filter '${ref.name}' bound to ${ref.api.name}/${ref.app.name} but the dao is bound to ${options.api.name}/${options.app.name}`,
        );
      }
    }
    // Method-kind validation (declaration is complete — every kind is checked).
    if (method.type === 'insert' || method.type === 'update' || method.type === 'upsert') {
      const m = method as InsertSchema | UpdateSchema | UpsertSchema;
      validateWriteArgs(schema, key, m.args as EntitySchema);
      validateRowCarriedColumns(schema, key, m);
      if (method.type === 'update') {
        validateUpdateSet(schema, key, method as UpdateSchema);
        validateUpdateWhere(schema, key, method as UpdateSchema);
      }
      if (method.type === 'upsert') {
        validateUpsert(schema, key, method as UpsertSchema);
      }
    }
    if (method.type === 'get' || method.type === 'delete') {
      validateKeyArgs(schema, key, (method as GetSchema | DeleteSchema).args);
    }
    if (method.type === 'find' || method.type === 'get') {
      validateRowColumns(schema, key, (method as FindSchema | GetSchema).results);
    }
    if (method.type === 'find') {
      validateOrderBy(schema, key, method as FindSchema);
    }
    if (method.type === 'aggregate') {
      validateAggregateResults(schema, key, method as AggregateSchema);
    }
    // Spread of a union loses discriminant correlation; the cast is safe
    // (the builder only adds name and the back-reference field).
    schema.methods[key] = { ...method, name: key, schema } as DaoMethodSchema;
  }
  return schema;
}

function isFilterSchema(value: unknown): value is FilterSchema {
  if (typeof value !== 'object' || value === null) return false;
  const v = value as Record<string, unknown>;
  return v.type === 'filter' && typeof v.name === 'string';
}

// DAO method kinds. Signature only: name + params + result. No logic —
// complex SQL goes into description (text), never into schema.
// Query conditions are FilterSchema references (the single shared filter
// model) — never inline criteria.

/** Sort specification for find results. */
export interface OrderBySchema {
  /** Column to sort by. */
  column: Field;
  /** Sort direction. */
  sort: 'asc' | 'desc';
}

/** Query rows by criteria; returns a list of row objects. */
export interface FindSchema extends SchemaBase {
  type: 'find';
  schema: DaoSchema;
  /** Query filter; omit = all rows. */
  args?: FilterSchema;
  /** Pagination marker: 'page' (page/pageSize) or 'limit' (position/limit).
   *  Only presence matters — parameter shapes are a generator convention. */
  mode?: 'page' | 'limit';
  /** Result sort; omit = no explicit order. */
  orderBy?: OrderBySchema | OrderBySchema[];
  /** Result row: selected columns (multi-table via external reference columns). */
  results: EntitySchema;
}

/** Fetch a single row by exact key — any column or AND combination of
 *  columns (single PK, composite PK, getByXX are all legal declarations).
 *  May miss the row: the generated signature returns Row | null. */
export interface GetSchema extends SchemaBase {
  type: 'get';
  schema: DaoSchema;
  /** Key column(s): one field or AND-exact-matched fields. */
  args: Field | Field[];
  /** Additional criteria AND-combined onto the key (e.g. state condition
   *  for optimistic reads). */
  where?: FilterSchema;
  /** Result row: selected columns (multi-table via external reference columns). */
  results: EntitySchema;
}

/** Insert one row; returns the generated key (string). */
export interface InsertSchema extends SchemaBase {
  type: 'insert';
  schema: DaoSchema;
  /** Row object. */
  args: EntitySchema;
}

/** Update one row; returns number (affected rows). The where key is derived
 *  from the PK columns inside args; tenant/version columns are extracted
 *  from the row into WHERE and never SET (optimistic lock auto-manages
 *  `version = version + 1`). */
export interface UpdateSchema extends SchemaBase {
  type: 'update';
  schema: DaoSchema;
  /** Row object containing the PK columns plus the columns to set. */
  args: EntitySchema;
  /** Expression-set columns (`col = expr`) beyond direct assignment. */
  set?: SetExpr[];
  /** Additional criteria beyond the derived PK (AND-combined). */
  where?: FilterSchema;
}

/** Delete one row by exact key; returns number (affected rows). */
export interface DeleteSchema extends SchemaBase {
  type: 'delete';
  schema: DaoSchema;
  /** Key column(s): one field or AND-exact-matched fields. */
  args: Field | Field[];
}

/** Insert-or-update (MySQL ON DUPLICATE KEY UPDATE): atomic idempotent write.
 *  Returns affected rows (1 = inserted, 2 = updated, 0 = unchanged).
 *  No id generation — every conflict key must be carried in args. Tables with
 *  an auto-increment pk are forbidden (MySQL auto_increment burns ids on
 *  duplicate-key updates). */
export interface UpsertSchema extends SchemaBase {
  type: 'upsert';
  schema: DaoSchema;
  /** Row object: must carry every conflict key plus the columns to merge. */
  args: EntitySchema;
  /** Conflict keys: must equal the pk or a complete unique index column set. */
  keys: Field | Field[];
}

export type DaoMethodSchema =
  | FindSchema
  | GetSchema
  | InsertSchema
  | UpdateSchema
  | DeleteSchema
  | UpsertSchema
  | AggregateSchema;

/** Aggregate query (find upgraded at the select level): the result entity
 *  mixes plain columns (the GROUP BY dimensions) and aggregate fields
 *  (aggField — count/sum/avg).
 *
 *  Usage — aggregate fields are defined in the entity file, the dao method
 *  only references the entity:
 *
 *  ```ts
 *  // entity_schema/{api}/{app}/entity/order.entity.ts
 *  export const orderStatusStats = defineEntity({
 *    name: 'OrderStatusStats',
 *    api, app,
 *    columns: [
 *      order.columns.status,                      // GROUP BY dimension
 *      aggField('total', Compute.count()),        // count → jsType 'number'
 *      aggField('sumAmt', Compute.sum(order.columns.amount)), // sum(decimal) → 'string'
 *    ],
 *  });
 *
 *  // dao_schema/{api}/{app}/dao/order.dao.ts
 *  stats: { type: 'aggregate', args: orderFilter, results: orderStatusStats },
 *  ``` */
export interface AggregateSchema extends SchemaBase {
  type: 'aggregate';
  schema: DaoSchema;
  /** Criteria filter, same shape as find. */
  args?: FilterSchema;
  /** Result entity: plain columns group the rows (one row per group);
   *  aggregate fields become the computed output columns. The entity — with
   *  its aggField definitions — lives in the entity file ({table}.entity.ts);
   *  the dao method only references it, never defines aggregate fields inline. */
  results: EntitySchema;
}