import 'server-only'

import { prisma } from '../db'
import {
  ADAPTERS,
  ADAPTER_PROTOCOLS,
  FORBIDDEN_VERBS,
  assessLiveReadiness,
  type AdapterSpec,
  type ReadinessReport,
} from '../engine/adapters'
import {
  assessSourceQuality,
  connectorHealth,
  connectorLabel,
  faultEffect,
  normaliseRecord,
  pipelineSteps,
  type ConnectorMode,
  type ConnectorStatus,
  type FaultKind,
  type FieldMapping,
  type PipelineStep,
  type SourceQuality,
} from '../engine/integration'
import { round, seededRange } from '../engine/rng'
import { getGridSnapshot } from './grid-service'

/**
 * The Integration Hub's server side.
 *
 * The honesty rule this file exists to hold: **a demo connector is never described as
 * connected without the word demo attached.** `mode` is stored on the row, the label is
 * derived from the pair, and the API returns both — so an interface cannot show "CONNECTED"
 * against a generator by forgetting a flag.
 *
 * The second rule is that a sync is a *real* pass through the pipeline. Records are
 * generated from the same simulation engine the rest of the platform runs on, then put
 * through mapping, unit conversion, validation, duplicate and outlier detection, and the
 * counts that come out are the counts the quality score is computed from. Nothing is
 * asserted: when a fault is injected, quality falls because records genuinely failed.
 *
 * Credentials never leave the server. `secretRef` names the environment variable a
 * connector would read; the value is never selected, returned or logged.
 */

const MIN = 60_000

// ─────────────────────────────────────────────────────────────────────────────
// Reading
// ─────────────────────────────────────────────────────────────────────────────

export interface ConnectorView {
  code: string
  name: string
  nameAr: string
  kind: string
  protocol: string
  mode: ConnectorMode
  status: ConnectorStatus
  /** The only string a UI should show for connection state. */
  label: string
  direction: string
  description: string
  descriptionAr: string
  endpoint: string
  authKind: string
  /** The *name* of the secret, never its value. */
  secretRef: string
  pollSeconds: number
  timeoutMs: number
  lastConnectedAt: number | null
  lastSyncAt: number | null
  latencyMs: number
  recordsReceived: number
  errorCount: number
  healthScore: number
  dataQuality: number
  eventsPerMin: number
  faultKind: string
  mappingCount: number
  isActive: boolean
}

export interface HubOverview {
  total: number
  connected: number
  disconnected: number
  degraded: number
  error: number
  demo: number
  live: number
  lastSyncAt: number | null
  throughputPerMin: number
  meanQuality: number
  meanHealth: number
  connectors: ConnectorView[]
}

function toView(row: {
  code: string
  name: string
  nameAr: string
  kind: string
  protocol: string
  mode: string
  status: string
  direction: string
  description: string
  descriptionAr: string
  endpoint: string
  authKind: string
  secretRef: string
  pollSeconds: number
  timeoutMs: number
  lastConnectedAt: Date | null
  lastSyncAt: Date | null
  latencyMs: number
  recordsReceived: number
  errorCount: number
  healthScore: number
  dataQuality: number
  eventsPerMin: number
  faultKind: string
  isActive: boolean
  _count?: { mappings: number }
}): ConnectorView {
  const mode = (row.mode === 'live' ? 'live' : 'demo') as ConnectorMode
  const status = row.status as ConnectorStatus
  return {
    code: row.code,
    name: row.name,
    nameAr: row.nameAr,
    kind: row.kind,
    protocol: row.protocol,
    mode,
    status,
    label: connectorLabel(mode, status),
    direction: row.direction,
    description: row.description,
    descriptionAr: row.descriptionAr,
    endpoint: row.endpoint,
    authKind: row.authKind,
    secretRef: row.secretRef,
    pollSeconds: row.pollSeconds,
    timeoutMs: row.timeoutMs,
    lastConnectedAt: row.lastConnectedAt?.getTime() ?? null,
    lastSyncAt: row.lastSyncAt?.getTime() ?? null,
    latencyMs: row.latencyMs,
    recordsReceived: row.recordsReceived,
    errorCount: row.errorCount,
    healthScore: row.healthScore,
    dataQuality: row.dataQuality,
    eventsPerMin: row.eventsPerMin,
    faultKind: row.faultKind,
    mappingCount: row._count?.mappings ?? 0,
    isActive: row.isActive,
  }
}

