import {
  checkIssueEscalationUseCase,
  type DiscoverIssueResult,
  discoverIssueUseCase,
  refreshIssueDetailsUseCase,
  removeScoreFromIssueUseCase,
  sweepEscalatingIssuesUseCase,
} from "@domain/issues"
import {
  type QueueConsumer,
  QueuePublisher,
  type QueuePublisherShape,
  WorkflowStarter,
  type WorkflowStarterShape,
} from "@domain/queue"
import type { ScoreSource } from "@domain/scores"
import { OrganizationId } from "@domain/shared"
import { withAi } from "@platform/ai"
import { AIGenerateLive } from "@platform/ai-vercel"
import { AIEmbedLive } from "@platform/ai-voyage"
import type { RedisClient } from "@platform/cache-redis"
import type { ClickHouseClient } from "@platform/db-clickhouse"
import { ScoreAnalyticsRepositoryLive, withClickHouse } from "@platform/db-clickhouse"
import {
  AlertIncidentRepositoryLive,
  EvaluationRepositoryLive,
  IssueRepositoryLive,
  OutboxEventWriterLive,
  type PostgresClient,
  ScoreRepositoryLive,
  SettingsReaderLive,
  withPostgres,
} from "@platform/db-postgres"
import { createLogger, withTracing } from "@repo/observability"
import { Effect, Layer } from "effect"
import { getAdminPostgresClient, getClickhouseClient, getPostgresClient, getRedisClient } from "../clients.ts"

const logger = createLogger("issues")

const formatDiscoveryOutcome = (outcome: DiscoverIssueResult) => {
  switch (outcome.action) {
    case "workflow-started":
      return `${outcome.action}:${outcome.workflow}`
    case "skipped":
      return `${outcome.action}:${outcome.reason}`
    case "already-assigned":
      return `${outcome.action}:${outcome.issueId}`
  }
}

interface IssuesDeps {
  consumer: QueueConsumer
  publisher: QueuePublisherShape
  workflowStarter: WorkflowStarterShape
  postgresClient?: PostgresClient
  /**
   * Admin Postgres client — used by the `sweepEscalating` handler to read
   * `alert_incidents` across orgs (RLS bypass). All other handlers stay on
   * the org-scoped `postgresClient`.
   */
  adminPostgresClient?: PostgresClient
  clickhouseClient?: ClickHouseClient
  redisClient?: RedisClient
}

