import type { DomainEvent, EventsPublisher } from "@domain/events"
import { Data, Effect, Fiber, Result, Schedule } from "effect"
import type { Pool, PoolClient } from "pg"

export interface OutboxEventRow {
  id: string
  event_name: string
  aggregate_id: string
  workspace_id: string
  payload: Record<string, unknown>
  published: boolean
  published_at: string | null
  occurred_at: Date
  created_at: Date
}

export interface PollingOutboxConsumerConfig {
  readonly pool: Pool
  readonly pollIntervalMs: number
  readonly batchSize: number
}

export class OutboxConsumerError extends Data.TaggedError("OutboxConsumerError")<{
  readonly cause: unknown
}> {}

export interface OutboxConsumer {
  readonly start: () => Effect.Effect<void, OutboxConsumerError>
  readonly stop: () => Effect.Effect<void>
}

const SELECT_UNPUBLISHED_EVENTS = `
  SELECT
    id,
    event_name,
    aggregate_id,
    workspace_id,
    payload,
    published,
    published_at,
    occurred_at,
    created_at
  FROM latitude.outbox_events
  WHERE published = false
  ORDER BY created_at ASC
  LIMIT $1
  FOR UPDATE SKIP LOCKED
`

const MARK_EVENTS_PUBLISHED = `
  UPDATE latitude.outbox_events
  SET published = true, published_at = NOW()
  WHERE id = ANY($1::text[])
`

const processBatchEffect = <TPublishError>(
  client: PoolClient,
  publisher: EventsPublisher<TPublishError>,
  batchSize: number,
): Effect.Effect<number, unknown, never> =>
  Effect.gen(function* () {
    const result = yield* Effect.tryPromise(() => client.query<OutboxEventRow>(SELECT_UNPUBLISHED_EVENTS, [batchSize]))

    if (result.rows.length === 0) {
      return 0
    }

    const processedIds: string[] = []

    for (const row of result.rows) {
      const event = {
        name: row.event_name,
        organizationId: row.workspace_id,
        payload: row.payload,
      } as DomainEvent

      const publishResult = yield* Effect.result(publisher.publish(event))

      if (Result.isSuccess(publishResult)) {
        processedIds.push(row.id)
      } else {
        yield* Effect.logError(`Failed to publish event ${row.id}: ${publishResult.failure}`)
      }
    }

    if (processedIds.length > 0) {
      yield* Effect.tryPromise(() => client.query(MARK_EVENTS_PUBLISHED, [processedIds]))
    }

    return processedIds.length
  })

const pollEffect = <TPublishError>(
  config: PollingOutboxConsumerConfig,
  publisher: EventsPublisher<TPublishError>,
): Effect.Effect<number, unknown, never> =>
  Effect.gen(function* () {
    const client = yield* Effect.tryPromise(() => config.pool.connect())

    const result = yield* Effect.gen(function* () {
      yield* Effect.tryPromise(() => client.query("BEGIN"))

      const processedCount = yield* processBatchEffect(client, publisher, config.batchSize)

      yield* Effect.tryPromise(() => client.query("COMMIT"))

      if (processedCount > 0) {
        yield* Effect.logInfo(`Processed ${processedCount} events`)
      }

      return processedCount
    }).pipe(
      Effect.tapError(() => Effect.tryPromise(() => client.query("ROLLBACK")).pipe(Effect.ignore)),
      Effect.ensuring(Effect.sync(() => client.release())),
    )

    return result
  })

export const createPollingOutboxConsumer = <TPublishError>(
  config: PollingOutboxConsumerConfig,
  publisher: EventsPublisher<TPublishError>,
): Effect.Effect<OutboxConsumer, OutboxConsumerError> =>
  Effect.gen(function* () {
    let fiber: Fiber.Fiber<void, unknown> | null = null

    const schedule = Schedule.spaced(config.pollIntervalMs)

    const start = (): Effect.Effect<void, OutboxConsumerError> =>
      Effect.gen(function* () {
        if (fiber !== null) return

        const pollingEffect = Effect.repeat(pollEffect(config, publisher), schedule).pipe(Effect.asVoid)

        fiber = yield* Effect.forkDetach(pollingEffect)

        yield* Effect.logInfo("Polling outbox consumer started")
      }).pipe(Effect.catchCause((cause) => Effect.fail(new OutboxConsumerError({ cause }))))

    const stop = (): Effect.Effect<void> =>
      Effect.gen(function* () {
        if (fiber === null) return

        yield* Fiber.interrupt(fiber)
        fiber = null
        yield* Effect.logInfo("Polling outbox consumer stopped")
      }).pipe(Effect.tapError(Effect.logError), Effect.ignore)

    return { start, stop }
  })