/** Never selects `secretRef`'s value because there is no value column — only its name. */
export async function getHubOverview(): Promise<HubOverview> {
  const rows = await prisma.integrationConnector.findMany({
    orderBy: [{ status: 'asc' }, { name: 'asc' }],
    include: { _count: { select: { mappings: true } } },
  })

  const connectors = rows.map(toView)
  const active = connectors.filter((entry) => entry.isActive)
  const syncTimes = connectors
    .map((entry) => entry.lastSyncAt)
    .filter((value): value is number => value !== null)

  const mean = (values: number[]) =>
    values.length ? round(values.reduce((a, b) => a + b, 0) / values.length, 1) : 0

  return {
    total: connectors.length,
    connected: connectors.filter((entry) => entry.status === 'connected').length,
    disconnected: connectors.filter((entry) => entry.status === 'disconnected').length,
    degraded: connectors.filter((entry) => entry.status === 'degraded').length,
    error: connectors.filter((entry) => entry.status === 'error').length,
    demo: connectors.filter((entry) => entry.mode === 'demo').length,
    live: connectors.filter((entry) => entry.mode === 'live').length,
    lastSyncAt: syncTimes.length ? Math.max(...syncTimes) : null,
    throughputPerMin: round(
      active.reduce((sum, entry) => sum + entry.eventsPerMin, 0),
      1,
    ),
    // Averaged over connectors that are actually carrying data. Including a disabled
    // connector's zero would drag the headline down for a source nobody is using.
    meanQuality: mean(active.filter((e) => e.status !== 'disabled').map((e) => e.dataQuality)),
    meanHealth: mean(active.filter((e) => e.status !== 'disabled').map((e) => e.healthScore)),
    connectors,
  }
}

// ─────────────────────────────────────────────────────────────────────────────
// Live readiness (5.0 §8–§10)
// ─────────────────────────────────────────────────────────────────────────────

export interface ReadinessRow {
  code: string
  name: string
  nameAr: string
  kind: string
  mode: ConnectorMode
  report: ReadinessReport
}

export interface ReadinessOverview {
  /** Connectors that may honestly be described as live. Often zero, and that is the point. */
  liveReady: number
  demoVerified: number
  incomplete: number
  /** Adapters the platform ships, so "what can it speak?" is answerable without a sales call. */
  adapters: AdapterSpec[]
  /** Control verbs no adapter implements. Stated as data so a test can hold it. */
  refuses: readonly string[]
  rows: ReadinessRow[]
  note: string
  noteAr: string
}

/**
 * What each configured source would need before it could carry live decisions.
 *
 * The honest answer for this deployment is "none of them are live", and this function is
 * how the platform says so with a reason per connector rather than a disclaimer in a
 * footer. Nothing here is a promise about a future integration — every check reads a
 * field that exists now.
 */
export async function getLiveReadiness(now = Date.now()): Promise<ReadinessOverview> {
  const overview = await getHubOverview()

  const rows: ReadinessRow[] = overview.connectors.map((connector) => ({
    code: connector.code,
    name: connector.name,
    nameAr: connector.nameAr,
    kind: connector.kind,
    mode: connector.mode,
    report: assessLiveReadiness({
      protocol: connector.protocol,
      mode: connector.mode,
      status: connector.status,
      endpoint: connector.endpoint,
      authKind: connector.authKind,
      secretRef: connector.secretRef,
      mappings: connector.mappingCount,
      recordsReceived: connector.recordsReceived,
      lastSyncAt: connector.lastSyncAt,
      latencyMs: connector.latencyMs,
      dataQuality: connector.dataQuality,
      ts: now,
    }),
  }))

  return {
    liveReady: rows.filter((row) => row.report.canClaimLive).length,
    demoVerified: rows.filter((row) => row.report.state === 'demo_verified').length,
    incomplete: rows.filter((row) => row.report.blocking.length > 0).length,
    adapters: ADAPTER_PROTOCOLS.map((protocol) => ADAPTERS[protocol]),
    refuses: FORBIDDEN_VERBS,
    rows,
    note:
      'Readiness, not intent. A connector is described as live only when a real endpoint is configured, reachable and delivering — every other source is labelled by what it actually is.',
    noteAr:
      'جاهزية لا نية. لا يُوصف الموصل بأنه حي إلا مع نقطة اتصال حقيقية تصل وتُسلّم البيانات، وكل مصدر آخر يُوسم بما هو عليه فعلًا.',
  }
}

