lightspeed

//// Connector and retry/dead-letter contracts for M32 orchestration parity.

import gleam/int
import gleam/option.{type Option, None, Some}

/// Source-side connector abstraction.
pub type SourceConnector {
  DatabaseSource(name: String, query: String, batch_size: Int)
  FileSource(name: String, path: String, format: String, batch_size: Int)
  QueueSource(name: String, topic: String, prefetch: Int)
}

/// Sink-side connector abstraction.
pub type SinkConnector {
  DatabaseSink(name: String, table: String, upsert_key: String)
  PubSubSink(name: String, topic: String, ordering_key: String)
}

/// Retry policy for connector failures.
pub type RetryPolicy {
  RetryPolicy(max_attempts: Int, base_backoff_ms: Int, max_backoff_ms: Int)
}

/// Dead-letter policy after retry exhaustion.
pub type DeadLetterPolicy {
  DeadLetterPolicy(enabled: Bool, destination: String, max_payload_bytes: Int)
}

/// Optional reprocess-window contract.
pub type ReprocessWindow {
  ReprocessWindow(start_sequence: Int, end_sequence: Int, reason: String)
}

/// Full connector plan contract.
pub type ConnectorPlan {
  ConnectorPlan(
    name: String,
    source: SourceConnector,
    sink: SinkConnector,
    retry_policy: RetryPolicy,
    dead_letter_policy: DeadLetterPolicy,
    reprocess_window: Option(ReprocessWindow),
    max_batch_records: Int,
  )
}

/// Deterministic failure action.
pub type FailureAction {
  Retry(next_attempt: Int, at_ms: Int)
  DeadLetter(destination: String, reason: String)
  Reject(reason: String)
}

/// Build a database source connector.
pub fn database_source(
  name: String,
  query: String,
  batch_size: Int,
) -> SourceConnector {
  DatabaseSource(name: name, query: query, batch_size: batch_size)
}

/// Build a file source connector.
pub fn file_source(
  name: String,
  path: String,
  format: String,
  batch_size: Int,
) -> SourceConnector {
  FileSource(name: name, path: path, format: format, batch_size: batch_size)
}

/// Build a queue source connector.
pub fn queue_source(
  name: String,
  topic: String,
  prefetch: Int,
) -> SourceConnector {
  QueueSource(name: name, topic: topic, prefetch: prefetch)
}

/// Build a database sink connector.
pub fn database_sink(
  name: String,
  table: String,
  upsert_key: String,
) -> SinkConnector {
  DatabaseSink(name: name, table: table, upsert_key: upsert_key)
}

/// Build a pubsub sink connector.
pub fn pubsub_sink(
  name: String,
  topic: String,
  ordering_key: String,
) -> SinkConnector {
  PubSubSink(name: name, topic: topic, ordering_key: ordering_key)
}

/// Build a connector plan.
pub fn connector_plan(
  name: String,
  source: SourceConnector,
  sink: SinkConnector,
  retry_policy: RetryPolicy,
  dead_letter_policy: DeadLetterPolicy,
  reprocess_window: Option(ReprocessWindow),
  max_batch_records: Int,
) -> ConnectorPlan {
  ConnectorPlan(
    name: name,
    source: source,
    sink: sink,
    retry_policy: retry_policy,
    dead_letter_policy: dead_letter_policy,
    reprocess_window: reprocess_window,
    max_batch_records: max_batch_records,
  )
}

/// Default M32 connector plan.
pub fn default_plan() -> ConnectorPlan {
  ConnectorPlan(
    name: "orders_connector_plan",
    source: QueueSource(name: "orders_queue", topic: "orders.raw", prefetch: 4),
    sink: DatabaseSink(
      name: "orders_store",
      table: "orders_projection",
      upsert_key: "order_id",
    ),
    retry_policy: RetryPolicy(
      max_attempts: 3,
      base_backoff_ms: 250,
      max_backoff_ms: 2000,
    ),
    dead_letter_policy: DeadLetterPolicy(
      enabled: True,
      destination: "orders.dead_letter",
      max_payload_bytes: 65_536,
    ),
    reprocess_window: None,
    max_batch_records: 500,
  )
}

/// Validate connector plan invariants.
pub fn valid(plan: ConnectorPlan) -> Bool {
  plan.name != ""
  && source_valid(plan.source)
  && sink_valid(plan.sink)
  && retry_valid(plan.retry_policy)
  && dead_letter_valid(plan.dead_letter_policy)
  && reprocess_window_valid(plan.reprocess_window)
  && plan.max_batch_records > 0
}

/// Determine failure action from retry and dead-letter policy.
pub fn classify_failure(
  plan: ConnectorPlan,
  attempt: Int,
  reason: String,
  now_ms: Int,
) -> FailureAction {
  case attempt < plan.retry_policy.max_attempts {
    True -> {
      let next_attempt = attempt + 1
      Retry(
        next_attempt: next_attempt,
        at_ms: max(0, now_ms) + backoff_ms(plan.retry_policy, next_attempt),
      )
    }
    False ->
      case plan.dead_letter_policy.enabled {
        True ->
          DeadLetter(
            destination: plan.dead_letter_policy.destination,
            reason: reason,
          )
        False -> Reject(reason: "retry_exhausted:" <> reason)
      }
  }
}

/// Check whether one sequence is allowed by the optional reprocess window.
pub fn within_reprocess_window(plan: ConnectorPlan, sequence: Int) -> Bool {
  case plan.reprocess_window {
    None -> sequence > 0
    Some(window) ->
      sequence >= window.start_sequence && sequence <= window.end_sequence
  }
}