export const createIssuesWorker = async ({
  consumer,
  publisher,
  workflowStarter,
  postgresClient,
  adminPostgresClient,
  clickhouseClient,
  redisClient,
}: IssuesDeps) => {
  const pgClient = postgresClient ?? getPostgresClient()
  const adminPgClient = adminPostgresClient ?? getAdminPostgresClient()
  const chClient = clickhouseClient ?? getClickhouseClient()
  const rdClient = redisClient ?? getRedisClient()

  consumer.subscribe("issues", {
    discovery: (payload) =>
      discoverIssueUseCase(payload).pipe(
        Effect.tap((outcome) =>
          Effect.sync(() => {
            logger.info(
              `Processed issue discovery for ${payload.projectId}/${payload.scoreId} (${formatDiscoveryOutcome(outcome)})`,
            )
          }),
        ),
        Effect.tapError((error) =>
          Effect.sync(() => logger.error(`Issue discovery failed for ${payload.projectId}/${payload.scoreId}`, error)),
        ),
        withPostgres(
          Layer.mergeAll(EvaluationRepositoryLive, IssueRepositoryLive, OutboxEventWriterLive, ScoreRepositoryLive),
          pgClient,
          OrganizationId(payload.organizationId),
        ),
        withClickHouse(ScoreAnalyticsRepositoryLive, chClient, OrganizationId(payload.organizationId)),
        withAi(AIEmbedLive, rdClient),
        withTracing,
        Effect.provide(Layer.succeed(WorkflowStarter, workflowStarter)),
        Effect.asVoid,
      ),
    refresh: (payload) =>
      refreshIssueDetailsUseCase(payload).pipe(
        withPostgres(
          Layer.mergeAll(EvaluationRepositoryLive, IssueRepositoryLive, ScoreRepositoryLive),
          pgClient,
          OrganizationId(payload.organizationId),
        ),
        withAi(AIGenerateLive, rdClient),
        withTracing,
        Effect.provide(Layer.succeed(QueuePublisher, publisher)),
        Effect.tap(() =>
          Effect.sync(() => logger.info(`Refreshed issue details for ${payload.projectId}/${payload.issueId}`)),
        ),
        Effect.tapError((error) =>
          Effect.sync(() => logger.error(`Issue refresh failed for ${payload.projectId}/${payload.issueId}`, error)),
        ),
        Effect.asVoid,
      ),
    // Evaluate escalation state from the analytics aggregate + the current
    // `alert_incidents`-derived `lifecycle.isEscalating` flag, and emit the
    // matching transition event. The use case does not write the issue —
    // the open/closed `alert_incidents` row is the stored truth. The
    // alert-incidents worker inserts/closes that row in response to
    // `IssueEscalated` / `IssueEscalationEnded`. Triggered by two paths:
    // the throttled `issues:check-escalation` publish from
    // `ScoreAssignedToIssue` (entry + active-burst exit detection), and the
    // hourly `sweepEscalating` cron below (exit detection on quiet issues
    // plus cold-start recovery for already-stuck rows).
    checkEscalation: (payload) =>
      checkIssueEscalationUseCase(payload).pipe(
        withPostgres(
          Layer.mergeAll(IssueRepositoryLive, OutboxEventWriterLive, AlertIncidentRepositoryLive, SettingsReaderLive),
          pgClient,
          OrganizationId(payload.organizationId),
        ),
        withClickHouse(ScoreAnalyticsRepositoryLive, chClient, OrganizationId(payload.organizationId)),
        Effect.tap((result) =>
          Effect.sync(() =>
            logger.info(
              `Escalation check for ${payload.projectId}/${payload.issueId}: transition=${result.transition} currentlyEscalating=${result.currentlyEscalating}`,
            ),
          ),
        ),
        Effect.tapError((error) =>
          Effect.sync(() => logger.error(`Escalation check failed for ${payload.projectId}/${payload.issueId}`, error)),
        ),
        withTracing,
        Effect.asVoid,
      ),
    // Fired by the hourly cron. Reads every open `issue.escalating` row
    // (across orgs, via admin Postgres) and enqueues one `checkEscalation`
    // per incident. Separate dedupeKey from the throttled publish so a
    // pending throttled job's jobId doesn't shadow the sweep publish — the
    // BullMQ dedup uses dedupeKey as the jobId.
    sweepEscalating: () =>
      sweepEscalatingIssuesUseCase({
        publish: (payload) =>
          publisher.publish("issues", "checkEscalation", payload, {
            dedupeKey: `issues:check-escalation-sweep:${payload.issueId}`,
          }),
      }).pipe(
        withPostgres(AlertIncidentRepositoryLive, adminPgClient),
        Effect.tap((result) =>
          Effect.sync(() =>
            logger.info(
              `Escalation sweep: published=${result.published} failed=${result.failed} attempted=${result.attempted}`,
            ),
          ),
        ),
        Effect.tapError((error) => Effect.sync(() => logger.error("Escalation sweep failed", error))),
        withTracing,
        Effect.asVoid,
      ),
    removeScore: (payload) =>
      removeScoreFromIssueUseCase({
        organizationId: payload.organizationId,
        projectId: payload.projectId,
        issueId: payload.issueId,
        draftedAt: payload.draftedAt ? new Date(payload.draftedAt) : null,
        feedback: payload.feedback,
        source: payload.source as ScoreSource,
        createdAt: new Date(payload.createdAt),
      }).pipe(
        withPostgres(IssueRepositoryLive, pgClient, OrganizationId(payload.organizationId)),
        withAi(AIEmbedLive, rdClient),
        withTracing,
        Effect.tap((result) =>
          Effect.sync(() => {
            if (result.action === "removed") {
              logger.info(`Removed score contribution from issue centroid for ${payload.projectId}/${payload.issueId}`)
            } else if (result.action === "issue-not-found") {
              logger.info(`Issue ${payload.issueId} not found when removing score contribution`)
            }
          }),
        ),
        Effect.tapError((error) =>
          Effect.sync(() =>
            logger.error(`Failed to remove score from issue ${payload.projectId}/${payload.issueId}`, error),
          ),
        ),
        Effect.asVoid,
      ),
  })
}