export interface ConnectorDetail extends ConnectorView {
  mappings: Array<FieldMapping & { notes: string; transform: string }>
  events: Array<{
    kind: string
    recordsIn: number
    recordsAccepted: number
    recordsRejected: number
    latencyMs: number
    detail: string
    detailAr: string
    severity: string
    ts: number
  }>
  samples: Array<{
    ts: number
    healthScore: number
    dataQuality: number
    latencyMs: number
    completeness: number
    freshness: number
    validity: number
    consistency: number
    availability: number
    trust: number
  }>
  pipeline: PipelineStep[]
}

export async function getConnector(code: string): Promise<ConnectorDetail | null> {
  const row = await prisma.integrationConnector.findUnique({
    where: { code },
    include: {
      _count: { select: { mappings: true } },
      mappings: { orderBy: { sourceField: 'asc' } },
      events: { orderBy: { ts: 'desc' }, take: 25 },
      samples: { orderBy: { ts: 'desc' }, take: 40 },
    },
  })
  if (!row) return null

  const latest = row.events.find((event) => event.kind === 'received' || event.kind === 'sync')

  return {
    ...toView(row),
    mappings: row.mappings.map((mapping) => ({
      sourceField: mapping.sourceField,
      targetField: mapping.targetField,
      dataType: mapping.dataType as FieldMapping['dataType'],
      sourceUnit: mapping.sourceUnit,
      targetUnit: mapping.targetUnit,
      defaultValue: mapping.defaultValue,
      minValue: mapping.minValue,
      maxValue: mapping.maxValue,
      isRequired: mapping.isRequired,
      transform: mapping.transform,
      notes: mapping.notes,
    })),
    events: row.events.map((event) => ({
      kind: event.kind,
      recordsIn: event.recordsIn,
      recordsAccepted: event.recordsAccepted,
      recordsRejected: event.recordsRejected,
      latencyMs: event.latencyMs,
      detail: event.detail,
      detailAr: event.detailAr,
      severity: event.severity,
      ts: event.ts.getTime(),
    })),
    samples: [...row.samples].reverse().map((sample) => ({
      ts: sample.ts.getTime(),
      healthScore: sample.healthScore,
      dataQuality: sample.dataQuality,
      latencyMs: sample.latencyMs,
      completeness: sample.completeness,
      freshness: sample.freshness,
      validity: sample.validity,
      consistency: sample.consistency,
      availability: sample.availability,
      trust: sample.trust,
    })),
    pipeline: pipelineSteps({
      received: latest?.recordsIn ?? 0,
      accepted: latest?.recordsAccepted ?? 0,
      rejected: latest?.recordsRejected ?? 0,
      mapped: latest?.recordsAccepted ?? 0,
      quality: row.dataQuality,
    }),
  }
}

// ─────────────────────────────────────────────────────────────────────────────
// Synchronising
// ─────────────────────────────────────────────────────────────────────────────

export interface SyncResult {
  code: string
  mode: ConnectorMode
  status: ConnectorStatus
  label: string
  recordsIn: number
  accepted: number
  rejected: number
  latencyMs: number
  quality: SourceQuality
  health: number
  /** Every distinct rejection reason in the batch, with a count. */
  rejectionReasons: Array<{ reason: string; count: number; detail: string; detailAr: string }>
  pipeline: PipelineStep[]
  preview: Array<{ assetCode: string; ts: number; fields: Array<{ field: string; value: unknown; unit: string; sourceValue: unknown; sourceUnit: string }> }>
  faultKind: string
}