/// Source connector label.
pub fn source_label(source: SourceConnector) -> String {
  case source {
    DatabaseSource(name, query, batch_size) ->
      "database_source:"
      <> name
      <> ":"
      <> query
      <> ":"
      <> int.to_string(batch_size)
    FileSource(name, path, format, batch_size) ->
      "file_source:"
      <> name
      <> ":"
      <> path
      <> ":"
      <> format
      <> ":"
      <> int.to_string(batch_size)
    QueueSource(name, topic, prefetch) ->
      "queue_source:" <> name <> ":" <> topic <> ":" <> int.to_string(prefetch)
  }
}

/// Sink connector label.
pub fn sink_label(sink: SinkConnector) -> String {
  case sink {
    DatabaseSink(name, table, upsert_key) ->
      "database_sink:" <> name <> ":" <> table <> ":" <> upsert_key
    PubSubSink(name, topic, ordering_key) ->
      "pubsub_sink:" <> name <> ":" <> topic <> ":" <> ordering_key
  }
}

/// Retry policy label.
pub fn retry_policy_label(policy: RetryPolicy) -> String {
  "max_attempts="
  <> int.to_string(policy.max_attempts)
  <> "|base_backoff_ms="
  <> int.to_string(policy.base_backoff_ms)
  <> "|max_backoff_ms="
  <> int.to_string(policy.max_backoff_ms)
}

/// Dead-letter policy label.
pub fn dead_letter_policy_label(policy: DeadLetterPolicy) -> String {
  "enabled="
  <> bool_label(policy.enabled)
  <> "|destination="
  <> policy.destination
  <> "|max_payload_bytes="
  <> int.to_string(policy.max_payload_bytes)
}

/// Failure action label.
pub fn failure_action_label(action: FailureAction) -> String {
  case action {
    Retry(next_attempt, at_ms) ->
      "retry:" <> int.to_string(next_attempt) <> ":" <> int.to_string(at_ms)
    DeadLetter(destination, reason) ->
      "dead_letter:" <> destination <> ":" <> reason
    Reject(reason) -> "reject:" <> reason
  }
}

/// Stable connector-plan signature.
pub fn signature(plan: ConnectorPlan) -> String {
  "name="
  <> plan.name
  <> "|source="
  <> source_label(plan.source)
  <> "|sink="
  <> sink_label(plan.sink)
  <> "|retry="
  <> retry_policy_label(plan.retry_policy)
  <> "|dead_letter="
  <> dead_letter_policy_label(plan.dead_letter_policy)
  <> "|reprocess="
  <> reprocess_window_label(plan.reprocess_window)
  <> "|max_batch_records="
  <> int.to_string(plan.max_batch_records)
}

/// Access source connector.
pub fn source(plan: ConnectorPlan) -> SourceConnector {
  plan.source
}

/// Access sink connector.
pub fn sink(plan: ConnectorPlan) -> SinkConnector {
  plan.sink
}

/// Access retry policy.
pub fn retry_policy(plan: ConnectorPlan) -> RetryPolicy {
  plan.retry_policy
}

/// Access dead-letter policy.
pub fn dead_letter_policy(plan: ConnectorPlan) -> DeadLetterPolicy {
  plan.dead_letter_policy
}

/// Access max batch records.
pub fn max_batch_records(plan: ConnectorPlan) -> Int {
  plan.max_batch_records
}

fn source_valid(source: SourceConnector) -> Bool {
  case source {
    DatabaseSource(name, query, batch_size) ->
      name != "" && query != "" && batch_size > 0
    FileSource(name, path, format, batch_size) ->
      name != "" && path != "" && format != "" && batch_size > 0
    QueueSource(name, topic, prefetch) ->
      name != "" && topic != "" && prefetch > 0
  }
}

fn sink_valid(sink: SinkConnector) -> Bool {
  case sink {
    DatabaseSink(name, table, upsert_key) ->
      name != "" && table != "" && upsert_key != ""
    PubSubSink(name, topic, ordering_key) ->
      name != "" && topic != "" && ordering_key != ""
  }
}

fn retry_valid(policy: RetryPolicy) -> Bool {
  policy.max_attempts > 0
  && policy.base_backoff_ms > 0
  && policy.max_backoff_ms >= policy.base_backoff_ms
}

fn dead_letter_valid(policy: DeadLetterPolicy) -> Bool {
  case policy.enabled {
    True -> policy.destination != "" && policy.max_payload_bytes > 0
    False -> True
  }
}

fn reprocess_window_valid(window: Option(ReprocessWindow)) -> Bool {
  case window {
    None -> True
    Some(window) ->
      window.start_sequence > 0
      && window.end_sequence >= window.start_sequence
      && window.reason != ""
  }
}

fn backoff_ms(policy: RetryPolicy, attempt: Int) -> Int {
  min(policy.max_backoff_ms, policy.base_backoff_ms * max(1, attempt))
}

fn reprocess_window_label(window: Option(ReprocessWindow)) -> String {
  case window {
    None -> "none"
    Some(window) ->
      int.to_string(window.start_sequence)
      <> "-"
      <> int.to_string(window.end_sequence)
      <> ":"
      <> window.reason
  }
}

fn bool_label(value: Bool) -> String {
  case value {
    True -> "true"
    False -> "false"
  }
}

fn max(left: Int, right: Int) -> Int {
  case left >= right {
    True -> left
    False -> right
  }
}

fn min(left: Int, right: Int) -> Int {
  case left <= right {
    True -> left
    False -> right
  }
}