/**
 * Run one pass of the pipeline.
 *
 * For a demo connector the records come from the platform's own simulation engine, which
 * is what makes the exercise honest: the values are the same ones the twin is built on,
 * so a mapping error or a unit mistake shows up as a genuinely wrong reading rather than
 * as a discrepancy between two unrelated fictions.
 *
 * A live connector would fetch here. There is no live connector in this build, and the
 * function says so rather than pretending: `mode: 'live'` with no reachable endpoint is
 * returned as an error status, not as a successful sync.
 */
export async function syncConnector(input: {
  code: string
  now?: number
  batchSize?: number
  persist?: boolean
}): Promise<SyncResult | null> {
  const now = input.now ?? Date.now()
  const batchSize = Math.min(500, Math.max(1, input.batchSize ?? 60))

  const row = await prisma.integrationConnector.findUnique({
    where: { code: input.code },
    include: { mappings: true },
  })
  if (!row) return null

  const mode = (row.mode === 'live' ? 'live' : 'demo') as ConnectorMode

  // A live connector with no endpoint is not a connector that failed to fetch — it is a
  // connector that was never wired up, and saying so is the whole point of §"do not claim
  // an integration exists if it is not implemented".
  if (mode === 'live' && !row.endpoint) {
    const result = await recordFailedSync(row.id, row.code, 'No endpoint is configured for this live connector')
    return result
  }

  const mappings: FieldMapping[] = row.mappings.map((mapping) => ({
    sourceField: mapping.sourceField,
    targetField: mapping.targetField,
    dataType: mapping.dataType as FieldMapping['dataType'],
    sourceUnit: mapping.sourceUnit,
    targetUnit: mapping.targetUnit,
    defaultValue: mapping.defaultValue,
    minValue: mapping.minValue,
    maxValue: mapping.maxValue,
    isRequired: mapping.isRequired,
  }))

  const fault = row.faultKind ? faultEffect(row.faultKind as FaultKind) : null

  // ── Generate the batch from the platform's own model ──────────────────────
  const snapshot = await getGridSnapshot({ ts: now })
  const knownAssets = new Set(snapshot.assets.map((entry) => entry.profile.code))
  const pool = snapshot.assets.slice(0, batchSize)

  const raw: Array<Record<string, unknown>> = []
  for (const [index, entry] of pool.entries()) {
    // A dropped record is one that never arrives, so it is omitted from the batch rather
    // than added and rejected — the difference is exactly what completeness measures.
    if (fault && seededRange(entry.profile.seed, index, 0, 1) < fault.dropRate) continue

    const corrupt = fault ? seededRange(entry.profile.seed, index + 7, 0, 1) < fault.corruptionRate : false

    raw.push(
      buildRecord(row.kind, entry, {
        ts: now - (fault?.extraAgeMin ?? 0) * MIN,
        corrupt,
      }),
    )
  }

  // ── Put it through the pipeline ───────────────────────────────────────────
  const seen = new Set<string>()
  const records = raw.map((record) =>
    normaliseRecord(record, { mappings, knownAssets, seen, now }),
  )

  const accepted = records.filter((record) => record.accepted)
  const rejected = records.filter((record) => !record.accepted)

  const reasonCounts = new Map<string, { count: number; detail: string; detailAr: string }>()
  for (const record of rejected) {
    for (const rejection of record.rejections) {
      const existing = reasonCounts.get(rejection.reason)
      if (existing) existing.count += 1
      else reasonCounts.set(rejection.reason, { count: 1, detail: rejection.detail, detailAr: rejection.detailAr })
    }
  }

  const latencyMs = round(
    (row.latencyMs || 40) * (fault?.latencyFactor ?? 1),
    1,
  )

  const quality = assessSourceQuality({
    expectedRecords: pool.length,
    receivedRecords: records.length,
    acceptedRecords: accepted.length,
    ageMin: (fault?.extraAgeMin ?? 0) + 0.1,
    expectedIntervalMin: Math.max(0.1, row.pollSeconds / 60),
    uptimeRatio: fault?.status === 'disconnected' ? 0 : 1,
    agreementRatio: records.length ? accepted.length / records.length : 0,
    mode,
  })

  const status: ConnectorStatus = fault ? fault.status : accepted.length > 0 ? 'connected' : 'degraded'

  const health = connectorHealth({
    status,
    quality: quality.score,
    latencyMs,
    timeoutMs: row.timeoutMs,
    errorCount: rejected.length,
    recordsReceived: records.length,
  })

  const result: SyncResult = {
    code: row.code,
    mode,
    status,
    label: connectorLabel(mode, status),
    recordsIn: records.length,
    accepted: accepted.length,
    rejected: rejected.length,
    latencyMs,
    quality,
    health,
    rejectionReasons: [...reasonCounts.entries()].map(([reason, value]) => ({
      reason,
      count: value.count,
      detail: value.detail,
      detailAr: value.detailAr,
    })),
    pipeline: pipelineSteps({
      received: records.length,
      accepted: accepted.length,
      rejected: rejected.length,
      mapped: accepted.length,
      quality: quality.score,
    }),
    preview: accepted.slice(0, 5).map((record) => ({
      assetCode: record.assetCode,
      ts: record.ts,
      fields: record.fields.map((field) => ({
        field: field.targetField,
        value: field.value,
        unit: field.unit,
        sourceValue: field.sourceValue,
        sourceUnit: field.sourceUnit,
      })),
    })),
    faultKind: row.faultKind,
  }

  if (input.persist !== false) {
    await prisma.$transaction([
      prisma.integrationConnector.update({
        where: { id: row.id },
        data: {
          status,
          lastSyncAt: new Date(now),
          lastConnectedAt: status === 'connected' ? new Date(now) : row.lastConnectedAt,
          latencyMs,
          recordsReceived: row.recordsReceived + records.length,
          errorCount: row.errorCount + rejected.length,
          healthScore: health,
          dataQuality: quality.score,
          eventsPerMin: round(records.length / Math.max(0.1, row.pollSeconds / 60), 1),
        },
      }),
      prisma.integrationEvent.create({
        data: {
          connectorId: row.id,
          kind: 'sync',
          recordsIn: records.length,
          recordsAccepted: accepted.length,
          recordsRejected: rejected.length,
          latencyMs,
          detail: rejected.length
            ? `${rejected.length} of ${records.length} records rejected`
            : `${accepted.length} records accepted`,
          detailAr: rejected.length
            ? `رُفض ${rejected.length} من ${records.length} سجلًا`
            : `قُبل ${accepted.length} سجلًا`,
          severity: rejected.length > records.length * 0.2 ? 'warning' : 'info',
          ts: new Date(now),
        },
      }),
      prisma.integrationHealthSample.create({
        data: {
          connectorId: row.id,
          healthScore: health,
          dataQuality: quality.score,
          latencyMs,
          completeness: quality.completeness,
          freshness: quality.freshness,
          validity: quality.validity,
          consistency: quality.consistency,
          availability: quality.availability,
          trust: quality.trust,
          ts: new Date(now),
        },
      }),
    ])
  }

  return result
}

async function recordFailedSync(id: string, code: string, reason: string): Promise<SyncResult> {
  await prisma.$transaction([
    prisma.integrationConnector.update({
      where: { id },
      data: { status: 'error', errorCount: { increment: 1 }, healthScore: 0, dataQuality: 0 },
    }),
    prisma.integrationEvent.create({
      data: {
        connectorId: id,
        kind: 'error',
        detail: reason,
        detailAr: 'الموصّل الحي غير مهيأ بنقطة نهاية',
        severity: 'error',
      },
    }),
  ])

  const empty = assessSourceQuality({
    expectedRecords: 0,
    receivedRecords: 0,
    acceptedRecords: 0,
    ageMin: 0,
    expectedIntervalMin: 1,
    uptimeRatio: 0,
    agreementRatio: 0,
    mode: 'live',
  })

  return {
    code,
    mode: 'live',
    status: 'error',
    label: 'error',
    recordsIn: 0,
    accepted: 0,
    rejected: 0,
    latencyMs: 0,
    quality: empty,
    health: 0,
    rejectionReasons: [],
    pipeline: pipelineSteps({ received: 0, accepted: 0, rejected: 0, mapped: 0, quality: 0 }),
    preview: [],
    faultKind: '',
  }
}

/**
 * Build one raw record in the shape the source would send it.
 *
 * Deliberately *not* in the platform's units or field names — a SCADA gateway sends
 * `transformer_temperature` in Fahrenheit and `active_power` in kilowatts, and if the
 * generator emitted canonical values the mapping layer would never be exercised.
 */
function buildRecord(
  kind: string,
  entry: { profile: { code: string; seed: number }; sample: { tempC: number; loadMw: number; loadPct: number; voltageKv: number; currentA: number; ambientC: number } },
  options: { ts: number; corrupt: boolean },
): Record<string, unknown> {
  const { profile, sample } = entry
  const base: Record<string, unknown> = { asset_id: profile.code, timestamp: options.ts }

  switch (kind) {
    case 'weather':
      return {
        ...base,
        ambient_temp_f: options.corrupt ? 'n/a' : round((sample.ambientC * 9) / 5 + 32, 2),
        humidity_pct: round(seededRange(profile.seed, 3, 8, 40), 1),
      }
    case 'cmms':
      return {
        ...base,
        work_order_open: seededRange(profile.seed, 5, 0, 1) > 0.85,
        last_service_days: Math.round(seededRange(profile.seed, 6, 20, 400)),
      }
    case 'iot':
      return {
        ...base,
        transformer_temperature: options.corrupt ? 9_999 : round((sample.tempC * 9) / 5 + 32, 2),
        vibration_mm_s: round(seededRange(profile.seed, 7, 0.2, 3.5), 2),
      }
    default:
      // SCADA and friends: power in kilowatts, temperature in Fahrenheit, voltage in volts.
      return {
        ...base,
        active_power: options.corrupt ? -999_999 : round(sample.loadMw * 1_000, 1),
        transformer_temperature: round((sample.tempC * 9) / 5 + 32, 2),
        bus_voltage: round(sample.voltageKv * 1_000, 1),
        line_current: round(sample.currentA, 1),
        loading_pct: round(Math.abs(sample.loadPct), 2),
      }
  }
}

// ─────────────────────────────────────────────────────────────────────────────
// Testing, faults and mapping
// ─────────────────────────────────────────────────────────────────────────────

export interface TestResult {
  code: string
  reachable: boolean
  mode: ConnectorMode
  latencyMs: number
  message: string
  messageAr: string
}

/**
 * Test a connector without ingesting anything.
 *
 * A demo connector's generator is always reachable and the result says so in those words.
 * A live connector with no endpoint reports unreachable — the one thing this must never do
 * is return success for a connection that does not exist.
 */
export async function testConnector(code: string, now = Date.now()): Promise<TestResult | null> {
  const row = await prisma.integrationConnector.findUnique({ where: { code } })
  if (!row) return null

  const mode = (row.mode === 'live' ? 'live' : 'demo') as ConnectorMode
  const reachable = mode === 'demo' ? true : Boolean(row.endpoint)
  const latencyMs = reachable ? round(row.latencyMs || 40, 1) : 0

  await prisma.integrationEvent.create({
    data: {
      connectorId: row.id,
      kind: reachable ? 'connected' : 'error',
      latencyMs,
      detail: reachable
        ? mode === 'demo'
          ? 'Demo generator responded'
          : 'Endpoint responded'
        : 'No endpoint configured',
      detailAr: reachable
        ? mode === 'demo'
          ? 'استجاب المولّد التجريبي'
          : 'استجابت نقطة النهاية'
        : 'لا توجد نقطة نهاية مهيأة',
      severity: reachable ? 'info' : 'error',
      ts: new Date(now),
    },
  })

  return {
    code,
    reachable,
    mode,
    latencyMs,
    message: reachable
      ? mode === 'demo'
        ? 'Demo generator responded. This connector produces synthetic data and reaches no external system.'
        : 'Endpoint responded.'
      : 'No endpoint is configured, so this connector reaches nothing.',
    messageAr: reachable
      ? mode === 'demo'
        ? 'استجاب المولّد التجريبي. هذا الموصّل ينتج بيانات اصطناعية ولا يصل إلى أي نظام خارجي.'
        : 'استجابت نقطة النهاية.'
      : 'لا توجد نقطة نهاية مهيأة، فهذا الموصّل لا يصل إلى شيء.',
  }
}

/**
 * Inject or clear a fault on a demo connector.
 *
 * Refused on a live connector: deliberately degrading a real feed is not a demonstration,
 * it is an outage.
 */
export async function setFault(input: {
  code: string
  faultKind: FaultKind | ''
  now?: number
}): Promise<{ ok: boolean; reason?: string }> {
  const row = await prisma.integrationConnector.findUnique({ where: { code: input.code } })
  if (!row) return { ok: false, reason: 'not_found' }
  if (row.mode === 'live') return { ok: false, reason: 'live_connector' }

  const now = input.now ?? Date.now()
  const effect = input.faultKind ? faultEffect(input.faultKind) : null

  await prisma.$transaction([
    prisma.integrationConnector.update({
      where: { id: row.id },
      data: {
        faultKind: input.faultKind,
        status: effect ? effect.status : 'connected',
      },
    }),
    prisma.integrationEvent.create({
      data: {
        connectorId: row.id,
        kind: input.faultKind ? 'fault_injected' : 'fault_cleared',
        detail: effect ? effect.label : 'Fault cleared',
        detailAr: effect ? effect.labelAr : 'أُزيل العطل',
        severity: input.faultKind ? 'warning' : 'info',
        ts: new Date(now),
      },
    }),
  ])

  return { ok: true }
}

export async function upsertMapping(input: {
  code: string
  sourceField: string
  targetField: string
  dataType: string
  sourceUnit: string
  targetUnit: string
  defaultValue?: string
  minValue?: number | null
  maxValue?: number | null
  isRequired?: boolean
  notes?: string
}): Promise<boolean> {
  const row = await prisma.integrationConnector.findUnique({
    where: { code: input.code },
    select: { id: true },
  })
  if (!row) return false

  const data = {
    targetField: input.targetField,
    dataType: input.dataType,
    sourceUnit: input.sourceUnit,
    targetUnit: input.targetUnit,
    defaultValue: input.defaultValue ?? '',
    minValue: input.minValue ?? null,
    maxValue: input.maxValue ?? null,
    isRequired: input.isRequired ?? false,
    notes: input.notes ?? '',
  }

  await prisma.integrationMapping.upsert({
    where: { connectorId_sourceField: { connectorId: row.id, sourceField: input.sourceField } },
    create: { connectorId: row.id, sourceField: input.sourceField, ...data },
    update: data,
  })

  return true
}

export async function deleteMapping(code: string, sourceField: string): Promise<boolean> {
  const row = await prisma.integrationConnector.findUnique({
    where: { code },
    select: { id: true },
  })
  if (!row) return false
  const result = await prisma.integrationMapping.deleteMany({
    where: { connectorId: row.id, sourceField },
  })
  return result.count > 0
}

/**
 * The quality figure the rest of the platform sees.
 *
 * Exposed so the twin's fidelity and the uncertainty engine can be told how good the
 * incoming data actually is, rather than assuming it — which is the point of the hub
 * feeding the twin rather than sitting beside it.
 */
export async function ingestQuality(): Promise<{ score: number; sources: number; degraded: number }> {
  const rows = await prisma.integrationConnector.findMany({
    where: { isActive: true, status: { notIn: ['disabled'] } },
    select: { dataQuality: true, status: true },
  })
  if (!rows.length) return { score: 0, sources: 0, degraded: 0 }

  return {
    score: round(rows.reduce((sum, row) => sum + row.dataQuality, 0) / rows.length, 1),
    sources: rows.length,
    degraded: rows.filter((row) => row.status !== 'connected').length,
  }
}
